Fix bug when enqueuing the job for a recurring task fails

As `perform_later` in this case returns `false`, which crashes when we
try to access `job_id` on it. `perform_later` returns the job instance
with `successfully_enqueued` set to true when it goes well, but `false`
when it fails. You can provide a block and the job instance with
`successfully_enqueued` set to `false` and an error set in
`enqueue_error` is yielded to that block, but it's not returned.

At first I tried to make this a bit more sophisticated, passing a block
to get the job and populate the instrumentation payload with it, but the
code was getting quite cumbersome and the error is instrumented by
Active Job anyway, so in the end I just simplified this and opted for
handling this case to avoid crashing.

Author: rosa

app/models/solid_queue/recurring_execution.rb

@@ -2,17 +2,21 @@
 
 module SolidQueue
   class RecurringExecution < Execution
+    class AlreadyRecorded < StandardError; end
+
     scope :clearable, -> { where.missing(:job) }
 
     class << self
       def record(task_key, run_at, &block)
         transaction do
           block.call.tap do |active_job|
-            create!(job_id: active_job.provider_job_id, task_key: task_key, run_at: run_at)
+            if active_job
+              create!(job_id: active_job.provider_job_id, task_key: task_key, run_at: run_at)
+            end
           end
         end
-      rescue ActiveRecord::RecordNotUnique
-        # Task already dispatched
+      rescue ActiveRecord::RecordNotUnique => e
+        raise AlreadyRecorded
       end
 
       def clear_in_batches(batch_size: 500)

lib/solid_queue/dispatcher/recurring_task.rb

@@ -31,15 +31,17 @@ def next_time
 
     def enqueue(at:)
       SolidQueue.instrument(:enqueue_recurring_task, task: key, at: at) do |payload|
-        if using_solid_queue_adapter?
+        active_job = 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
+
+        payload[:active_job_id] = active_job.job_id if active_job
+      rescue RecurringExecution::AlreadyRecorded
+        payload[:skipped] = true
       end
     end
 

lib/solid_queue/log_subscriber.rb

@@ -40,12 +40,18 @@ def release_blocked(event)
   end
 
   def enqueue_recurring_task(event)
-    attributes = event.payload.slice(:task, :at, :active_job_id)
+    attributes = event.payload.slice(:task, :active_job_id, :at)
 
     if event.payload[:other_adapter]
-      debug formatted_event(event, action: "Enqueued recurring task outside Solid Queue", **attributes)
+      action = attributes[:active_job_id].present? ? "Enqueued recurring task outside Solid Queue" : "Error enqueuing recurring task"
+      info formatted_event(event, action: action, **attributes)
     else
-      action = attributes[:active_job_id].present? ? "Enqueued recurring task" : "Skipped recurring task – already dispatched"
+      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)
     end
   end

test/integration/instrumentation_test.rb

@@ -280,6 +280,26 @@ class InstrumentationTest < ActiveSupport::TestCase
     end
   end
 
+  test "an error enqueuing a recurring task is reflected in the enqueue_recurring_task event" do
+    recurring_task = { example_task: { class: "AddToBufferJob", schedule: "every second", args: 42 } }
+    SolidQueue::Job.stubs(:create!).raises(ActiveRecord::Deadlocked)
+
+    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
+    assert event.last[:at].present?
+    assert_nil event.last[:other_adapter]
+  end
+
   test "thread errors emit thread_error events" do
     previous_thread_report_on_exception, Thread.report_on_exception = Thread.report_on_exception, false
 

test/unit/log_subscriber_test.rb

@@ -26,6 +26,27 @@ def set_logger(logger)
     assert_match_logged :debug, "Unblock jobs", "limit: 42, size: 10"
   end
 
+  test "recurring task enqueued succesfully" do
+    attach_log_subscriber
+    instrument "enqueue_recurring_task.solid_queue", task: :example_task, active_job_id: "b944ddbc-6a37-43c0-b661-4b56e57195f5", at: Time.now
+
+    assert_match_logged :info, "Enqueued recurring task", "task: :example_task, active_job_id: \"b944ddbc-6a37-43c0-b661-4b56e57195f5\""
+  end
+
+  test "recurring task skipped" do
+    attach_log_subscriber
+    instrument "enqueue_recurring_task.solid_queue", task: :example_task, skipped: true, at: Time.now
+
+    assert_match_logged :info, "Skipped recurring task – already dispatched", "task: :example_task"
+  end
+
+  test "error enqueuing recurring task" do
+    attach_log_subscriber
+    instrument "enqueue_recurring_task.solid_queue", task: :example_task, at: Time.now
+
+    assert_match_logged :info, "Error enqueuing recurring task", "task: :example_task"
+  end
+
   private
     def attach_log_subscriber
       ActiveSupport::LogSubscriber.attach_to :solid_queue, SolidQueue::LogSubscriber.new

test/unit/recurring_task_test.rb

@@ -25,6 +25,14 @@ def perform(value, options = {}, **kwargs)
     end
   end
 
+  class JobUsingAsyncAdapter < ApplicationJob
+    self.queue_adapter = :async
+
+    def perform
+      JobBuffer.add "job_using_async_adapter"
+    end
+  end
+
   setup do
     @worker = SolidQueue::Worker.new(queues: "*")
     @worker.mode = :inline
@@ -58,6 +66,37 @@ def perform(value, options = {}, **kwargs)
     enqueue_and_assert_performed_with_result task, [ "multiple_types", nil, 42 ]
   end
 
+  test "job using a different adapter" do
+    task = recurring_task_with(class_name: "JobUsingAsyncAdapter")
+    previous_size = JobBuffer.size
+
+    task.enqueue(at: Time.now)
+    wait_while_with_timeout!(0.5.seconds) { JobBuffer.size == previous_size }
+
+    assert_equal "job_using_async_adapter", JobBuffer.last_value
+  end
+
+  test "error when enqueuing job before recording task" do
+    SolidQueue::Job.stubs(:create!).raises(ActiveRecord::Deadlocked)
+
+    task = recurring_task_with(class_name: "JobWithoutArguments")
+    assert_no_difference -> { SolidQueue::Job.count } do
+      task.enqueue(at: Time.now)
+    end
+  end
+
+  test "error when enqueuing job using another adapter" do
+    ActiveJob::QueueAdapters::AsyncAdapter.any_instance.stubs(:enqueue).raises(ActiveJob::EnqueueError)
+    previous_size = JobBuffer.size
+
+    task = recurring_task_with(class_name: "JobUsingAsyncAdapter")
+    task.enqueue(at: Time.now)
+
+    wait_while_with_timeout(0.5.seconds) { JobBuffer.size == previous_size }
+
+    assert_equal previous_size, JobBuffer.size
+  end
+
   test "valid and invalid schedules" do
     assert_not recurring_task_with(class_name: "JobWithoutArguments", schedule: "once a year").valid?
     assert_not recurring_task_with(class_name: "JobWithoutArguments", schedule: "tomorrow").valid?