Pass process.name all the way to claimed executions when claiming jobs

Author: rosa

app/models/solid_queue/claimed_execution.rb

@@ -1,7 +1,7 @@
 # frozen_string_literal: true
 
 class SolidQueue::ClaimedExecution < SolidQueue::Execution
-  belongs_to :process
+  belongs_to :process, primary_key: :name, foreign_key: :process_name
 
   scope :orphaned, -> { where.missing(:process) }
 
@@ -12,12 +12,13 @@ def success?
   end
 
   class << self
-    def claiming(job_ids, process_id, &block)
-      job_data = Array(job_ids).collect { |job_id| { job_id: job_id, process_id: process_id } }
+    def claiming(job_ids, process, &block)
+      process_data = { process_id: process.id, process_name: process.name }
+      job_data = Array(job_ids).collect { |job_id| { job_id: job_id }.merge(process_data) }
 
-      SolidQueue.instrument(:claim, process_id: process_id, job_ids: job_ids) do |payload|
+      SolidQueue.instrument(:claim, job_ids: job_ids, **process_data) do |payload|
         insert_all!(job_data)
-        where(job_id: job_ids, process_id: process_id).load.tap do |claimed|
+        where(job_id: job_ids, process_id: process.id).load.tap do |claimed|
           block.call(claimed)
 
           payload[:size] = claimed.size
@@ -45,6 +46,7 @@ def fail_all_with(error)
           end
 
           payload[:process_ids] = executions.map(&:process_id).uniq
+          payload[:process_names] = executions.map(&:process_name).uniq
           payload[:job_ids] = executions.map(&:job_id).uniq
           payload[:size] = executions.size
         end
@@ -74,7 +76,7 @@ def perform
   end
 
   def release
-    SolidQueue.instrument(:release_claimed, job_id: job.id, process_id: process_id) do
+    SolidQueue.instrument(:release_claimed, job_id: job.id, process_id: process_id, process_name: process_name) do
       transaction do
         job.dispatch_bypassing_concurrency_limits
         destroy!

app/models/solid_queue/process/executor.rb

@@ -6,7 +6,7 @@ module Executor
       extend ActiveSupport::Concern
 
       included do
-        has_many :claimed_executions
+        has_many :claimed_executions, primary_key: :name, foreign_key: :process_name
 
         after_destroy :release_all_claimed_executions
       end

app/models/solid_queue/ready_execution.rb

@@ -7,9 +7,9 @@ class ReadyExecution < Execution
     assumes_attributes_from_job
 
     class << self
-      def claim(queue_list, limit, process_id)
+      def claim(queue_list, limit, process)
         QueueSelector.new(queue_list, self).scoped_relations.flat_map do |queue_relation|
-          select_and_lock(queue_relation, process_id, limit).tap do |locked|
+          select_and_lock(queue_relation, process, limit).tap do |locked|
             limit -= locked.size
           end
         end
@@ -20,23 +20,23 @@ def aggregated_count_across(queue_list)
       end
 
       private
-        def select_and_lock(queue_relation, process_id, limit)
+        def select_and_lock(queue_relation, process, limit)
           return [] if limit <= 0
 
           transaction do
             candidates = select_candidates(queue_relation, limit)
-            lock_candidates(candidates, process_id)
+            lock_candidates(candidates, process)
           end
         end
 
         def select_candidates(queue_relation, limit)
           queue_relation.ordered.limit(limit).non_blocking_lock.select(:id, :job_id)
         end
 
-        def lock_candidates(executions, process_id)
+        def lock_candidates(executions, process)
           return [] if executions.none?
 
-          SolidQueue::ClaimedExecution.claiming(executions.map(&:job_id), process_id) do |claimed|
+          SolidQueue::ClaimedExecution.claiming(executions.map(&:job_id), process) do |claimed|
             ids_to_delete = executions.index_by(&:job_id).values_at(*claimed.map(&:job_id)).map(&:id)
             where(id: ids_to_delete).delete_all
           end

lib/solid_queue/worker.rb

@@ -38,7 +38,7 @@ def poll
 
       def claim_executions
         with_polling_volume do
-          SolidQueue::ReadyExecution.claim(queues, pool.idle_threads, process_id)
+          SolidQueue::ReadyExecution.claim(queues, pool.idle_threads, process)
         end
       end
 

test/integration/instrumentation_test.rb

@@ -150,7 +150,7 @@ class InstrumentationTest < ActiveSupport::TestCase
     3.times { |i| StoreResultJob.set(queue: :new_queue).perform_later(i) }
     jobs = SolidQueue::Job.last(3)
 
-    SolidQueue::ReadyExecution.claim("*", 5, process.id)
+    SolidQueue::ReadyExecution.claim("*", 5, process)
 
     events = subscribed(/fail.*_claimed\.solid_queue/) do
       SolidQueue::Process.prune

test/models/solid_queue/claimed_execution_test.rb

