Add configurable retry backoff before dead-lettering

src/Dafda.Tests/Builders/ConsumerBuilder.cs

@@ -18,6 +18,7 @@ internal class ConsumerBuilder
     private IDeadLetterQueue _deadLetterQueue = NullDeadLetterQueue.Instance;
     private int _maxRetries;
     private Func<Exception, bool> _deadLetterQueueBypass;
+    private Func<int, TimeSpan> _retryBackoff;
 
     public ConsumerBuilder WithUnitOfWork(IHandlerUnitOfWork unitOfWork)
     {
@@ -78,6 +79,12 @@ public ConsumerBuilder WithDeadLetterQueueBypass(Func<Exception, bool> deadLette
         return this;
     }
 
+    public ConsumerBuilder WithRetryBackoff(Func<int, TimeSpan> retryBackoff)
+    {
+        _retryBackoff = retryBackoff;
+        return this;
+    }
+
     public Consumer Build() =>
         new Consumer(
             _registry,
@@ -89,5 +96,6 @@ public Consumer Build() =>
             _enableAutoCommit,
             _deadLetterQueue,
             _maxRetries,
-            _deadLetterQueueBypass);
+            _deadLetterQueueBypass,
+            _retryBackoff);
 }

src/Dafda.Tests/Configuration/TestDeadLetterQueueOptions.cs

@@ -66,4 +66,163 @@ public void bypass_predicate_is_not_affected_by_predicates_registered_afterwards
         Assert.False(predicate(new FormatException()));
         Assert.True(sut.BypassPredicate(new FormatException()));
     }
