Instrument recurring task enqueuing

Author: rosa

app/models/solid_queue/recurring_execution.rb

@@ -7,12 +7,12 @@ class RecurringExecution < Execution
     class << self
       def record(task_key, run_at, &block)
         transaction do
-          if job_id = block.call
-            create!(job_id: job_id, task_key: task_key, run_at: run_at)
+          block.call.tap do |active_job|
+            create!(job_id: active_job.provider_job_id, task_key: task_key, run_at: run_at)
           end
         end
       rescue ActiveRecord::RecordNotUnique
-        SolidQueue.logger.info("[SolidQueue] Skipped recurring task #{task_key} at #{run_at} — already dispatched")
+        # Task already dispatched
       end
 
       def clear_in_batches(batch_size: 500)

lib/solid_queue/dispatcher/recurring_task.rb

@@ -30,10 +30,16 @@ def next_time
     end
 
     def enqueue(at:)
-      if using_solid_queue_adapter?
-        perform_later_and_record(run_at: at)
-      else
-        perform_later
+      SolidQueue.instrument(:enqueue_recurring_task, task: key, at: at) do |payload|
+        if using_solid_queue_adapter?
+          perform_later_and_record(run_at: at)
+        else
+          payload[:other_adapter] = true
+
+          perform_later
+        end.tap do |active_job|
+          payload[:active_job_id] = active_job&.job_id
+        end
       end
     end
 
@@ -59,7 +65,7 @@ def using_solid_queue_adapter?
       end
 
       def perform_later_and_record(run_at:)
-        RecurringExecution.record(key, run_at) { perform_later.provider_job_id }
+        RecurringExecution.record(key, run_at) { perform_later }
       end
 
       def perform_later

lib/solid_queue/log_subscriber.rb

@@ -39,6 +39,17 @@ def release_blocked(event)
     debug formatted_event(event, action: "Release blocked job", **event.payload.slice(:job_id, :concurrency_key, :released))
   end
 
+  def enqueue_recurring_task(event)
+    attributes = event.payload.slice(:task, :at, :active_job_id)
+
+    if event.payload[:other_adapter]
+      debug formatted_event(event, action: "Enqueued recurring task outside Solid Queue", **attributes)
+    else
+      action = attributes[:active_job_id].present? ? "Enqueued recurring task" : "Skipped recurring task – already dispatched"
+      info formatted_event(event, action: action, **attributes)
+    end
+  end
+
   def register_process(event)
     attributes = event.payload.slice(:kind, :pid, :hostname)
 

test/integration/instrumentation_test.rb

@@ -25,10 +25,11 @@ class InstrumentationTest < ActiveSupport::TestCase
     events = subscribed(/release.*_claimed\.solid_queue/) do
       worker = SolidQueue::Worker.new.tap(&:start)
 
-      wait_while_with_timeout(0.5.seconds) { SolidQueue::ReadyExecution.any? }
+      wait_while_with_timeout(1.seconds) { SolidQueue::ReadyExecution.any? }
       process = SolidQueue::Process.last
 
       worker.stop
+      wait_for_registered_processes(0, timeout: 1.second)
     end
 
     assert_equal 2, events.size
@@ -43,7 +44,7 @@ class InstrumentationTest < ActiveSupport::TestCase
 
     events = subscribed(/(register|deregister)_process\.solid_queue/) do
       worker = SolidQueue::Worker.new.tap(&:start)
-      wait_for_registered_processes(1, timeout: 1.second)
+      wait_while_with_timeout(1.seconds) { SolidQueue::ReadyExecution.any? }
 
       process = SolidQueue::Process.last
 
@@ -197,6 +198,48 @@ class InstrumentationTest < ActiveSupport::TestCase
     assert_event events.second, "release_many_blocked", limit: 5, size: 0
   end
 
+  test "enqueuing recurring task emits enqueue_recurring_task event" do
+    recurring_task = { example_task: { class: "AddToBufferJob", schedule: "every second", args: 42 } }
+    dispatcher = SolidQueue::Dispatcher.new(concurrency_maintenance: false, recurring_tasks: recurring_task)
+
+    events = subscribed("enqueue_recurring_task.solid_queue") do
+      dispatcher.start
+      sleep 1.01
+      dispatcher.stop
+    end
+
+    assert events.size >= 1
+    event = events.last
+
+    assert_event event, "enqueue_recurring_task", task: :example_task, active_job_id: SolidQueue::Job.last.active_job_id
+    assert event.last[:at].present?
+    assert_nil event.last[:other_adapter]
+  end
+
+  test "skipping a recurring task is reflected in the enqueue_recurring_task event" do
+    recurring_task = { example_task: { class: "AddToBufferJob", schedule: "every second", args: 42 } }
+    dispatchers = 2.times.collect { SolidQueue::Dispatcher.new(concurrency_maintenance: false, recurring_tasks: recurring_task) }
+
+    events = subscribed("enqueue_recurring_task.solid_queue") do
+      dispatchers.each(&:start)
+      sleep 1.01
+      dispatchers.each(&:stop)
+    end
+
+    assert events.size >= 2
+    events.each do |event|
+      assert_event event, "enqueue_recurring_task", task: :example_task
+    end
+
+    active_job_ids = SolidQueue::Job.all.map(&:active_job_id)
+    events.group_by { |event| event.last[:at] }.each do |_, events_by_time|
+      if events_by_time.many?
+        assert events_by_time.any? { |e| e.last[:active_job_id].nil? }
+        assert events_by_time.any? { |e| e.last[:active_job_id].in? active_job_ids }
+      end
+    end
+  end
+
   private
     def subscribed(name, &block)
       [].tap do |events|