@@ -87,7 +87,7 @@ def prepare_and_claim_job(active_job, process: @process)
 
       job.prepare_for_execution
       assert_difference -> { SolidQueue::ClaimedExecution.count } => +1 do
-        SolidQueue::ReadyExecution.claim(job.queue_name, 1, process.id)
+        SolidQueue::ReadyExecution.claim(job.queue_name, 1, process)
       end
 
       SolidQueue::ClaimedExecution.last

test/models/solid_queue/process_test.rb

@@ -20,7 +20,7 @@ class SolidQueue::ProcessTest < ActiveSupport::TestCase
     3.times { |i| StoreResultJob.set(queue: :new_queue).perform_later(i) }
     jobs = SolidQueue::Job.last(3)
 
-    SolidQueue::ReadyExecution.claim("*", 5, process.id)
+    SolidQueue::ReadyExecution.claim("*", 5, process)
 
     travel_to 10.minutes.from_now
 
@@ -40,7 +40,7 @@ class SolidQueue::ProcessTest < ActiveSupport::TestCase
     3.times { |i| StoreResultJob.set(queue: :new_queue).perform_later(i) }
     jobs = SolidQueue::Job.last(3)
 
-    SolidQueue::ReadyExecution.claim("*", 5, process.id)
+    SolidQueue::ReadyExecution.claim("*", 5, process)
 
     travel_to 10.minutes.from_now
 

test/models/solid_queue/ready_execution_test.rb

@@ -6,12 +6,13 @@ class SolidQueue::ReadyExecutionTest < ActiveSupport::TestCase
       AddToBufferJob.set(queue: "backend", priority: 5 - i).perform_later(i)
     end
 
+    @process = SolidQueue::Process.register(kind: "Worker", pid: 42, name: "worker-123", metadata: { queue: "background" })
     @jobs = SolidQueue::Job.where(queue_name: "backend").order(:priority)
   end
 
   test "claim all jobs for existing queue" do
     assert_claimed_jobs(@jobs.count) do
-      SolidQueue::ReadyExecution.claim("backend", @jobs.count + 1, 42)
+      SolidQueue::ReadyExecution.claim("backend", @jobs.count + 1, @process)
     end
 
     @jobs.each(&:reload)
@@ -21,13 +22,13 @@ class SolidQueue::ReadyExecutionTest < ActiveSupport::TestCase
 
   test "claim jobs for queue without jobs at the moment" do
     assert_no_difference [ -> { SolidQueue::ReadyExecution.count }, -> { SolidQueue::ClaimedExecution.count } ] do
-      SolidQueue::ReadyExecution.claim("some_non_existing_queue", 10, 42)
+      SolidQueue::ReadyExecution.claim("some_non_existing_queue", 10, @process)
     end
   end
 
   test "claim some jobs for existing queue" do
     assert_claimed_jobs(2) do
-      SolidQueue::ReadyExecution.claim("backend", 2, 42)
+      SolidQueue::ReadyExecution.claim("backend", 2, @process)
     end
 
     @jobs.first(2).each do |job|
@@ -45,23 +46,23 @@ class SolidQueue::ReadyExecutionTest < ActiveSupport::TestCase
     AddToBufferJob.perform_later("hey")
 
     assert_claimed_jobs(6) do
-      SolidQueue::ReadyExecution.claim(%w[ backend background ], SolidQueue::Job.count + 1, 42)
+      SolidQueue::ReadyExecution.claim(%w[ backend background ], SolidQueue::Job.count + 1, @process)
     end
   end
 
   test "claim jobs using a wildcard" do
     AddToBufferJob.perform_later("hey")
 
     assert_claimed_jobs(6) do
-      SolidQueue::ReadyExecution.claim("*", SolidQueue::Job.count + 1, 42)
+      SolidQueue::ReadyExecution.claim("*", SolidQueue::Job.count + 1, @process)
     end
   end
 
   test "claim jobs using queue prefixes" do
     AddToBufferJob.perform_later("hey")
 
     assert_claimed_jobs(1) do
-      SolidQueue::ReadyExecution.claim("backgr*", SolidQueue::Job.count + 1, 42)
+      SolidQueue::ReadyExecution.claim("backgr*", SolidQueue::Job.count + 1, @process)
     end
 
     assert @jobs.none?(&:claimed?)
@@ -73,7 +74,7 @@ class SolidQueue::ReadyExecutionTest < ActiveSupport::TestCase
     SolidQueue::Queue.find_by_name("backend").pause
 
     assert_claimed_jobs(1) do
-      SolidQueue::ReadyExecution.claim("*", SolidQueue::Job.count + 1, 42)
+      SolidQueue::ReadyExecution.claim("*", SolidQueue::Job.count + 1, @process)
     end
 
     @jobs.each(&:reload)
@@ -84,15 +85,15 @@ class SolidQueue::ReadyExecutionTest < ActiveSupport::TestCase
     AddToBufferJob.perform_later("hey")
 
     assert_claimed_jobs(6) do