+
+    [Fact]
+    public void has_no_retry_backoff_by_default()
+    {
+        var sut = new DeadLetterQueueOptions("dlq");
+
+        Assert.Null(sut.RetryBackoff);
+    }
+
+    [Fact]
+    public void fixed_retry_backoff_returns_the_same_delay_for_every_attempt()
+    {
+        var sut = new DeadLetterQueueOptions("dlq")
+            .WithRetryBackoff(TimeSpan.FromSeconds(3));
+
+        Assert.Equal(TimeSpan.FromSeconds(3), sut.RetryBackoff(1));
+        Assert.Equal(TimeSpan.FromSeconds(3), sut.RetryBackoff(2));
+        Assert.Equal(TimeSpan.FromSeconds(3), sut.RetryBackoff(7));
+    }
+
+    [Fact]
+    public void exponential_retry_backoff_doubles_by_default()
+    {
+        var sut = new DeadLetterQueueOptions("dlq")
+            .WithExponentialRetryBackoff(TimeSpan.FromSeconds(1));
+
+        Assert.Equal(TimeSpan.FromSeconds(1), sut.RetryBackoff(1));
+        Assert.Equal(TimeSpan.FromSeconds(2), sut.RetryBackoff(2));
+        Assert.Equal(TimeSpan.FromSeconds(4), sut.RetryBackoff(3));
+        Assert.Equal(TimeSpan.FromSeconds(8), sut.RetryBackoff(4));
+    }
+
+    [Fact]
+    public void exponential_retry_backoff_uses_the_supplied_factor()
+    {
+        var sut = new DeadLetterQueueOptions("dlq")
+            .WithExponentialRetryBackoff(TimeSpan.FromSeconds(1), factor: 3);
+
+        Assert.Equal(TimeSpan.FromSeconds(1), sut.RetryBackoff(1));
+        Assert.Equal(TimeSpan.FromSeconds(3), sut.RetryBackoff(2));
+        Assert.Equal(TimeSpan.FromSeconds(9), sut.RetryBackoff(3));
+    }
+
+    [Fact]
+    public void exponential_retry_backoff_is_capped_by_max_delay()
+    {
+        var sut = new DeadLetterQueueOptions("dlq")
+            .WithExponentialRetryBackoff(TimeSpan.FromSeconds(1), maxDelay: TimeSpan.FromSeconds(5));
+
+        Assert.Equal(TimeSpan.FromSeconds(1), sut.RetryBackoff(1));
+        Assert.Equal(TimeSpan.FromSeconds(2), sut.RetryBackoff(2));
+        Assert.Equal(TimeSpan.FromSeconds(4), sut.RetryBackoff(3));
+        Assert.Equal(TimeSpan.FromSeconds(5), sut.RetryBackoff(4));
+        Assert.Equal(TimeSpan.FromSeconds(5), sut.RetryBackoff(100));
+    }
+
+    [Fact]
+    public void throws_when_fixed_retry_backoff_delay_is_negative()
+    {
+        var sut = new DeadLetterQueueOptions("dlq");
+
+        Assert.Throws<InvalidConfigurationException>(() => sut.WithRetryBackoff(TimeSpan.FromSeconds(-1)));
+    }
+
+    [Fact]
+    public void throws_when_exponential_retry_backoff_initial_delay_is_negative()
+    {
+        var sut = new DeadLetterQueueOptions("dlq");
+
+        Assert.Throws<InvalidConfigurationException>(() => sut.WithExponentialRetryBackoff(TimeSpan.FromSeconds(-1)));
+    }
+
+    [Theory]
+    [InlineData(0d)]
+    [InlineData(-1d)]
+    [InlineData(double.NaN)]
+    public void throws_when_exponential_retry_backoff_factor_is_not_positive(double factor)
+    {
+        var sut = new DeadLetterQueueOptions("dlq");
+
+        Assert.Throws<InvalidConfigurationException>(() => sut.WithExponentialRetryBackoff(TimeSpan.FromSeconds(1), factor));
+    }
+
+    [Fact]
+    public void throws_when_exponential_retry_backoff_max_delay_is_negative()
+    {
+        var sut = new DeadLetterQueueOptions("dlq");
+
+        Assert.Throws<InvalidConfigurationException>(
+            () => sut.WithExponentialRetryBackoff(TimeSpan.FromSeconds(1), maxDelay: TimeSpan.FromSeconds(-1)));
+    }
+
+    [Fact]
+    public void throws_when_fixed_retry_backoff_delay_exceeds_the_supported_maximum()
+    {
+        var sut = new DeadLetterQueueOptions("dlq");
+
+        Assert.Throws<InvalidConfigurationException>(
+            () => sut.WithRetryBackoff(DeadLetterQueueOptions.MaxSupportedRetryDelay + TimeSpan.FromMilliseconds(1)));
+    }
+
+    [Fact]
+    public void throws_when_exponential_retry_backoff_initial_delay_exceeds_the_supported_maximum()
+    {
+        var sut = new DeadLetterQueueOptions("dlq");
+
+        Assert.Throws<InvalidConfigurationException>(
+            () => sut.WithExponentialRetryBackoff(DeadLetterQueueOptions.MaxSupportedRetryDelay + TimeSpan.FromMilliseconds(1)));
+    }
+
+    [Fact]
+    public void throws_when_exponential_retry_backoff_max_delay_exceeds_the_supported_maximum()
+    {
+        var sut = new DeadLetterQueueOptions("dlq");
+
+        Assert.Throws<InvalidConfigurationException>(
+            () => sut.WithExponentialRetryBackoff(
+                TimeSpan.FromSeconds(1),
+                maxDelay: DeadLetterQueueOptions.MaxSupportedRetryDelay + TimeSpan.FromMilliseconds(1)));
+    }
+
+    [Fact]
+    public void exponential_retry_backoff_clamps_overflow_to_the_supported_maximum()
+    {
+        var sut = new DeadLetterQueueOptions("dlq")
+            .WithExponentialRetryBackoff(TimeSpan.FromSeconds(1));
+
+        Assert.Equal(DeadLetterQueueOptions.MaxSupportedRetryDelay, sut.RetryBackoff(64));
+        Assert.Equal(DeadLetterQueueOptions.MaxSupportedRetryDelay, sut.RetryBackoff(int.MaxValue));
+    }
+
+    [Fact]
+    public void exponential_retry_backoff_prefers_an_explicit_max_delay_over_the_supported_maximum()
+    {
+        var sut = new DeadLetterQueueOptions("dlq")
+            .WithExponentialRetryBackoff(TimeSpan.FromSeconds(1), maxDelay: TimeSpan.FromSeconds(30));
+
+        Assert.Equal(TimeSpan.FromSeconds(30), sut.RetryBackoff(64));
+        Assert.Equal(TimeSpan.FromSeconds(30), sut.RetryBackoff(int.MaxValue));
+    }
+
+    [Fact]
+    public void exponential_retry_backoff_returns_zero_for_a_zero_initial_delay()
+    {
+        var sut = new DeadLetterQueueOptions("dlq")
+            .WithExponentialRetryBackoff(TimeSpan.Zero, factor: 1000);
+
+        Assert.Equal(TimeSpan.Zero, sut.RetryBackoff(1));
+        Assert.Equal(TimeSpan.Zero, sut.RetryBackoff(64));
+        Assert.Equal(TimeSpan.Zero, sut.RetryBackoff(int.MaxValue));
+    }
+
+    [Fact]
+    public void throws_when_max_retries_is_negative()
+    {
+        var sut = new DeadLetterQueueOptions("dlq");
+
+        Assert.Throws<InvalidConfigurationException>(() => sut.WithMaxRetries(-1));
+    }
 }

