Emit claim events when claiming jobs

Author: rosa

app/models/solid_queue/claimed_execution.rb

@@ -15,9 +15,14 @@ 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 } }
 
-      insert_all!(job_data)
-      where(job_id: job_ids, process_id: process_id).load.tap do |claimed|
-        block.call(claimed)
+      SolidQueue.instrument(:claim, process_id: process_id, job_ids: job_ids) do |payload|
+        insert_all!(job_data)
+        where(job_id: job_ids, process_id: process_id).load.tap do |claimed|
+          block.call(claimed)
+
+          payload[:size] = claimed.size
+          payload[:claimed_job_ids] = claimed.map(&:job_id)
+        end
       end
     end
 

lib/solid_queue/log_subscriber.rb

@@ -7,6 +7,10 @@ def dispatch_scheduled(event)
     debug formatted_event(event, action: "Dispatch scheduled jobs", **event.payload.slice(:batch_size, :size))
   end
 
+  def claim(event)
+    debug formatted_event(event, action: "Claim jobs", **event.payload.slice(:process_id, :job_ids, :claimed_job_ids, :size))
+  end
+
   def release_many_claimed(event)
     debug formatted_event(event, action: "Release claimed jobs", **event.payload.slice(:size))
   end
@@ -45,15 +49,15 @@ def enqueue_recurring_task(event)
 
     if event.payload[:other_adapter]
       action = attributes[:active_job_id].present? ? "Enqueued recurring task outside Solid Queue" : "Error enqueuing recurring task"
-      info formatted_event(event, action: action, **attributes)
+      debug formatted_event(event, action: action, **attributes)
     else
       action = case
       when event.payload[:skipped].present? then "Skipped recurring task – already dispatched"
       when attributes[:active_job_id].nil?  then "Error enqueuing recurring task"
       else                                       "Enqueued recurring task"
       end
 
-      info formatted_event(event, action: action, **attributes)
+      debug formatted_event(event, action: action, **attributes)
     end
   end
 

test/integration/instrumentation_test.rb

@@ -18,6 +18,25 @@ class InstrumentationTest < ActiveSupport::TestCase
     assert_event events.first, "dispatch_scheduled", batch_size: 10, size: 8
   end
 
+  test "claiming jobs emits claim events" do
+    3.times { StoreResultJob.perform_later(42) }
+    process = nil
+    jobs = SolidQueue::Job.last(3)
+
+    events = subscribed("claim.solid_queue") do
+      worker = SolidQueue::Worker.new.tap(&:start)
+
+      wait_while_with_timeout!(3.seconds) { SolidQueue::ReadyExecution.any? }
+      process = SolidQueue::Process.last
+
+      worker.stop
+      wait_for_registered_processes(0, timeout: 3.second)
+    end
+
+    assert_equal 1, events.size
+    assert_event events.first, "claim", process_id: process.id, job_ids: jobs.map(&:id), claimed_job_ids: jobs.map(&:id), size: 3
+  end
+
   test "stopping a worker with claimed executions emits release_claimed events" do
     StoreResultJob.perform_later(42, pause: SolidQueue.shutdown_timeout + 100.second)
     process = nil