2 files changed, 27 insertions, 35 deletions

diff --git a/activejob/lib/active_job/exceptions.rb b/activejob/lib/active_job/exceptions.rb
index bb25afbca6..bc9e168971 100644
--- a/activejob/lib/active_job/exceptions.rb
+++ b/activejob/lib/active_job/exceptions.rb
@@ -49,18 +49,12 @@ module ActiveJob
           if executions < attempts
             retry_job wait: determine_delay(wait), queue: queue, priority: priority, error: error
           else
-            payload = {
-              job: self,
-              adapter: self.class.queue_adapter,
-              error: error
-            }
-
             if block_given?
-              ActiveSupport::Notifications.instrument("retry_stopped.active_job", payload) do
+              instrument :retry_stopped, error: error do
                 yield self, error
               end
             else
-              ActiveSupport::Notifications.instrument("retry_stopped.active_job", payload)
+              instrument :retry_stopped, error: error
               raise error
             end
           end
@@ -87,16 +81,8 @@ module ActiveJob
       #  end
       def discard_on(*exceptions)
         rescue_from(*exceptions) do |error|
-          payload = {
-            job: self,
-            adapter: self.class.queue_adapter,
-            error: error
-          }
-
-          ActiveSupport::Notifications.instrument("discard.active_job", payload) do
-            if block_given?
-              yield self, error
-            end
+          instrument :discard, error: error do
+            yield self, error if block_given?
           end
         end
       end
@@ -124,14 +110,7 @@ module ActiveJob
     #    end
     #  end
     def retry_job(options = {})
-      payload = {
-        job: self,
-        adapter: self.class.queue_adapter,
-        error: options[:error],
-        wait: options[:wait]
-      }
-
-      ActiveSupport::Notifications.instrument("enqueue_retry.active_job", payload) do
+      instrument :enqueue_retry, options.slice(:error, :wait) do
         enqueue options
       end
     end
@@ -154,5 +133,11 @@ module ActiveJob
           raise "Couldn't determine a delay based on #{seconds_or_duration_or_algorithm.inspect}"
         end
       end
+
+      def instrument(name, error: nil, wait: nil, &block)
+        payload = { job: self, adapter: self.class.queue_adapter, error: error, wait: wait }
+
+        ActiveSupport::Notifications.instrument("#{name}.active_job", payload, &block)
+      end
   end
 end
diff --git a/activejob/lib/active_job/test_helper.rb b/activejob/lib/active_job/test_helper.rb
index bb9e3e6ca4..9efc8c0c12 100644
--- a/activejob/lib/active_job/test_helper.rb
+++ b/activejob/lib/active_job/test_helper.rb
@@ -345,7 +345,7 @@ module ActiveJob
     #   end
     def assert_enqueued_with(job: nil, args: nil, at: nil, queue: nil)
       expected = { job: job, args: args, at: at, queue: queue }.compact
-      serialized_args = serialize_args_for_assertion(expected)
+      expected_args = prepare_args_for_assertion(expected)
 
       if block_given?
         original_enqueued_jobs_count = enqueued_jobs.count
@@ -358,7 +358,8 @@ module ActiveJob
       end
 
       matching_job = jobs.find do |enqueued_job|
-        serialized_args.all? { |key, value| value == enqueued_job[key] }
+        deserialized_job = deserialize_args_for_assertion(enqueued_job)
+        expected_args.all? { |key, value| value == deserialized_job[key] }
       end
 
       assert matching_job, "No enqueued job found with #{expected}"
@@ -396,7 +397,7 @@ module ActiveJob
     #   end
     def assert_performed_with(job: nil, args: nil, at: nil, queue: nil, &block)
       expected = { job: job, args: args, at: at, queue: queue }.compact
-      serialized_args = serialize_args_for_assertion(expected)
+      expected_args = prepare_args_for_assertion(expected)
 
       if block_given?
         original_performed_jobs_count = performed_jobs.count
@@ -408,8 +409,9 @@ module ActiveJob
         jobs = performed_jobs
       end
 
-      matching_job = jobs.find do |performed_job|
-        serialized_args.all? { |key, value| value == performed_job[key] }
+      matching_job = jobs.find do |enqueued_job|
+        deserialized_job = deserialize_args_for_assertion(enqueued_job)
+        expected_args.all? { |key, value| value == deserialized_job[key] }
       end
 
       assert matching_job, "No performed job found with #{expected}"
@@ -552,10 +554,15 @@ module ActiveJob
         end
       end
 
-      def serialize_args_for_assertion(args)
-        args.dup.tap do |serialized_args|
-          serialized_args[:args] = ActiveJob::Arguments.serialize(serialized_args[:args]) if serialized_args[:args]
-          serialized_args[:at]   = serialized_args[:at].to_f if serialized_args[:at]
+      def prepare_args_for_assertion(args)
+        args.dup.tap do |arguments|
+          arguments[:at] = arguments[:at].to_f if arguments[:at]
+        end
+      end
+
+      def deserialize_args_for_assertion(job)
+        job.dup.tap do |new_job|
+          new_job[:args] = ActiveJob::Arguments.deserialize(new_job[:args]) if new_job[:args]
         end
       end