[Deferred] Add configurable retry options (#215)

Add retry_on class method to Rage::Deferred::Task allowing per-task configuration of retry behavior:
- Custom retry count via attempts: (default: 5)
- Retryable exception classes (default: all exceptions)

Author: Digvijay-x1

CHANGELOG.md

@@ -2,6 +2,7 @@
 
 ### Added
 
+- [Deferred] Support configurable retry options with `retry_on` by [@Digvijay](https://github.com/Digvijay-x1) (#215).
 - [Cable] Add support for `stop_stream_from` and `stop_stream_for`  by [@Digvijay](https://github.com/Digvijay-x1) (#217).
 
 ## [1.21.1] - 2026-02-27

lib/rage/deferred/queue.rb

@@ -38,13 +38,13 @@ def schedule(task_id, context, publish_in: nil)
         Fiber.schedule do
           Iodine.task_inc!
 
-          is_completed = task.new.__perform(context)
+          result = task.new.__perform(context)
 
-          if is_completed
+          if result == true
             @backend.remove(task_id)
           else
             attempts = Rage::Deferred::Context.inc_attempts(context)
-            if task.__should_retry?(attempts)
+            if task.__should_retry?(attempts, result)
               enqueue(context, delay: task.__next_retry_in(attempts), task_id:)
             else
               @backend.remove(task_id)

lib/rage/deferred/task.rb

@@ -85,7 +85,7 @@ def __perform(context)
         Rage.logger.error("Deferred task failed with exception: #{e.class} (#{e.message}):\n#{e.backtrace.join("\n")}")
       end
     end
-    false
+    e
   end
 
   private def restore_log_info(context)
@@ -105,6 +105,20 @@ def self.included(klass)
   end
 
   module ClassMethods
+    # Configure retry behavior for this task.
+    #
+    # @param exception_classes [Class] one or more exception classes to retry on.
+    #   If none are specified, the task will be retried on any exception.
+    # @param attempts [Integer] the maximum number of retry attempts (default: 5)
+    # @example Retry on specific exceptions with custom attempts
+    #   retry_on Net::OpenTimeout, Net::ReadTimeout, attempts: 10
+    # @example Retry on all exceptions with custom attempts
+    #   retry_on attempts: 3
+    def retry_on(*exception_classes, attempts: MAX_ATTEMPTS)
+      @__retry_exceptions = exception_classes.empty? ? nil : exception_classes
+      @__retry_max_attempts = attempts
+    end
+
     def enqueue(*args, delay: nil, delay_until: nil, **kwargs)
       context = Rage::Deferred::Context.build(self, args, kwargs)
 
@@ -118,8 +132,11 @@ def enqueue(*args, delay: nil, delay_until: nil, **kwargs)
     end
 
     # @private
-    def __should_retry?(attempts)
-      attempts < MAX_ATTEMPTS
+    def __should_retry?(attempts, exception = nil)
+      return false if attempts >= (@__retry_max_attempts || MAX_ATTEMPTS)
+      return true if @__retry_exceptions.nil?
+
+      @__retry_exceptions.any? { |klass| exception.is_a?(klass) }
     end
 
     # @private

spec/deferred/queue_spec.rb

@@ -2,8 +2,9 @@
 
 RSpec.describe Rage::Deferred::Queue do
   let(:backend) { double("backend", add: "task-id", remove: nil) }
-  let(:task_class) { double("task_class", __should_retry?: false) }
+  let(:task_class) { double("task_class") }
   let(:task_instance) { double("task_instance", __perform: true) }
+
   let(:task_context) { "TestTask" }
   let(:backpressure_config) { nil }
 
@@ -99,14 +100,16 @@
     end
 
     context "when task fails" do
+      let(:error) { StandardError.new("Something went wrong") }
+
       before do
-        allow(task_instance).to receive(:__perform).and_return(false)
+        allow(task_instance).to receive(:__perform).and_return(error)
         allow(Rage::Deferred::Context).to receive(:inc_attempts).with(task_context).and_return(1)
       end
 
       context "and should be retried" do
         it "re-enqueues the task with a delay" do
-          allow(task_class).to receive(:__should_retry?).with(1).and_return(true)
+          allow(task_class).to receive(:__should_retry?).with(1, error).and_return(true)
           allow(task_class).to receive(:__next_retry_in).with(1).and_return(30)
           expect(subject).to receive(:enqueue).with(task_context, delay: 30, task_id: "task-id")
           subject.schedule("task-id", task_context)
@@ -115,7 +118,7 @@
 
       context "and should not be retried" do
         it "removes the task from the backend" do
-          allow(task_class).to receive(:__should_retry?).with(1).and_return(false)
+          allow(task_class).to receive(:__should_retry?).with(1, error).and_return(false)
           expect(backend).to receive(:remove).with("task-id")
           subject.schedule("task-id", task_context)
         end

spec/deferred/task_spec.rb

@@ -63,6 +63,56 @@ def perform(arg, kwarg:)
     end
   end
 
+  describe ".retry_on" do
+    context "with specific exceptions" do
+      let(:retryable_error_a) { Class.new(StandardError) }
+      let(:retryable_error_b) { Class.new(StandardError) }
+
+      before { task_class.retry_on(retryable_error_a, retryable_error_b, attempts: 3) }
+
+      it "retries on matching exception" do
+        expect(task_class.__should_retry?(1, retryable_error_a.new)).to be(true)
+      end
+
+      it "retries on subclass of matching exception" do
+        expect(task_class.__should_retry?(1, retryable_error_b.new)).to be(true)
+      end
+
+      it "does not retry on non-matching exception" do
+        expect(task_class.__should_retry?(1, StandardError.new)).to be(false)
+      end
+
+      it "respects custom attempts" do
+        expect(task_class.__should_retry?(2, retryable_error_a.new)).to be(true)
+        expect(task_class.__should_retry?(3, retryable_error_a.new)).to be(false)
+      end
+    end
+
+    context "with attempts only" do
+      before { task_class.retry_on(attempts: 2) }
+
+      it "retries on any exception" do
+        expect(task_class.__should_retry?(1, StandardError.new)).to be(true)
+      end
+
+      it "stops after max attempts" do
+        expect(task_class.__should_retry?(2, StandardError.new)).to be(false)
+      end
+    end
+
+    context "without retry_on (default behavior)" do
+      it "retries up to 5 times on any exception" do
+        expect(task_class.__should_retry?(4, RuntimeError.new)).to be(true)
+        expect(task_class.__should_retry?(5, RuntimeError.new)).to be(false)
+      end
+
+      it "retries without exception argument" do
+        expect(task_class.__should_retry?(4)).to be(true)
+        expect(task_class.__should_retry?(5)).to be(false)
+      end
+    end
+  end
+
   describe "#__perform" do
     let(:task) { task_class.new }
     let(:context) { double }
@@ -145,8 +195,8 @@ def perform(arg, kwarg:)
         expect(logger).to have_received(:error).with("Deferred task failed with exception: StandardError (Something went wrong):\nline 1\nline 2")
       end
 
-      it "returns false" do
-        expect(task.__perform(context)).to be(false)
+      it "returns the exception" do
+        expect(task.__perform(context)).to be(error)
       end
 
       context "with suppressed exception logging" do