Write tests for instrumentation and fix a few bugs found along the way

Fix bug found for async execution mode when worker didn't correctly finish
shutting down.

Also, always instrument retry_all and release_many_blocked even if none is
retried/released, so we have that information as well, with the number of jobs
that actually were acted on.

Author: rosa

app/models/solid_queue/execution/dispatching.rb

@@ -9,8 +9,11 @@ module Dispatching
         def dispatch_jobs(job_ids)
           jobs = Job.where(id: job_ids)
 
-          Job.dispatch_all(jobs).map(&:id).tap do |dispatched_job_ids|
-            where(job_id: dispatched_job_ids).order(:job_id).delete_all
+          Job.dispatch_all(jobs).map(&:id).then do |dispatched_job_ids|
+            if dispatched_job_ids.none? then 0
+            else
+              where(job_id: dispatched_job_ids).order(:job_id).delete_all
+            end
           end
         end
       end

app/models/solid_queue/scheduled_execution.rb

@@ -12,12 +12,12 @@ class ScheduledExecution < Execution
 
     class << self
       def dispatch_next_batch(batch_size)
-        SolidQueue.instrument(:dispatch_scheduled, batch_size: batch_size, count: 0) do |payload|
-          transaction do
-            job_ids = next_batch(batch_size).non_blocking_lock.pluck(:job_id)
-            if job_ids.empty? then []
-            else
-              payload[:count] = dispatch_jobs(job_ids)
+        transaction do
+          job_ids = next_batch(batch_size).non_blocking_lock.pluck(:job_id)
+          if job_ids.empty? then []
+          else
+            SolidQueue.instrument(:dispatch_scheduled, batch_size: batch_size) do |payload|
+              payload[:size] = dispatch_jobs(job_ids)
             end
           end
         end

lib/solid_queue/log_subscriber.rb

@@ -3,12 +3,8 @@
 require "active_support/log_subscriber"
 
 class SolidQueue::LogSubscriber < ActiveSupport::LogSubscriber
-  def release_many_blocked(event)
-    debug formatted_event(event, action: "Unblock jobs", **event.payload.slice(:limit, :size))
-  end
-
-  def release_blocked(event)
-    debug formatted_event(event, action: "Release blocked job", **event.payload.slice(:job_id, :concurrency_key, :released))
+  def dispatch_scheduled(event)
+    debug formatted_event(event, action: "Dispatch scheduled jobs", **event.payload.slice(:batch_size, :size))
   end
 
   def release_many_claimed(event)
@@ -19,10 +15,6 @@ def release_claimed(event)
     debug formatted_event(event, action: "Release claimed job", **event.payload.slice(:job_id, :process_id))
   end
 
-  def dispatch_scheduled(event)
-    debug formatted_event(event, action: "Dispatch scheduled jobs", **event.payload.slice(:batch_size, :size))
-  end
-
   def retry_all(event)
     debug formatted_event(event, action: "Retry failed jobs", **event.payload.slice(:jobs_size, :size))
   end
@@ -31,6 +23,14 @@ def retry(event)
     debug formatted_event(event, action: "Retry failed job", **event.payload.slice(:job_id))
   end
 
+  def release_many_blocked(event)
+    debug formatted_event(event, action: "Unblock jobs", **event.payload.slice(:limit, :size))
+  end
+
+  def release_blocked(event)
+    debug formatted_event(event, action: "Release blocked job", **event.payload.slice(:job_id, :concurrency_key, :released))
+  end
+
   def register_process(event)
     attributes = event.payload.slice(:kind, :pid, :hostname)
 

lib/solid_queue/worker.rb

@@ -32,10 +32,10 @@ def claim_executions
       end
 
       def shutdown
-        super
-
         pool.shutdown
         pool.wait_for_termination(SolidQueue.shutdown_timeout)
+
+        super
       end
 
       def all_work_completed?

test/integration/instrumentation_test.rb