-      SolidQueue::ReadyExecution.claim(%w[ backe* background ], SolidQueue::Job.count + 1, 42)
+      SolidQueue::ReadyExecution.claim(%w[ backe* background ], SolidQueue::Job.count + 1, @process)
     end
   end
 
   test "claim jobs for queue without jobs at the moment using prefixes" do
     AddToBufferJob.perform_later("hey")
 
     assert_claimed_jobs(0) do
-      SolidQueue::ReadyExecution.claim(%w[ none* ], SolidQueue::Job.count + 1, 42)
+      SolidQueue::ReadyExecution.claim(%w[ none* ], SolidQueue::Job.count + 1, @process)
     end
   end
 
@@ -101,7 +102,7 @@ class SolidQueue::ReadyExecutionTest < ActiveSupport::TestCase
     job = SolidQueue::Job.last
 
     assert_claimed_jobs(3) do
-      SolidQueue::ReadyExecution.claim("*", 3, 42)
+      SolidQueue::ReadyExecution.claim("*", 3, @process)
     end
 
     assert job.reload.claimed?
@@ -117,7 +118,7 @@ class SolidQueue::ReadyExecutionTest < ActiveSupport::TestCase
     assert_equal "background", job.queue_name
 
     assert_claimed_jobs(3) do
-      SolidQueue::ReadyExecution.claim(%w[ background backend ], 3, 42)
+      SolidQueue::ReadyExecution.claim(%w[ background backend ], 3, @process)
     end
 
     assert job.reload.claimed?
@@ -136,7 +137,7 @@ class SolidQueue::ReadyExecutionTest < ActiveSupport::TestCase
     claimed_jobs = []
     4.times do
       assert_claimed_jobs(2) do
-        SolidQueue::ReadyExecution.claim(%w[ queue_b* queue_a* ], 2, 42)
+        SolidQueue::ReadyExecution.claim(%w[ queue_b* queue_a* ], 2, @process)
       end
 
       claimed_jobs += SolidQueue::ClaimedExecution.order(:id).last(2).map(&:job)
@@ -157,7 +158,7 @@ class SolidQueue::ReadyExecutionTest < ActiveSupport::TestCase
     claimed_jobs = []
     5.times do
       assert_claimed_jobs(2) do
-        SolidQueue::ReadyExecution.claim(%w[ queue_a2 queue_c1 queue_b* queue_c2 queue_a* ], 2, 42)
+        SolidQueue::ReadyExecution.claim(%w[ queue_a2 queue_c1 queue_b* queue_c2 queue_a* ], 2, @process)
       end
 
       claimed_jobs += SolidQueue::ClaimedExecution.order(:id).last(2).map(&:job)

test/unit/process_recovery_test.rb

@@ -28,7 +28,7 @@ class ProcessRecoveryTest < ActiveSupport::TestCase
 
     # Enqueue a job and manually claim it for the worker to avoid timing races
     job = enqueue_store_result_job(42)
-    claimed_execution = SolidQueue::ReadyExecution.claim("*", 5, worker_process.id).first
+    claimed_execution = SolidQueue::ReadyExecution.claim("*", 5, worker_process).first
     assert claimed_execution.present?
     assert_equal worker_process.id, claimed_execution.process_id
 

test/unit/supervisor_test.rb

@@ -111,7 +111,7 @@ class SupervisorTest < ActiveSupport::TestCase
     3.times { |i| StoreResultJob.set(queue: :new_queue).perform_later(i) }
     process = SolidQueue::Process.register(kind: "Worker", pid: 42, name: "worker-123")
 
-    SolidQueue::ReadyExecution.claim("*", 5, process.id)
+    SolidQueue::ReadyExecution.claim("*", 5, process)
 
     assert_equal 3, SolidQueue::ClaimedExecution.count
     assert_equal 0, SolidQueue::ReadyExecution.count
@@ -138,7 +138,7 @@ class SupervisorTest < ActiveSupport::TestCase
     4.times { |i| ThrottledUpdateResultJob.set(queue: :new_queue).perform_later(result) }
     process = SolidQueue::Process.register(kind: "Worker", pid: 42, name: "worker-123")
 
-    SolidQueue::ReadyExecution.claim("*", 5, process.id)
+    SolidQueue::ReadyExecution.claim("*", 5, process)
 
     assert_equal 3, SolidQueue::ClaimedExecution.count
     assert_equal 0, SolidQueue::ReadyExecution.count
@@ -193,7 +193,7 @@ class SupervisorTest < ActiveSupport::TestCase
     worker_process = SolidQueue::Process.register(kind: "Worker", pid: 999_999, name: worker_name)
 
     job = StoreResultJob.perform_later(42)
-    claimed_execution = SolidQueue::ReadyExecution.claim("*", 1, worker_process.id).first
+    claimed_execution = SolidQueue::ReadyExecution.claim("*", 1, worker_process).first
 
     terminated_fork = Struct.new(:name).new(worker_name)