Raise a custom exception instead of ActiveJob::EnqueueError on enqueue

The reason is that we can't easily control what happens when something
goes wrong on enqueuing if the jobs are enqueued by Rails or other gems
(eg. Turbo::Streams::BroadcastJob, ActiveStorage::AnalyzeJob...),
because the only way is to check what the return value of the call to
`perform_later` is, or to pass a block to that method to fetch the error
that gets set in `enqueue_error`. In this way at least the error will
bubble up and you can rescue or let it be raised.

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
@@ -38,7 +40,7 @@ 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 = ActiveJob::EnqueueError.new("#{e.class.name}: #{e.message}").tap do |error|
+          enqueue_error = EnqueueError.new("#{e.class.name}: #{e.message}").tap do |error|
             error.set_backtrace e.backtrace
           end
           raise enqueue_error

lib/solid_queue/dispatcher/recurring_task.rb

@@ -36,12 +36,18 @@ def enqueue(at:)
         else
           payload[:other_adapter] = true
 
-          perform_later
+          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
 
@@ -70,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,7 +40,7 @@ def release_blocked(event)
   end
 
   def enqueue_recurring_task(event)
-    attributes = event.payload.slice(:task, :active_job_id, :at)
+    attributes = event.payload.slice(:task, :active_job_id, :enqueue_error, :at)
 
     if event.payload[:other_adapter]
       action = attributes[:active_job_id].present? ? "Enqueued recurring task outside Solid Queue" : "Error enqueuing recurring task"

test/integration/instrumentation_test.rb

@@ -280,7 +280,7 @@ class InstrumentationTest < ActiveSupport::TestCase
     end
   end
 
-  test "an error enqueuing a recurring task is reflected in the enqueue_recurring_task event" do
+  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)
 
@@ -295,11 +295,34 @@ class InstrumentationTest < ActiveSupport::TestCase
     assert events.size >= 1
     event = events.last
 
-    assert_event event, "enqueue_recurring_task", task: :example_task
+    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,20 +241,17 @@ class NonOverlappingGroupedJob2 < NonOverlappingJob
     end
   end
 
-  test "raise ActiveJob::EnqueueError when there's an ActiveRecordError" do
+  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 ActiveJob::EnqueueError do
+    assert_raises SolidQueue::Job::EnqueueError do
       SolidQueue::Job.enqueue(active_job)
     end
 
-    enqueue_result = AddToBufferJob.perform_later(1) do |job|
-      assert job.enqueue_error.is_a? ActiveJob::EnqueueError
-      assert_not job.successfully_enqueued?
+    assert_raises SolidQueue::Job::EnqueueError do
+      AddToBufferJob.perform_later(1)
     end
-
-    assert_not enqueue_result
   end
 
   if ENV["SEPARATE_CONNECTION"] && ENV["TARGET_DB"] != "sqlite"

test/unit/log_subscriber_test.rb

@@ -42,9 +42,9 @@ def set_logger(logger)
 
   test "error enqueuing recurring task" do
     attach_log_subscriber
-    instrument "enqueue_recurring_task.solid_queue", task: :example_task, at: Time.now
+    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"
+    assert_match_logged :info, "Error enqueuing recurring task", "task: :example_task, enqueue_error: \"Everything is broken\""
   end
 
   private

test/unit/recurring_task_test.rb

@@ -85,7 +85,7 @@ def perform
     end
   end
 
-  test "error when enqueuing job using another adapter" do
+  test "error when enqueuing job using another adapter that raises ActiveJob::EnqueueError" do
     ActiveJob::QueueAdapters::AsyncAdapter.any_instance.stubs(:enqueue).raises(ActiveJob::EnqueueError)
     previous_size = JobBuffer.size