Support in-flight jobs stored before individual execution counters for `retry_on` (#34731)

Also, make tests and examples for individual execution counters
clearer, as it wasn't entierly clear what would happen in this case:

```
retry_on CustomException, OtherException, attempts: 3
```

The job would be retried at most 3 times in total, for both
CustomException and OtherException. To have the job retry 3 times at
most for each exception individually, the following retry_on
declarations are necessary:

```
retry_on CustomException, attempts: 3
retry_on OtherException, attempts: 3
```

3 files changed, 39 insertions, 21 deletions

diff --git a/activejob/lib/active_job/exceptions.rb b/activejob/lib/active_job/exceptions.rb
index 53984a4e49..48b35c8d05 100644
--- a/activejob/lib/active_job/exceptions.rb
+++ b/activejob/lib/active_job/exceptions.rb
@@ -30,21 +30,28 @@ module ActiveJob
       #  class RemoteServiceJob < ActiveJob::Base
       #    retry_on CustomAppException # defaults to 3s wait, 5 attempts
       #    retry_on AnotherCustomAppException, wait: ->(executions) { executions * 2 }
+      #
+      #    retry_on ActiveRecord::Deadlocked, wait: 5.seconds, attempts: 3
+      #    retry_on Net::OpenTimeout, Timeout::Error, wait: :exponentially_longer, attempts: 10 # retries at most 10 times for Net::OpenTimeout and Timeout::Error combined
+      #    # To retry at most 10 times for each individual exception:
+      #    # retry_on Net::OpenTimeout, wait: :exponentially_longer, attempts: 10
+      #    # retry_on Timeout::Error, wait: :exponentially_longer, attempts: 10
+      #
       #    retry_on(YetAnotherCustomAppException) do |job, error|
       #      ExceptionNotifier.caught(error)
       #    end
-      #    retry_on ActiveRecord::Deadlocked, wait: 5.seconds, attempts: 3
-      #    retry_on Net::OpenTimeout, wait: :exponentially_longer, attempts: 10
       #
       #    def perform(*args)
       #      # Might raise CustomAppException, AnotherCustomAppException, or YetAnotherCustomAppException for something domain specific
       #      # Might raise ActiveRecord::Deadlocked when a local db deadlock is detected
-      #      # Might raise Net::OpenTimeout when the remote service is down
+      #      # Might raise Net::OpenTimeout or Timeout::Error when the remote service is down
       #    end
       #  end
       def retry_on(*exceptions, wait: 3.seconds, attempts: 5, queue: nil, priority: nil)
         rescue_from(*exceptions) do |error|
-          exception_executions[exceptions.to_s] += 1
+          # Guard against jobs that were persisted before we started having individual executions counters per retry_on
+          self.exception_executions ||= Hash.new(0)
+          self.exception_executions[exceptions.to_s] += 1
 
           if exception_executions[exceptions.to_s] < attempts
             retry_job wait: determine_delay(wait), queue: queue, priority: priority, error: error
diff --git a/activejob/test/cases/exceptions_test.rb b/activejob/test/cases/exceptions_test.rb
index 67327e0bbd..cac48cb6cb 100644
--- a/activejob/test/cases/exceptions_test.rb
+++ b/activejob/test/cases/exceptions_test.rb
@@ -30,25 +30,44 @@ class ExceptionsTest < ActiveJob::TestCase
     end
   end
 
-  test "keeps the same attempts counter when several exceptions are listed in the same declaration" do
+  test "keeps the same attempts counter for several exceptions listed in the same retry_on declaration" do
     exceptions_to_raise = %w(FirstRetryableErrorOfTwo FirstRetryableErrorOfTwo FirstRetryableErrorOfTwo
                              SecondRetryableErrorOfTwo SecondRetryableErrorOfTwo)
 
     assert_raises SecondRetryableErrorOfTwo do
       perform_enqueued_jobs do
-        ExceptionRetryJob.perform_later(exceptions_to_raise)
+        RetryJob.perform_later(exceptions_to_raise, 5)
       end
+
+      assert_equal [
+        "Raised FirstRetryableErrorOfTwo for the 1st time",
+        "Raised FirstRetryableErrorOfTwo for the 2nd time",
+        "Raised FirstRetryableErrorOfTwo for the 3rd time",
+        "Raised SecondRetryableErrorOfTwo for the 4th time",
+        "Raised SecondRetryableErrorOfTwo for the 5th time",
+      ], JobBuffer.values
     end
   end
 
-  test "keeps a separate attempts counter for each individual declaration" do
-    exceptions_to_raise = %w(FirstRetryableErrorOfTwo FirstRetryableErrorOfTwo FirstRetryableErrorOfTwo
-                             DefaultsError DefaultsError)
+  test "keeps a separate attempts counter for each individual retry_on declaration" do
+    exceptions_to_raise = %w(DefaultsError DefaultsError DefaultsError DefaultsError
+                             FirstRetryableErrorOfTwo FirstRetryableErrorOfTwo FirstRetryableErrorOfTwo)
 
     assert_nothing_raised do
       perform_enqueued_jobs do
-        ExceptionRetryJob.perform_later(exceptions_to_raise)
+        RetryJob.perform_later(exceptions_to_raise, 10)
       end
+
+      assert_equal [
+        "Raised DefaultsError for the 1st time",
+        "Raised DefaultsError for the 2nd time",
+        "Raised DefaultsError for the 3rd time",
+        "Raised DefaultsError for the 4th time",
+        "Raised FirstRetryableErrorOfTwo for the 5th time",
+        "Raised FirstRetryableErrorOfTwo for the 6th time",
+        "Raised FirstRetryableErrorOfTwo for the 7th time",
+        "Successfully completed job"
+      ], JobBuffer.values
     end
   end
 
diff --git a/activejob/test/jobs/retry_job.rb b/activejob/test/jobs/retry_job.rb
index 2d19d4c41e..3b0dce1a3c 100644
--- a/activejob/test/jobs/retry_job.rb
+++ b/activejob/test/jobs/retry_job.rb
@@ -18,7 +18,7 @@ class CustomDiscardableError < StandardError; end
 
 class RetryJob < ActiveJob::Base
   retry_on DefaultsError
-  retry_on FirstRetryableErrorOfTwo, SecondRetryableErrorOfTwo
+  retry_on FirstRetryableErrorOfTwo, SecondRetryableErrorOfTwo, attempts: 4
   retry_on LongWaitError, wait: 1.hour, attempts: 10
   retry_on ShortWaitTenAttemptsError, wait: 1.second, attempts: 10
   retry_on ExponentialWaitTenAttemptsError, wait: :exponentially_longer, attempts: 10
@@ -31,7 +31,8 @@ class RetryJob < ActiveJob::Base
   discard_on(CustomDiscardableError) { |job, error| JobBuffer.add("Dealt with a job that was discarded in a custom way. Message: #{error.message}") }
 
   def perform(raising, attempts)
-    if executions < attempts
+    raising = raising.shift if raising.is_a?(Array)
+    if raising && executions < attempts
       JobBuffer.add("Raised #{raising} for the #{executions.ordinalize} time")
       raise raising.constantize
     else
@@ -39,12 +40,3 @@ class RetryJob < ActiveJob::Base
     end
   end
 end
-
-class ExceptionRetryJob < ActiveJob::Base
-  retry_on FirstRetryableErrorOfTwo, SecondRetryableErrorOfTwo, attempts: 4
-  retry_on DefaultsError
-
-  def perform(exceptions)
-    raise exceptions.shift.constantize.new unless exceptions.empty?
-  end
-end