From 2b79470feb25f22e04fad50345fa6888ee053d19 Mon Sep 17 00:00:00 2001
From: Christoph Haas <email@christoph-haas.de>
Date: Sun, 23 Aug 2026 00:13:50 +0200
Subject: [PATCH] Simplify vectorization task to sequential processing

Drop the threaded worker pool (queue, mutexes, connection pooling).
The task now processes packages one by one via find_each - plenty
fast since the external embedding service dominates latency, and
much easier to reason about.
---
 lib/tasks/vectorize_packages.rake | 67 +++++++------------------------
 1 file changed, 15 insertions(+), 52 deletions(-)

diff --git a/lib/tasks/vectorize_packages.rake b/lib/tasks/vectorize_packages.rake
index ce873d8..4fbfe7b 100644
--- a/lib/tasks/vectorize_packages.rake
+++ b/lib/tasks/vectorize_packages.rake
@@ -5,21 +5,12 @@
 # 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.'
+       'and long description (missing ones only unless FORCE=1)'
   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
 
@@ -30,61 +21,33 @@ namespace :debshots do
 
     logger.info "Computing vectors for #{total} packages " \
                 "(#{force ? 'recomputing' : 'missing only'}, " \
-                "service: #{Vectorizer.service_url}, " \
-                "concurrency: #{concurrency.zero? ? 1 : concurrency})"
+                "service: #{Vectorizer.service_url})"
 
-    stats_mutex = Mutex.new
-    stats = { done: 0, failed: 0 }
+    done = 0
+    failed = 0
 
-    process_package = lambda do |package|
+    scope.find_each do |package|
       begin
         raise ArgumentError, 'no description text' if package.embedding_text.blank?
 
         package.update_embedding!
-        stats_mutex.synchronize { stats[:done] += 1 }
+        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 }
+        # A single broken package must never stop the whole run - log
+        # enough detail to diagnose it and move on.
+        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?
+      processed = done + failed
+      next unless (processed % 100).zero? || processed == total
 
-        percent = (processed.to_f / total * 100).round(1)
-        logger.info "Progress: #{processed}/#{total} (#{percent}%)"
-      end
+      percent = (processed.to_f / total * 100).round(1)
+      logger.info "Progress: #{processed}/#{total} (#{percent}%)"
     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."
+    logger.info "Done. #{done} vectors computed, #{failed} failures."
   end
 end
+# rubocop:enable Metrics/BlockLength