Merge pull request #252 from rails/enqueue-errors

Improve error handling on enqueuing

Author: rosa

README.md

@@ -173,6 +173,10 @@ There are several settings that control how Solid Queue works that you can set a
 - `default_concurrency_control_period`: the value to be used as the default for the `duration` parameter in [concurrency controls](#concurrency-controls). It defaults to 3 minutes.
 - `enqueue_after_transaction_commit`: whether the job queuing is deferred to after the current Active Record transaction is committed. The default is `false`. [Read more](https://github.com/rails/rails/pull/51426).
 
+## Errors when enqueuing
+Solid Queue will raise a `SolidQueue::Job::EnqueueError` for any Active Record errors that happen when enqueuing a job. The reason for not raising `ActiveJob::EnqueueError` is that this one gets handled by Active Job, causing `perform_later` to return `false` and set `job.enqueue_error`, yielding the job to a block that you need to pass to `perform_later`. This works very well for your own jobs, but makes failure very hard to handle for jobs enqueued by Rails or other gems, such as `Turbo::Streams::BroadcastJob` or `ActiveStorage::AnalyzeJob`, because you don't control the call to `perform_later` in that cases.
+
+In the case of recurring tasks, if such error is raised when enqueuing the job corresponding to the task, it'll be handled and logged but it won't bubble up.
 
 ## Concurrency controls
 Solid Queue extends Active Job with concurrency controls, that allows you to limit how many jobs of a certain type or with certain arguments can run at the same time. When limited in this way, jobs will be blocked from running, and they'll stay blocked until another job finishes and unblocks them, or after the set expiry time (concurrency limit's _duration_) elapses. Jobs are never discarded or lost, only blocked.

app/models/solid_queue/job.rb

@@ -2,6 +2,8 @@
 
 module SolidQueue
   class Job < Record
+    class EnqueueError < StandardError; end
+
     include Executable, Clearable, Recurrable
 
     serialize :arguments, coder: JSON
@@ -37,6 +39,11 @@ def enqueue(active_job, scheduled_at: Time.current)
 
         def create_from_active_job(active_job)
           create!(**attributes_from_active_job(active_job))
+        rescue ActiveRecord::ActiveRecordError => e
+          enqueue_error = EnqueueError.new("#{e.class.name}: #{e.message}").tap do |error|
+            error.set_backtrace e.backtrace
+          end
+          raise enqueue_error
         end
 
         def create_all_from_active_jobs(active_jobs)

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.rb

@@ -51,6 +51,10 @@ def unload_recurring_schedule
         recurring_schedule.unload_tasks
       end
 
+      def all_work_completed?
+        SolidQueue::ScheduledExecution.none? && recurring_schedule.empty?
+      end
+
       def set_procline
         procline "waiting"
       end

lib/solid_queue/dispatcher/recurring_schedule.rb

@@ -11,6 +11,10 @@ def initialize(tasks)
       @scheduled_tasks = Concurrent::Hash.new
     end
 
+    def empty?
+      configured_tasks.empty?
+    end
+
     def load_tasks
       configured_tasks.each do |task|
         load_task(task)

lib/solid_queue/dispatcher/recurring_task.rb

@@ -31,15 +31,23 @@ 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
+          perform_later do |job|
+            unless job.successfully_enqueued?
+              payload[:enqueue_error] = job.enqueue_error&.message
+            end
+          end
         end
+
+        payload[:active_job_id] = active_job.job_id if active_job
+      rescue RecurringExecution::AlreadyRecorded
+        payload[:skipped] = true
+      rescue Job::EnqueueError => error
+        payload[:enqueue_error] = error.message
       end
     end
 
@@ -68,8 +76,8 @@ def perform_later_and_record(run_at:)
         RecurringExecution.record(key, run_at) { perform_later }
       end
 
-      def perform_later
-        job_class.perform_later(*arguments_with_kwargs)
+      def perform_later(&block)
+        job_class.perform_later(*arguments_with_kwargs, &block)
       end
 
       def arguments_with_kwargs

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, :enqueue_error, :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,49 @@ class InstrumentationTest < ActiveSupport::TestCase
     end
   end
 
+  test "an error enqueuing a recurring task in Solid Queue 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, enqueue_error: "ActiveRecord::Deadlocked: ActiveRecord::Deadlocked"
+    assert event.last[:at].present?
+    assert_nil event.last[:other_adapter]
+  end
+
+  test "an error enqueuing a recurring task with another adapter is reflected in the enqueue_recurring_task event" do
+    AddToBufferJob.queue_adapter = :async
+    ActiveJob::QueueAdapters::AsyncAdapter.any_instance.stubs(:enqueue).raises(ActiveJob::EnqueueError.new("All is broken"))
+    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, enqueue_error: "All is broken"
+    assert event.last[:at].present?
+    assert event.last[:other_adapter]
+  ensure
+    AddToBufferJob.queue_adapter = :solid_queue
+  end
+
   test "thread errors emit thread_error events" do
     previous_thread_report_on_exception, Thread.report_on_exception = Thread.report_on_exception, false
 

test/models/solid_queue/job_test.rb

@@ -241,6 +241,19 @@ class NonOverlappingGroupedJob2 < NonOverlappingJob
     end
   end
 
+  test "raise EnqueueError when there's an ActiveRecordError" do
+    SolidQueue::Job.stubs(:create!).raises(ActiveRecord::Deadlocked)
+
+    active_job = AddToBufferJob.new(1).set(priority: 8, queue: "test")
+    assert_raises SolidQueue::Job::EnqueueError do
+      SolidQueue::Job.enqueue(active_job)
+    end
+
+    assert_raises SolidQueue::Job::EnqueueError do
+      AddToBufferJob.perform_later(1)
+    end
+  end
+
   if ENV["SEPARATE_CONNECTION"] && ENV["TARGET_DB"] != "sqlite"
     test "uses a different connection and transaction than the one in use when connects_to is specified" do
       assert_difference -> { SolidQueue::Job.count } do

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, enqueue_error: "Everything is broken", at: Time.now
+
+    assert_match_logged :info, "Error enqueuing recurring task", "task: :example_task, enqueue_error: \"Everything is broken\""
+  end
+
   private
     def attach_log_subscriber
       ActiveSupport::LogSubscriber.attach_to :solid_queue, SolidQueue::LogSubscriber.new