Merge pull request #37830 from Edouard-chin/ec-activejob-enqueue-and-perform-log

Fix ActiveJob logging when callback chain is halted:

activejob/CHANGELOG.md

+*   Fix enqueuing and performing incorrect logging message.
+
+    Jobs will no longer always log "Enqueued MyJob" or "Performed MyJob" when they actually didn't get enqueued/performed.
+
+    ```ruby
+      class MyJob < ApplicationJob
+        before_enqueue { throw(:abort) }
+      end
+
+      MyJob.perform_later # Will no longer log "Enqueud MyJob" since job wasn't even enqueued through adapter.
+    ```
+
+    A new message will be logged in case a job couldn't be enqueued, either because the callback chain was halted or
+    because an exception happened during enqueing. (i.e. Redis is down when you try to enqueue your job)
+
+    *Edouard Chin*
+
 *   Add an option to disable logging of the job arguments when enqueuing and executing the job.
 
         class SensitiveJob < ApplicationJob

activejob/lib/active_job/instrumentation.rb

@@ -17,8 +17,13 @@ module Instrumentation #:nodoc:
 
     private
       def instrument(operation, payload = {}, &block)
+        enhanced_block = ->(event_payload) do
+          aborted = !block.call if block
+          event_payload[:aborted] = true if aborted
+        end
+
         ActiveSupport::Notifications.instrument \
-          "#{operation}.active_job", payload.merge(adapter: queue_adapter, job: self), &block
+          "#{operation}.active_job", payload.merge(adapter: queue_adapter, job: self), &enhanced_block
       end
   end
 end

activejob/lib/active_job/log_subscriber.rb

@@ -5,18 +5,42 @@
 module ActiveJob
   class LogSubscriber < ActiveSupport::LogSubscriber #:nodoc:
     def enqueue(event)
-      info do
       job = event.payload[:job]
+      ex = event.payload[:exception_object]
+
+      if ex
+        error do
+          "Failed enqueuing #{job.class.name} to #{queue_name(event)}: #{ex.class} (#{ex.message}):\n" + Array(ex.backtrace).join("\n")
+        end
+      elsif event.payload[:aborted]
+        info do
+          "Failed enqueuing #{job.class.name} to #{queue_name(event)}, a before_enqueue callback halted the enqueuing execution."
+        end
+      else
+        info do
           "Enqueued #{job.class.name} (Job ID: #{job.job_id}) to #{queue_name(event)}" + args_info(job)
         end
       end
+    end
 
     def enqueue_at(event)
-      info do
       job = event.payload[:job]
+      ex = event.payload[:exception_object]
+
+      if ex
+        error do
+          "Failed enqueuing #{job.class.name} to #{queue_name(event)}: #{ex.class} (#{ex.message}):\n" + Array(ex.backtrace).join("\n")
+        end
+      elsif event.payload[:aborted]
+        info do
+          "Failed enqueuing #{job.class.name} to #{queue_name(event)}, a before_enqueue callback halted the enqueuing execution."
+        end
+      else
+        info do
           "Enqueued #{job.class.name} (Job ID: #{job.job_id}) to #{queue_name(event)} at #{scheduled_at(event)}" + args_info(job)
         end
       end
+    end
 
     def perform_start(event)
       info do
@@ -32,6 +56,10 @@ def perform(event)
         error do
           "Error performing #{job.class.name} (Job ID: #{job.job_id}) from #{queue_name(event)} in #{event.duration.round(2)}ms: #{ex.class} (#{ex.message}):\n" + Array(ex.backtrace).join("\n")
         end
+      elsif event.payload[:aborted]
+        error do
+          "Error performing #{job.class.name} (Job ID: #{job.job_id}) from #{queue_name(event)} in #{event.duration.round(2)}ms: a before_perform callback halted the job execution"
+        end
       else
         info do
           "Performed #{job.class.name} (Job ID: #{job.job_id}) from #{queue_name(event)} in #{event.duration.round(2)}ms"