@@ -0,0 +1,170 @@
+# frozen_string_literal: true
+
+require "test_helper"
+
+class InstrumentationTest < ActiveSupport::TestCase
+  test "dispatcher polling emits dispatch_scheduled event" do
+    8.times { AddToBufferJob.set(wait: 1.day).perform_later("I'm scheduled") }
+
+    events = subscribed("dispatch_scheduled.solid_queue") do
+      travel_to 2.days.from_now
+      dispatcher = SolidQueue::Dispatcher.new(polling_interval: 0.1, batch_size: 10).tap(&:start)
+
+      wait_while_with_timeout(0.5.seconds) { SolidQueue::ScheduledExecution.any? }
+      dispatcher.stop
+    end
+
+    assert_equal 1, events.size
+    assert_event events.first, "dispatch_scheduled", batch_size: 10, size: 8
+  end
+
+  test "stopping a worker with claimed executions emits release_claimed events" do
+    StoreResultJob.perform_later(42, pause: SolidQueue.shutdown_timeout + 10.second)
+    process = nil
+
+    events = subscribed(/release.*_claimed\.solid_queue/) do
+      worker = SolidQueue::Worker.new.tap(&:start)
+
+      wait_while_with_timeout(0.5.seconds) { SolidQueue::ReadyExecution.any? }
+      process = SolidQueue::Process.last
+
+      worker.stop
+    end
+
+    assert_equal 2, events.size
+    release_one_event, release_many_event = events
+    assert_event release_one_event, "release_claimed", job_id: SolidQueue::Job.last.id, process_id: process.id
+    assert_event release_many_event, "release_many_claimed", size: 1
+  end
+
+  test "starting and stopping a worker emits register_process and deregister_process events" do
+    process = nil
+
+    events = subscribed(/(register|deregister)_process\.solid_queue/) do
+      worker = SolidQueue::Worker.new.tap(&:start)
+      wait_for_registered_processes(1, timeout: 1.second)
+
+      process = SolidQueue::Process.last
+
+      worker.stop
+      wait_for_registered_processes(0, timeout: 1.second)
+    end
+
+    assert_equal 2, events.size
+    register_event, deregister_event = events
+
+    assert_event register_event, "register_process", kind: "Worker", pid: ::Process.pid
+    assert_event deregister_event, "deregister_process", process: process, pruned: false
+  end
+
+  test "pruning processes emit prune_processes and deregister_process events" do
+    time = Time.now
+    processes = 3.times.collect { |i| SolidQueue::Process.create!(kind: "Worker", supervisor_id: 42, pid: 10 + i, hostname: "localhost", last_heartbeat_at: time) }
+
+    # Heartbeats will expire
+    travel_to 3.days.from_now
+
+    events = subscribed(/.*process.*\.solid_queue/) do
+      SolidQueue::Process.prune
+    end
+
+    # 1 prune event + 3 deregister events
+    assert_equal 4, events.count
+    deregister_events = events.first(3)
+    prune_event = events.last
+
+    assert_event prune_event, "prune_processes", size: 3
+    deregister_events.each_with_index do |event, i|
+      assert_event event, "deregister_process", process: processes[i], pruned: true
+    end
+  end
+
+  test "retrying failed job emits retry event" do
+    RaisingJob.perform_later(RuntimeError, "A")
+    job = SolidQueue::Job.last
+
+    worker = SolidQueue::Worker.new.tap(&:start)
+    wait_for_jobs_to_finish_for(3.seconds)
+    worker.stop
+
+    events = subscribed("retry.solid_queue") do
+      job.reload.retry
+    end
+
+    assert_equal 1, events.size
+    assert_event events.first, "retry", job_id: job.id
+  end
+
+  test "retrying failed jobs in bulk emits retry_all" do
+    3.times { RaisingJob.perform_later(RuntimeError, "A") }
+    AddToBufferJob.perform_later("A")
+
+    jobs = SolidQueue::Job.last(4)
+
+    worker = SolidQueue::Worker.new.tap(&:start)
+    wait_for_jobs_to_finish_for(3.seconds)
+    worker.stop
+
+    events = subscribed("retry_all.solid_queue") do
+      SolidQueue::FailedExecution.retry_all(jobs)
+      SolidQueue::FailedExecution.retry_all(jobs)
+    end
+
+    assert_equal 2, events.size
+    assert_event events.first, "retry_all", jobs_size: 4, size: 3
+    assert_event events.second, "retry_all", jobs_size: 4, size: 0
+  end
+
+  test "unblocking job emits release_blocked event" do
+    result = JobResult.create!
+    # 1 ready, 2 blocked
+    3.times { SequentialUpdateResultJob.perform_later(result, name: name, pause: 0.2.seconds) }
+
+    # Simulate expiry of the concurrency locks
+    travel_to 3.days.from_now
+    SolidQueue::Semaphore.expired.delete_all
+
+    blocked_jobs = SolidQueue::BlockedExecution.last(2).map(&:job)
+    concurrency_key = blocked_jobs.first.concurrency_key
+
+    events = subscribed("release_blocked.solid_queue") do
+      SolidQueue::BlockedExecution.release_one(concurrency_key)
+      SolidQueue::BlockedExecution.release_one(concurrency_key)
+    end
+
+    assert_equal 2, events.size
+    assert_event events.first, "release_blocked", job_id: blocked_jobs.first.id, concurrency_key: concurrency_key, released: true
+    assert_event events.second, "release_blocked", job_id: blocked_jobs.second.id, concurrency_key: concurrency_key, released: false
+  end
+
+  test "unblocking jobs in bulk emits release_many_blocked event" do
+    result = JobResult.create!
+    # 1 ready, 3 blocked
+    4.times { SequentialUpdateResultJob.perform_later(result, name: name, pause: 0.2.seconds) }
+
+    # Simulate expiry of the concurrency locks
+    travel_to 3.days.from_now
+    SolidQueue::Semaphore.expired.delete_all
+
+    events = subscribed("release_many_blocked.solid_queue") do
+      SolidQueue::BlockedExecution.unblock(5)
+      SolidQueue::BlockedExecution.unblock(5)
+    end
+
+    assert_equal 2, events.size
+    assert_event events.first, "release_many_blocked", limit: 5, size: 1
+    assert_event events.second, "release_many_blocked", limit: 5, size: 0
+  end
+
+  private
+    def subscribed(name, &block)
+      [].tap do |events|
+        ActiveSupport::Notifications.subscribed(->(*args) { events << args }, name, &block)
+      end
+    end
+
+    def assert_event(event, action, **attributes)
+      assert_equal "#{action}.solid_queue", event.first
+      assert_equal attributes, event.last.slice(*attributes.keys)
+    end
+end

test/unit/log_subscriber_test.rb

@@ -17,9 +17,9 @@ class LogSubscriberTest < ActiveSupport::TestCase
 
   test "unblock many jobs" do
     attach_log_subscriber
-    instrument "unblock_batch.solid_queue", batch_size: 42
+    instrument "release_many_blocked.solid_queue", limit: 42, size: 10
 
-    assert_match_logged :debug, "Unblock jobs", "batch_size: 42"
+    assert_match_logged :debug, "Unblock jobs", "limit: 42, size: 10"
   end
 
   private