Christoph Haas
3550a4e2e2
A single failing package used to kill its whole worker thread silently because only Vectorizer::Error was rescued per package. Now every StandardError is caught and logged with class and backtrace, and the embedding write is narrowed to update_column(:embedding) so the task only ever issues a minimal UPDATE - it cannot touch any other column or fire model callbacks. Progress stats no longer serialize database writes through the mutex.
90 lines
3.2 KiB
Ruby
90 lines
3.2 KiB
Ruby
# frozen_string_literal: true
|
|
|
|
# Long rake task bodies are normal; the block-length cop is meant for
|
|
# application code.
|
|
# rubocop:disable Metrics/BlockLength
|
|
namespace :debshots do
|
|
desc 'Compute embedding vectors for all packages based on their description ' \
|
|
'and long description (missing ones only unless FORCE=1). ' \
|
|
'Set CONCURRENCY=n to control parallel HTTP requests.'
|
|
task compute_vectors: :environment do
|
|
logger = Logger.new($stdout)
|
|
logger.level = Logger::INFO
|
|
|
|
force = ENV['FORCE'].present?
|
|
concurrency = [ENV.fetch('CONCURRENCY', '4').to_i, 1].max
|
|
|
|
# Never use more threads than the connection pool has connections.
|
|
# Keep one connection reserved for the main thread.
|
|
pool_size = ActiveRecord::Base.connection_pool.size
|
|
concurrency = [concurrency, pool_size - 1].min
|
|
concurrency = 0 if concurrency.negative?
|
|
|
|
scope = force ? Package.all : Package.where(embedding: nil)
|
|
total = scope.count
|
|
|
|
if total.zero?
|
|
logger.info 'No packages need vector computation.'
|
|
next
|
|
end
|
|
|
|
logger.info "Computing vectors for #{total} packages " \
|
|
"(#{force ? 'recomputing' : 'missing only'}, " \
|
|
"service: #{Vectorizer.service_url}, " \
|
|
"concurrency: #{concurrency.zero? ? 1 : concurrency})"
|
|
|
|
stats_mutex = Mutex.new
|
|
stats = { done: 0, failed: 0 }
|
|
|
|
process_package = lambda do |package|
|
|
begin
|
|
raise ArgumentError, 'no description text' if package.embedding_text.blank?
|
|
|
|
package.update_embedding!
|
|
stats_mutex.synchronize { stats[:done] += 1 }
|
|
rescue StandardError => e
|
|
# A single broken package must never take down a whole worker
|
|
# thread - log enough detail to diagnose it and move on.
|
|
stats_mutex.synchronize { stats[:failed] += 1 }
|
|
logger.error "Failed for package #{package.name}: #{e.class}: #{e.message}"
|
|
logger.error e.backtrace.first(3).join("\n") if e.backtrace
|
|
end
|
|
|
|
stats_mutex.synchronize do
|
|
processed = stats[:done] + stats[:failed]
|
|
return unless (processed % 100).zero?
|
|
|
|
percent = (processed.to_f / total * 100).round(1)
|
|
logger.info "Progress: #{processed}/#{total} (#{percent}%)"
|
|
end
|
|
end
|
|
|
|
if concurrency.zero?
|
|
# Sequential fallback when the connection pool has no room for workers
|
|
scope.find_each { |package| process_package.call(package) }
|
|
else
|
|
# Feed package IDs into a queue that the worker threads pull from.
|
|
work_queue = Queue.new
|
|
scope.pluck(:id).each { |id| work_queue << id }
|
|
concurrency.times { work_queue << nil } # one poison pill per worker
|
|
|
|
workers = Array.new(concurrency) do
|
|
Thread.new do
|
|
# Each thread gets its own database connection from the pool.
|
|
ActiveRecord::Base.connection_pool.with_connection do
|
|
while (package_id = work_queue.pop)
|
|
package = Package.find_by(id: package_id)
|
|
process_package.call(package) if package
|
|
end
|
|
end
|
|
rescue StandardError => e
|
|
logger.error "Worker crashed: #{e.message}"
|
|
end
|
|
end
|
|
|
|
workers.each(&:join)
|
|
end
|
|
|
|
logger.info "Done. #{stats[:done]} vectors computed, #{stats[:failed]} failures."
|
|
end
|
|
end
|