src/Dafda.Tests/Consuming/TestConsumer.cs

@@ -518,12 +518,115 @@ public async Task dead_letters_exceptions_that_do_not_match_the_bypass()
         Assert.Equal(1, deadLetterQueueSpy.SendCount);
     }
 
+    [Fact]
+    public async Task does_not_delay_between_retries_when_no_backoff_is_configured()
+    {
+        var handlerInvocations = 0;
+        var handler = new MessageHandlerSpy<FooMessage>(() =>
+        {
+            handlerInvocations++;
+            throw new InvalidOperationException("boom");
+        });
+
+        var deadLetterQueueSpy = new DeadLetterQueueSpy();
+
+        var sut = BuildConsumerWithHandler(
+            handler,
+            deadLetterQueue: deadLetterQueueSpy,
+            maxRetries: 2,
+            retryBackoff: null);
+
+        await sut.ConsumeSingle(CancellationToken.None);
+
+        Assert.Equal(3, handlerInvocations);
+        Assert.Equal(1, deadLetterQueueSpy.SendCount);
+    }
+
+    [Fact]
+    public async Task applies_retry_backoff_for_each_retry_attempt_and_dead_letters_after_max_retries()
+    {
+        var handlerInvocations = 0;
+        var handler = new MessageHandlerSpy<FooMessage>(() =>
+        {
+            handlerInvocations++;
+            throw new InvalidOperationException("boom");
+        });
+
+        var deadLetterQueueSpy = new DeadLetterQueueSpy();
+        var backoffAttempts = new List<int>();
+
+        var sut = BuildConsumerWithHandler(
+            handler,
+            deadLetterQueue: deadLetterQueueSpy,
+            maxRetries: 3,
+            retryBackoff: attempt =>
+            {
+                backoffAttempts.Add(attempt);
+                return TimeSpan.FromMilliseconds(1);
+            });
+
+        await sut.ConsumeSingle(CancellationToken.None);
+
+        Assert.Equal(new[] { 1, 2, 3 }, backoffAttempts);
+        Assert.Equal(4, handlerInvocations);
+        Assert.Equal(1, deadLetterQueueSpy.SendCount);
+    }
+
+    [Fact]
+    public async Task does_not_apply_retry_backoff_when_handler_succeeds()
+    {
+        var handler = new MessageHandlerSpy<FooMessage>(() => { });
+
+        var deadLetterQueueSpy = new DeadLetterQueueSpy();
+        var backoffInvocations = 0;
+
+        var sut = BuildConsumerWithHandler(
+            handler,
+            deadLetterQueue: deadLetterQueueSpy,
+            maxRetries: 3,
+            retryBackoff: _ =>
+            {
+                backoffInvocations++;
+                return TimeSpan.Zero;
+            });
+
+        await sut.ConsumeSingle(CancellationToken.None);
+
+        Assert.Equal(0, backoffInvocations);
+        Assert.Equal(0, deadLetterQueueSpy.SendCount);
+    }
+
+    [Fact]
+    public async Task does_not_dead_letter_when_cancelled_during_retry_backoff()
+    {
+        using var cts = new CancellationTokenSource();
+
+        var handler = new MessageHandlerSpy<FooMessage>(() => throw new InvalidOperationException("boom"));
+
+        var deadLetterQueueSpy = new DeadLetterQueueSpy();
+
+        var sut = BuildConsumerWithHandler(
+            handler,
+            deadLetterQueue: deadLetterQueueSpy,
+            maxRetries: 3,
+            retryBackoff: _ =>
+            {
+                cts.Cancel();
+                return TimeSpan.FromMinutes(5);
+            });
+
+        await Assert.ThrowsAnyAsync<OperationCanceledException>(() => sut.ConsumeSingle(cts.Token));
+
+        Assert.Equal(0, deadLetterQueueSpy.SendCount);
+    }
+
     private static Consumer BuildConsumerWithHandler(
         IMessageHandler<FooMessage> handler,
         Func<CancellationToken, Task> onCommit = null,
         IDeadLetterQueue deadLetterQueue = null,
         int maxRetries = 0,
-        Func<Exception, bool> deadLetterQueueBypass = null)
+        Func<Exception, bool> deadLetterQueueBypass = null,
+        Func<int, TimeSpan> retryBackoff = null)
     {
         var registration = new MessageRegistrationBuilder()
             .WithHandlerInstanceType(handler.GetType())
@@ -546,7 +649,8 @@ private static Consumer BuildConsumerWithHandler(
             .WithUnitOfWork(new UnitOfWorkStub(handler))
             .WithMessageHandlerRegistry(registry)
             .WithMaxRetries(maxRetries)
-            .WithDeadLetterQueueBypass(deadLetterQueueBypass);
+            .WithDeadLetterQueueBypass(deadLetterQueueBypass)
+            .WithRetryBackoff(retryBackoff);
 
         if (deadLetterQueue != null)
         {

src/Dafda/Configuration/ConsumerConfiguration.cs

@@ -14,7 +14,8 @@ internal class ConsumerConfiguration(
     IConsumerErrorHandler consumerErrorHandler,
     Func<IServiceProvider, IDeadLetterQueue> deadLetterQueueFactory,
     int maxRetries,
-    Func<Exception, bool> deadLetterQueueBypass)
+    Func<Exception, bool> deadLetterQueueBypass,
+    Func<int, TimeSpan> retryBackoff = null)
     : ConsumerConfigurationBase(configuration, factories.UnitOfWorkFactory, consumerErrorHandler)
 {
     public ConsumerConfigurationFactories Factories { get; } = factories;
@@ -23,4 +24,5 @@ internal class ConsumerConfiguration(
     public Func<IServiceProvider, IDeadLetterQueue> DeadLetterQueueFactory { get; } = deadLetterQueueFactory;
     public int MaxRetries { get; } = maxRetries;
     public Func<Exception, bool> DeadLetterQueueBypass { get; } = deadLetterQueueBypass;
+    public Func<int, TimeSpan> RetryBackoff { get; } = retryBackoff;
 }

src/Dafda/Configuration/ConsumerConfigurationBuilder.cs

@@ -208,6 +208,7 @@ internal ConsumerConfiguration Build()
         var deadLetterQueueFactory = BuildDeadLetterQueueFactory(configurations);
         var maxRetries = _deadLetterQueueOptions?.MaxRetries ?? 0;
         var deadLetterQueueBypass = _deadLetterQueueOptions?.BypassPredicate;
+        var retryBackoff = _deadLetterQueueOptions?.RetryBackoff;
 
         return new ConsumerConfiguration(
             configuration: configurations,
@@ -217,7 +218,8 @@ internal ConsumerConfiguration Build()
             consumerErrorHandler: _consumerErrorHandler,
             deadLetterQueueFactory: deadLetterQueueFactory,
             maxRetries: maxRetries,
-            deadLetterQueueBypass: deadLetterQueueBypass);
+            deadLetterQueueBypass: deadLetterQueueBypass,
+            retryBackoff: retryBackoff);
     }
 
     private Func<IServiceProvider, IDeadLetterQueue> BuildDeadLetterQueueFactory(IDictionary<string, string> configurations)

src/Dafda/Configuration/ConsumerServiceCollectionExtensions.cs

@@ -47,7 +47,8 @@ public static void AddConsumer(this IServiceCollection services, Action<Consumer
                 configuration.EnableAutoCommit,
                 configuration.DeadLetterQueueFactory(provider),
                 configuration.MaxRetries,
-                configuration.DeadLetterQueueBypass
+                configuration.DeadLetterQueueBypass,
+                configuration.RetryBackoff
             ),
             configuration.GroupId,
             configuration.ConsumerErrorHandler
@@ -84,7 +85,8 @@ public static void AddConsumer(this IServiceCollection services, Func<IServicePr
                     configuration.EnableAutoCommit,
                     configuration.DeadLetterQueueFactory(provider),
                     configuration.MaxRetries,
-                    configuration.DeadLetterQueueBypass
+                    configuration.DeadLetterQueueBypass,
+                    configuration.RetryBackoff
                 ),
                 configuration.GroupId,
                 configuration.ConsumerErrorHandler