Use ActiveJob 5.2 retry logic for old jobs

Rails 6 introduces retries per-exception, instead of a global count of
retries. Because ActiveJob 5.2 doesn't serialize the execution count
per-exception, when ActiveJob 6.0 picks up an "old" job it can't know
the exception count in the new format.

This can also be an issue if AJ 6.0 serializes a new job with
exception_executions which is later picked up by AJ 5.2, which would
clear exception_executions (since it has no knowledge of it).

Previously we handled this by resetting exception_executions, if it
wasn't defined on a job, which could result in the worst case retrying
the job 2x the times we should.

This commit changes how we handle loading a legacy job: instead of
resetting exception_executions, we instead will always use the global
executions count.

This way, jobs which only have one retry_on (and didn't have a behaviour
change in AJ 6) are backwards-and-forwards-compatible with counts
respected exactly.

Jobs with multiple retry_on will revert to the AJ5.2 behaviour if they
were ever run under AJ5.2.

activejob/lib/active_job/exceptions.rb

@@ -49,12 +49,10 @@ module ActiveJob
       #  end
       def retry_on(*exceptions, wait: 3.seconds, attempts: 5, queue: nil, priority: nil)
         rescue_from(*exceptions) do |error|
-          # Guard against jobs that were persisted before we started having individual executions counters per retry_on
+          executions = executions_for(exceptions)
-          self.exception_executions ||= {}
-          self.exception_executions[exceptions.to_s] = (exception_executions[exceptions.to_s] || 0) + 1
 
-          if exception_executions[exceptions.to_s] < attempts
+          if executions < attempts
-            retry_job wait: determine_delay(seconds_or_duration_or_algorithm: wait, executions: exception_executions[exceptions.to_s]), queue: queue, priority: priority, error: error
+            retry_job wait: determine_delay(seconds_or_duration_or_algorithm: wait, executions: executions), queue: queue, priority: priority, error: error
           else
             if block_given?
               instrument :retry_stopped, error: error do
@@ -146,5 +144,14 @@ module ActiveJob
 
         ActiveSupport::Notifications.instrument("#{name}.active_job", payload, &block)
       end
+
+      def executions_for(exceptions)
+        if exception_executions
+          exception_executions[exceptions.to_s] = (exception_executions[exceptions.to_s] || 0) + 1
+        else
+          # Guard against jobs that were persisted before we started having individual executions counters per retry_on
+          executions
+        end
+      end
   end
 end

activejob/test/cases/exceptions_test.rb

@@ -179,6 +179,31 @@ class ExceptionsTest < ActiveSupport::TestCase
     assert_equal ["Raised ActiveJob::DeserializationError for the 5 time"], JobBuffer.values
   end
 
+  test "running a job enqueued by AJ 5.2" do
+    job = RetryJob.new("DefaultsError", 6)
+    job.exception_executions = nil # This is how jobs from Rails 5.2 will look
+
+    assert_raises DefaultsError do
+      job.enqueue
+    end
+
+    assert_equal 5, JobBuffer.values.count
+  end
+
+  test "running a job enqueued and attempted under AJ 5.2" do
+    job = RetryJob.new("DefaultsError", 6)
+
+    # Fake 4 previous executions under AJ 5.2
+    job.exception_executions = nil
+    job.executions = 4
+
+    assert_raises DefaultsError do
+      job.enqueue
+    end
+
+    assert_equal ["Raised DefaultsError for the 5th time"], JobBuffer.values
+  end
+
   private
     def adapter_skips_scheduling?(queue_adapter)
       [