activejob/test/cases/logging_test.rb

@@ -10,6 +10,7 @@
 require "jobs/rescue_job"
 require "jobs/retry_job"
 require "jobs/disable_log_job"
+require "jobs/abort_before_enqueue_job"
 require "models/person"
 
 class LoggingTest < ActiveSupport::TestCase
@@ -112,6 +113,27 @@ def test_enqueue_job_logging
     assert_equal(key, "enqueue.active_job")
   end
 
+  def test_enqueue_job_log_error_when_callback_chain_is_halted
+    events = subscribed { AbortBeforeEnqueueJob.perform_later }
+    assert_match(/Failed enqueuing AbortBeforeEnqueueJob.* a before_enqueue callback halted/, @logger.messages)
+    assert_equal(events.count, 1)
+    key, * = events.first
+    assert_equal(key, "enqueue.active_job")
+  end
+
+  def test_enqueue_job_log_error_when_error_is_raised_during_callback_chain
+    events = subscribed do
+      assert_raises(AbortBeforeEnqueueJob::MyError) do
+        AbortBeforeEnqueueJob.perform_later(:raise)
+      end
+    end
+
+    assert_match(/Failed enqueuing AbortBeforeEnqueueJob/, @logger.messages)
+    assert_equal(events.count, 1)
+    key, * = events.first
+    assert_equal(key, "enqueue.active_job")
+  end
+
   def test_perform_job_logging
     perform_enqueued_jobs do
       LoggingJob.perform_later "Dummy"
@@ -123,6 +145,11 @@ def test_perform_job_logging
     end
   end
 
+  def test_perform_job_log_error_when_callback_chain_is_halted
+    subscribed { AbortBeforeEnqueueJob.perform_now }
+    assert_match(/Error performing AbortBeforeEnqueueJob.* a before_perform callback halted/, @logger.messages)
+  end
+
   def test_perform_disabled_job_logging
     perform_enqueued_jobs do
       DisableLogJob.perform_later "Dummy"
@@ -159,6 +186,27 @@ def test_enqueue_at_job_logging
     skip
   end
 
+  def test_enqueue_at_job_log_error_when_callback_chain_is_halted
+    events = subscribed { AbortBeforeEnqueueJob.set(wait: 1.second).perform_later }
+    assert_match(/Failed enqueuing AbortBeforeEnqueueJob.* a before_enqueue callback halted/, @logger.messages)
+    assert_equal(events.count, 1)
+    key, * = events.first
+    assert_equal(key, "enqueue_at.active_job")
+  end
+
+  def test_enqueue_at_job_log_error_when_error_is_raised_during_callback_chain
+    events = subscribed do
+      assert_raises(AbortBeforeEnqueueJob::MyError) do
+        AbortBeforeEnqueueJob.set(wait: 1.second).perform_later(:raise)
+      end
+    end
+
+    assert_match(/Failed enqueuing AbortBeforeEnqueueJob/, @logger.messages)
+    assert_equal(events.count, 1)
+    key, * = events.first
+    assert_equal(key, "enqueue_at.active_job")
+  end
+
   def test_enqueue_in_job_logging
     events = subscribed { HelloJob.set(wait: 2.seconds).perform_later "Cristian" }
     assert_match(/Enqueued HelloJob \(Job ID: .*\) to .*? at.*Cristian/, @logger.messages)

activejob/test/jobs/abort_before_enqueue_job.rb

 # frozen_string_literal: true
 
 class AbortBeforeEnqueueJob < ActiveJob::Base
-  before_enqueue { throw(:abort) }
+  MyError = Class.new(StandardError)
+
+  before_enqueue :throw_or_raise
+  before_perform { throw(:abort) }
 
   def perform
     raise "This should never be called"
   end
+
+  def throw_or_raise
+    if (arguments.first || :abort) == :abort
+      throw(:abort)
+    else
+      raise(MyError)
+    end
+  end
 end