From 66bd6206c8eacd237e67307ec7fb648672ea5d9a Mon Sep 17 00:00:00 2001 From: Nabi Sobhi Date: Sun, 30 Aug 2026 20:34:23 +0200 Subject: [PATCH 1/4] Allow specific exception types to bypass the dead letter queue Adds a configurable bypass predicate so fatal/systemic exceptions propagate and crash the consumer instead of being retried or dead-lettered. This prevents a systemic outage (e.g. database down) from silently draining an entire topic into the dead letter queue. New fluent API on DeadLetterQueueOptions: .BypassFor() .BypassWhen(ex => ...) Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- src/Dafda.Tests/Builders/ConsumerBuilder.cs | 11 ++++- src/Dafda.Tests/Consuming/TestConsumer.cs | 48 ++++++++++++++++++- .../Configuration/ConsumerConfiguration.cs | 4 +- .../ConsumerConfigurationBuilder.cs | 4 +- .../ConsumerServiceCollectionExtensions.cs | 6 ++- .../Configuration/DeadLetterQueueOptions.cs | 46 ++++++++++++++++++ src/Dafda/Consuming/Consumer.cs | 10 +++- 7 files changed, 120 insertions(+), 9 deletions(-) diff --git a/src/Dafda.Tests/Builders/ConsumerBuilder.cs b/src/Dafda.Tests/Builders/ConsumerBuilder.cs index d0861fd..bbbc094 100644 --- a/src/Dafda.Tests/Builders/ConsumerBuilder.cs +++ b/src/Dafda.Tests/Builders/ConsumerBuilder.cs @@ -1,5 +1,6 @@ namespace Dafda.Tests.Builders; +using System; using Dafda.Consuming; using Dafda.Consuming.MessageFilters; using TestDoubles; @@ -16,6 +17,7 @@ internal class ConsumerBuilder private MessageFilter _messageFilter = MessageFilter.Default; private IDeadLetterQueue _deadLetterQueue = NullDeadLetterQueue.Instance; private int _maxRetries; + private Func _deadLetterQueueBypass; public ConsumerBuilder WithUnitOfWork(IHandlerUnitOfWork unitOfWork) { @@ -70,6 +72,12 @@ public ConsumerBuilder WithMaxRetries(int maxRetries) return this; } + public ConsumerBuilder WithDeadLetterQueueBypass(Func deadLetterQueueBypass) + { + _deadLetterQueueBypass = deadLetterQueueBypass; + return this; + } + public Consumer Build() => new Consumer( _registry, @@ -80,5 +88,6 @@ public Consumer Build() => _messageHandlerExecutionStrategy, _enableAutoCommit, _deadLetterQueue, - _maxRetries); + _maxRetries, + _deadLetterQueueBypass); } \ No newline at end of file diff --git a/src/Dafda.Tests/Consuming/TestConsumer.cs b/src/Dafda.Tests/Consuming/TestConsumer.cs index ad9df7c..cf9f765 100644 --- a/src/Dafda.Tests/Consuming/TestConsumer.cs +++ b/src/Dafda.Tests/Consuming/TestConsumer.cs @@ -469,11 +469,54 @@ public void disposing_consumer_disposes_the_dead_letter_queue() Assert.Equal(1, deadLetterQueueSpy.DisposedCount); } + [Fact] + public async Task propagates_exception_and_bypasses_dead_letter_queue_for_bypassed_exception_type() + { + var handlerInvocations = 0; + var handler = new MessageHandlerSpy(() => + { + handlerInvocations++; + throw new InvalidOperationException("fatal"); + }); + + var deadLetterQueueSpy = new DeadLetterQueueSpy(); + + var sut = BuildConsumerWithHandler( + handler, + deadLetterQueue: deadLetterQueueSpy, + maxRetries: 3, + deadLetterQueueBypass: exception => exception is InvalidOperationException); + + await Assert.ThrowsAsync( + () => sut.ConsumeSingle(CancellationToken.None)); + + Assert.Equal(1, handlerInvocations); + Assert.Equal(0, deadLetterQueueSpy.SendCount); + } + + [Fact] + public async Task dead_letters_exceptions_that_do_not_match_the_bypass() + { + var handler = new MessageHandlerSpy(() => throw new InvalidOperationException("boom")); + + var deadLetterQueueSpy = new DeadLetterQueueSpy(); + + var sut = BuildConsumerWithHandler( + handler, + deadLetterQueue: deadLetterQueueSpy, + deadLetterQueueBypass: exception => exception is FormatException); + + await sut.ConsumeSingle(CancellationToken.None); + + Assert.Equal(1, deadLetterQueueSpy.SendCount); + } + private static Consumer BuildConsumerWithHandler( IMessageHandler handler, Func onCommit = null, IDeadLetterQueue deadLetterQueue = null, - int maxRetries = 0) + int maxRetries = 0, + Func deadLetterQueueBypass = null) { var registration = new MessageRegistrationBuilder() .WithHandlerInstanceType(handler.GetType()) @@ -495,7 +538,8 @@ private static Consumer BuildConsumerWithHandler( .WithConsumerScopeFactory(new ConsumerScopeFactoryStub(new ConsumerScopeStub(messageResult))) .WithUnitOfWork(new UnitOfWorkStub(handler)) .WithMessageHandlerRegistry(registry) - .WithMaxRetries(maxRetries); + .WithMaxRetries(maxRetries) + .WithDeadLetterQueueBypass(deadLetterQueueBypass); if (deadLetterQueue != null) { diff --git a/src/Dafda/Configuration/ConsumerConfiguration.cs b/src/Dafda/Configuration/ConsumerConfiguration.cs index 0859c41..b7ac7bc 100644 --- a/src/Dafda/Configuration/ConsumerConfiguration.cs +++ b/src/Dafda/Configuration/ConsumerConfiguration.cs @@ -13,7 +13,8 @@ internal class ConsumerConfiguration( MessageFilter messageFilter, IConsumerErrorHandler consumerErrorHandler, Func deadLetterQueueFactory, - int maxRetries) + int maxRetries, + Func deadLetterQueueBypass) : ConsumerConfigurationBase(configuration, factories.UnitOfWorkFactory, consumerErrorHandler) { public ConsumerConfigurationFactories Factories { get; } = factories; @@ -21,4 +22,5 @@ internal class ConsumerConfiguration( public MessageFilter MessageFilter { get; } = messageFilter; public Func DeadLetterQueueFactory { get; } = deadLetterQueueFactory; public int MaxRetries { get; } = maxRetries; + public Func DeadLetterQueueBypass { get; } = deadLetterQueueBypass; } \ No newline at end of file diff --git a/src/Dafda/Configuration/ConsumerConfigurationBuilder.cs b/src/Dafda/Configuration/ConsumerConfigurationBuilder.cs index b6d0bc2..a10ab1b 100644 --- a/src/Dafda/Configuration/ConsumerConfigurationBuilder.cs +++ b/src/Dafda/Configuration/ConsumerConfigurationBuilder.cs @@ -207,6 +207,7 @@ internal ConsumerConfiguration Build() var deadLetterQueueFactory = BuildDeadLetterQueueFactory(configurations); var maxRetries = _deadLetterQueueOptions?.MaxRetries ?? 0; + var deadLetterQueueBypass = _deadLetterQueueOptions?.BypassPredicate; return new ConsumerConfiguration( configuration: configurations, @@ -215,7 +216,8 @@ internal ConsumerConfiguration Build() messageFilter: _messageFilter, consumerErrorHandler: _consumerErrorHandler, deadLetterQueueFactory: deadLetterQueueFactory, - maxRetries: maxRetries); + maxRetries: maxRetries, + deadLetterQueueBypass: deadLetterQueueBypass); } private Func BuildDeadLetterQueueFactory(IDictionary configurations) diff --git a/src/Dafda/Configuration/ConsumerServiceCollectionExtensions.cs b/src/Dafda/Configuration/ConsumerServiceCollectionExtensions.cs index 63e0dfb..c41a8a9 100644 --- a/src/Dafda/Configuration/ConsumerServiceCollectionExtensions.cs +++ b/src/Dafda/Configuration/ConsumerServiceCollectionExtensions.cs @@ -46,7 +46,8 @@ public static void AddConsumer(this IServiceCollection services, Action /// Fluent options for configuring a dead letter queue on a consumer. /// Returned by . /// public sealed class DeadLetterQueueOptions { + private readonly List> _bypassPredicates = new(); + internal DeadLetterQueueOptions(string topicName) { TopicName = topicName; @@ -38,4 +44,44 @@ public DeadLetterQueueOptions WithMaxRetries(int maxRetries) MaxRetries = maxRetries; return this; } + + /// + /// A predicate matching exceptions that should bypass the dead letter queue. + /// When an exception matches, it is rethrown (crashing the consumer) instead + /// of being retried or forwarded to the dead letter queue. Returns null + /// when no bypass has been configured. + /// + internal Func BypassPredicate => + _bypassPredicates.Count == 0 + ? null + : exception => _bypassPredicates.Any(predicate => predicate(exception)); + + /// + /// Bypass the dead letter queue for the specified exception type (and any + /// derived types). When a message handler throws a matching exception, it is + /// rethrown so the consumer crashes instead of dead-lettering the message. + /// + /// The exception type to bypass the dead letter queue for. + public DeadLetterQueueOptions BypassFor() where TException : Exception + { + _bypassPredicates.Add(exception => exception is TException); + return this; + } + + /// + /// Bypass the dead letter queue for exceptions matching the supplied + /// . When it returns true, the exception is + /// rethrown so the consumer crashes instead of dead-lettering the message. + /// + /// Evaluates a thrown exception and returns true to bypass the dead letter queue. + public DeadLetterQueueOptions BypassWhen(Func predicate) + { + if (predicate == null) + { + throw new InvalidConfigurationException("The dead letter queue bypass predicate cannot be null."); + } + + _bypassPredicates.Add(predicate); + return this; + } } \ No newline at end of file diff --git a/src/Dafda/Consuming/Consumer.cs b/src/Dafda/Consuming/Consumer.cs index 0ca3943..166453b 100644 --- a/src/Dafda/Consuming/Consumer.cs +++ b/src/Dafda/Consuming/Consumer.cs @@ -16,7 +16,8 @@ internal class Consumer( IMessageHandlerExecutionStrategy messageHandlerExecutionStrategy, bool isAutoCommitEnabled = false, IDeadLetterQueue deadLetterQueue = null, - int maxRetries = 0) + int maxRetries = 0, + Func deadLetterQueueBypass = null) : IConsumer, IDisposable { private readonly LocalMessageDispatcher _localMessageDispatcher = new( @@ -70,7 +71,7 @@ private async Task Dispatch(MessageResult messageResult, CancellationToken cance await _localMessageDispatcher.Dispatch(messageResult, cancellationToken); return; } - catch (Exception exception) when (deadLetterQueueEnabled && !cancellationToken.IsCancellationRequested) + catch (Exception exception) when (deadLetterQueueEnabled && !cancellationToken.IsCancellationRequested && !ShouldBypassDeadLetterQueue(exception)) { if (attempt++ < maxRetries) { @@ -83,6 +84,11 @@ private async Task Dispatch(MessageResult messageResult, CancellationToken cance } } + private bool ShouldBypassDeadLetterQueue(Exception exception) + { + return deadLetterQueueBypass != null && deadLetterQueueBypass(exception); + } + public void Dispose() { (_deadLetterQueue as IDisposable)?.Dispose(); From b6a211e5f77f6d81ad75a870792a32c090ef6fe9 Mon Sep 17 00:00:00 2001 From: Nabi Sobhi Date: Sun, 30 Aug 2026 20:45:35 +0200 Subject: [PATCH 2/4] Add configurable retry backoff to dead letter queue retry loop Retries previously happened in a tight loop with no delay, hammering a failing downstream within microseconds. DeadLetterQueueOptions now exposes WithRetryBackoff(TimeSpan) for a fixed delay and WithExponentialRetryBackoff (TimeSpan, double, TimeSpan?) for an exponentially increasing delay, capped by an optional maxDelay. The resolved policy is threaded to Consumer as a Func mapping attempt number (first retry = 1) to delay; null means no delay, preserving the existing default behavior. The delay honors the cancellation token, so cancelling during a backoff propagates OperationCanceledException instead of dead-lettering. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- src/Dafda.Tests/Builders/ConsumerBuilder.cs | 10 +- .../TestDeadLetterQueueOptions.cs | 106 +++++++++++++++++ src/Dafda.Tests/Consuming/TestConsumer.cs | 108 +++++++++++++++++- .../Configuration/ConsumerConfiguration.cs | 4 +- .../ConsumerConfigurationBuilder.cs | 4 +- .../ConsumerServiceCollectionExtensions.cs | 6 +- .../Configuration/DeadLetterQueueOptions.cs | 69 +++++++++++ src/Dafda/Consuming/Consumer.cs | 18 ++- 8 files changed, 316 insertions(+), 9 deletions(-) create mode 100644 src/Dafda.Tests/Configuration/TestDeadLetterQueueOptions.cs diff --git a/src/Dafda.Tests/Builders/ConsumerBuilder.cs b/src/Dafda.Tests/Builders/ConsumerBuilder.cs index bbbc094..bedc6fe 100644 --- a/src/Dafda.Tests/Builders/ConsumerBuilder.cs +++ b/src/Dafda.Tests/Builders/ConsumerBuilder.cs @@ -18,6 +18,7 @@ internal class ConsumerBuilder private IDeadLetterQueue _deadLetterQueue = NullDeadLetterQueue.Instance; private int _maxRetries; private Func _deadLetterQueueBypass; + private Func _retryBackoff; public ConsumerBuilder WithUnitOfWork(IHandlerUnitOfWork unitOfWork) { @@ -78,6 +79,12 @@ public ConsumerBuilder WithDeadLetterQueueBypass(Func deadLette return this; } + public ConsumerBuilder WithRetryBackoff(Func retryBackoff) + { + _retryBackoff = retryBackoff; + return this; + } + public Consumer Build() => new Consumer( _registry, @@ -89,5 +96,6 @@ public Consumer Build() => _enableAutoCommit, _deadLetterQueue, _maxRetries, - _deadLetterQueueBypass); + _deadLetterQueueBypass, + _retryBackoff); } \ No newline at end of file diff --git a/src/Dafda.Tests/Configuration/TestDeadLetterQueueOptions.cs b/src/Dafda.Tests/Configuration/TestDeadLetterQueueOptions.cs new file mode 100644 index 0000000..78837a1 --- /dev/null +++ b/src/Dafda.Tests/Configuration/TestDeadLetterQueueOptions.cs @@ -0,0 +1,106 @@ +namespace Dafda.Tests.Configuration; + +using System; +using Dafda.Configuration; +using Xunit; + +public class TestDeadLetterQueueOptions +{ + [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(() => sut.WithRetryBackoff(TimeSpan.FromSeconds(-1))); + } + + [Fact] + public void throws_when_exponential_retry_backoff_initial_delay_is_negative() + { + var sut = new DeadLetterQueueOptions("dlq"); + + Assert.Throws(() => sut.WithExponentialRetryBackoff(TimeSpan.FromSeconds(-1))); + } + + [Theory] + [InlineData(0d)] + [InlineData(-1d)] + public void throws_when_exponential_retry_backoff_factor_is_not_positive(double factor) + { + var sut = new DeadLetterQueueOptions("dlq"); + + Assert.Throws(() => 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( + () => sut.WithExponentialRetryBackoff(TimeSpan.FromSeconds(1), maxDelay: TimeSpan.FromSeconds(-1))); + } + + [Fact] + public void throws_when_max_retries_is_negative() + { + var sut = new DeadLetterQueueOptions("dlq"); + + Assert.Throws(() => sut.WithMaxRetries(-1)); + } +} diff --git a/src/Dafda.Tests/Consuming/TestConsumer.cs b/src/Dafda.Tests/Consuming/TestConsumer.cs index cf9f765..91cc6c4 100644 --- a/src/Dafda.Tests/Consuming/TestConsumer.cs +++ b/src/Dafda.Tests/Consuming/TestConsumer.cs @@ -511,12 +511,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(() => + { + 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(() => + { + handlerInvocations++; + throw new InvalidOperationException("boom"); + }); + + var deadLetterQueueSpy = new DeadLetterQueueSpy(); + var backoffAttempts = new List(); + + 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(() => { }); + + 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(() => 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(() => sut.ConsumeSingle(cts.Token)); + + Assert.Equal(0, deadLetterQueueSpy.SendCount); + } + private static Consumer BuildConsumerWithHandler( IMessageHandler handler, Func onCommit = null, IDeadLetterQueue deadLetterQueue = null, int maxRetries = 0, - Func deadLetterQueueBypass = null) + Func deadLetterQueueBypass = null, + Func retryBackoff = null) { var registration = new MessageRegistrationBuilder() .WithHandlerInstanceType(handler.GetType()) @@ -539,7 +642,8 @@ private static Consumer BuildConsumerWithHandler( .WithUnitOfWork(new UnitOfWorkStub(handler)) .WithMessageHandlerRegistry(registry) .WithMaxRetries(maxRetries) - .WithDeadLetterQueueBypass(deadLetterQueueBypass); + .WithDeadLetterQueueBypass(deadLetterQueueBypass) + .WithRetryBackoff(retryBackoff); if (deadLetterQueue != null) { diff --git a/src/Dafda/Configuration/ConsumerConfiguration.cs b/src/Dafda/Configuration/ConsumerConfiguration.cs index b7ac7bc..6ed72be 100644 --- a/src/Dafda/Configuration/ConsumerConfiguration.cs +++ b/src/Dafda/Configuration/ConsumerConfiguration.cs @@ -14,7 +14,8 @@ internal class ConsumerConfiguration( IConsumerErrorHandler consumerErrorHandler, Func deadLetterQueueFactory, int maxRetries, - Func deadLetterQueueBypass) + Func deadLetterQueueBypass, + Func retryBackoff = null) : ConsumerConfigurationBase(configuration, factories.UnitOfWorkFactory, consumerErrorHandler) { public ConsumerConfigurationFactories Factories { get; } = factories; @@ -23,4 +24,5 @@ internal class ConsumerConfiguration( public Func DeadLetterQueueFactory { get; } = deadLetterQueueFactory; public int MaxRetries { get; } = maxRetries; public Func DeadLetterQueueBypass { get; } = deadLetterQueueBypass; + public Func RetryBackoff { get; } = retryBackoff; } \ No newline at end of file diff --git a/src/Dafda/Configuration/ConsumerConfigurationBuilder.cs b/src/Dafda/Configuration/ConsumerConfigurationBuilder.cs index a10ab1b..25a5d95 100644 --- a/src/Dafda/Configuration/ConsumerConfigurationBuilder.cs +++ b/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 BuildDeadLetterQueueFactory(IDictionary configurations) diff --git a/src/Dafda/Configuration/ConsumerServiceCollectionExtensions.cs b/src/Dafda/Configuration/ConsumerServiceCollectionExtensions.cs index c41a8a9..f700ad3 100644 --- a/src/Dafda/Configuration/ConsumerServiceCollectionExtensions.cs +++ b/src/Dafda/Configuration/ConsumerServiceCollectionExtensions.cs @@ -47,7 +47,8 @@ public static void AddConsumer(this IServiceCollection services, Action + /// Maps a retry attempt number (the first retry is attempt 1) to the delay + /// awaited before that attempt is made. Returns null when no backoff has been + /// configured, in which case retries happen without any delay. + /// + internal Func RetryBackoff { get; private set; } + + /// + /// Wait a fixed before each retry attempt. + /// + /// The delay awaited before every retry attempt. Must be zero or greater. + public DeadLetterQueueOptions WithRetryBackoff(TimeSpan delay) + { + if (delay < TimeSpan.Zero) + { + throw new InvalidConfigurationException("The retry backoff delay for a dead letter queue cannot be negative."); + } + + RetryBackoff = _ => delay; + return this; + } + + /// + /// Wait an exponentially increasing delay before each retry attempt. The delay before + /// retry attempt n (the first retry being attempt 1) is + /// multiplied by raised to the + /// power of n - 1, optionally capped by . + /// + /// The delay awaited before the first retry attempt. Must be zero or greater. + /// The multiplier applied for each subsequent attempt. Must be greater than zero. + /// An optional upper bound for the computed delay. Must be zero or greater when supplied. + public DeadLetterQueueOptions WithExponentialRetryBackoff(TimeSpan initialDelay, double factor = 2.0, TimeSpan? maxDelay = null) + { + if (initialDelay < TimeSpan.Zero) + { + throw new InvalidConfigurationException("The retry backoff delay for a dead letter queue cannot be negative."); + } + + if (factor <= 0) + { + throw new InvalidConfigurationException("The retry backoff factor for a dead letter queue must be greater than zero."); + } + + if (maxDelay.HasValue && maxDelay.Value < TimeSpan.Zero) + { + throw new InvalidConfigurationException("The maximum retry backoff delay for a dead letter queue cannot be negative."); + } + + RetryBackoff = attempt => CalculateExponentialDelay(initialDelay, factor, maxDelay, attempt); + return this; + } + + private static TimeSpan CalculateExponentialDelay(TimeSpan initialDelay, double factor, TimeSpan? maxDelay, int attempt) + { + var exponent = attempt < 1 ? 0 : attempt - 1; + var ticks = initialDelay.Ticks * Math.Pow(factor, exponent); + + var delay = ticks >= TimeSpan.MaxValue.Ticks + ? TimeSpan.MaxValue + : TimeSpan.FromTicks((long)ticks); + + if (maxDelay.HasValue && delay > maxDelay.Value) + { + return maxDelay.Value; + } + + return delay; + } + /// /// A predicate matching exceptions that should bypass the dead letter queue. /// When an exception matches, it is rethrown (crashing the consumer) instead diff --git a/src/Dafda/Consuming/Consumer.cs b/src/Dafda/Consuming/Consumer.cs index 166453b..ff8acc5 100644 --- a/src/Dafda/Consuming/Consumer.cs +++ b/src/Dafda/Consuming/Consumer.cs @@ -17,7 +17,8 @@ internal class Consumer( bool isAutoCommitEnabled = false, IDeadLetterQueue deadLetterQueue = null, int maxRetries = 0, - Func deadLetterQueueBypass = null) + Func deadLetterQueueBypass = null, + Func retryBackoff = null) : IConsumer, IDisposable { private readonly LocalMessageDispatcher _localMessageDispatcher = new( @@ -73,8 +74,16 @@ private async Task Dispatch(MessageResult messageResult, CancellationToken cance } catch (Exception exception) when (deadLetterQueueEnabled && !cancellationToken.IsCancellationRequested && !ShouldBypassDeadLetterQueue(exception)) { - if (attempt++ < maxRetries) + if (attempt < maxRetries) { + attempt++; + + var delay = GetRetryDelay(attempt); + if (delay > TimeSpan.Zero) + { + await Task.Delay(delay, cancellationToken); + } + continue; } @@ -84,6 +93,11 @@ private async Task Dispatch(MessageResult messageResult, CancellationToken cance } } + private TimeSpan GetRetryDelay(int attempt) + { + return retryBackoff == null ? TimeSpan.Zero : retryBackoff(attempt); + } + private bool ShouldBypassDeadLetterQueue(Exception exception) { return deadLetterQueueBypass != null && deadLetterQueueBypass(exception); From 3bfca4ac51ea2a34f7603a6cbba37f0907d63d29 Mon Sep 17 00:00:00 2001 From: Nabi Sobhi Date: Sun, 30 Aug 2026 20:54:38 +0200 Subject: [PATCH 3/4] Clamp dead letter queue retry backoff to the supported timer limit Task.Delay throws ArgumentOutOfRangeException past the platform timer limit, and because the delay is awaited inside the catch block in Consumer.Dispatch that would crash the consumer with a confusing argument error instead of backing off. Exponential backoff previously clamped overflow to TimeSpan.MaxValue, which is well beyond that limit. Introduce DeadLetterQueueOptions.MaxSupportedRetryDelay (int.MaxValue ms), clamp the computed exponential delay to it after applying an explicit maxDelay cap, and reject configured delays above it with InvalidConfigurationException. Also guard against NaN, which a zero initial delay combined with an overflowing factor produced and which previously fell through to a large negative TimeSpan. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- .../TestDeadLetterQueueOptions.cs | 60 ++++++++++++++++++ .../Configuration/DeadLetterQueueOptions.cs | 62 ++++++++++++++----- 2 files changed, 105 insertions(+), 17 deletions(-) diff --git a/src/Dafda.Tests/Configuration/TestDeadLetterQueueOptions.cs b/src/Dafda.Tests/Configuration/TestDeadLetterQueueOptions.cs index 78837a1..db173b0 100644 --- a/src/Dafda.Tests/Configuration/TestDeadLetterQueueOptions.cs +++ b/src/Dafda.Tests/Configuration/TestDeadLetterQueueOptions.cs @@ -96,6 +96,66 @@ public void throws_when_exponential_retry_backoff_max_delay_is_negative() () => 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( + () => 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( + () => 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( + () => 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() { diff --git a/src/Dafda/Configuration/DeadLetterQueueOptions.cs b/src/Dafda/Configuration/DeadLetterQueueOptions.cs index bc3f52f..987fcce 100644 --- a/src/Dafda/Configuration/DeadLetterQueueOptions.cs +++ b/src/Dafda/Configuration/DeadLetterQueueOptions.cs @@ -52,16 +52,22 @@ public DeadLetterQueueOptions WithMaxRetries(int maxRetries) /// internal Func RetryBackoff { get; private set; } + /// + /// The largest delay that can be awaited between retry attempts, limited by the + /// underlying platform timer. + /// + internal static readonly TimeSpan MaxSupportedRetryDelay = TimeSpan.FromMilliseconds(int.MaxValue); + /// /// Wait a fixed before each retry attempt. /// - /// The delay awaited before every retry attempt. Must be zero or greater. + /// + /// The delay awaited before every retry attempt. Must be zero or greater, and cannot + /// exceed . + /// public DeadLetterQueueOptions WithRetryBackoff(TimeSpan delay) { - if (delay < TimeSpan.Zero) - { - throw new InvalidConfigurationException("The retry backoff delay for a dead letter queue cannot be negative."); - } + EnsureDelayIsSupported(delay, "The retry backoff delay for a dead letter queue"); RetryBackoff = _ => delay; return this; @@ -73,45 +79,67 @@ public DeadLetterQueueOptions WithRetryBackoff(TimeSpan delay) /// multiplied by raised to the /// power of n - 1, optionally capped by . /// - /// The delay awaited before the first retry attempt. Must be zero or greater. + /// + /// The delay awaited before the first retry attempt. Must be zero or greater, and cannot + /// exceed . + /// /// The multiplier applied for each subsequent attempt. Must be greater than zero. - /// An optional upper bound for the computed delay. Must be zero or greater when supplied. + /// + /// An optional upper bound for the computed delay. Must be zero or greater and cannot exceed + /// when supplied. When omitted, the computed delay is + /// still clamped to . + /// public DeadLetterQueueOptions WithExponentialRetryBackoff(TimeSpan initialDelay, double factor = 2.0, TimeSpan? maxDelay = null) { - if (initialDelay < TimeSpan.Zero) - { - throw new InvalidConfigurationException("The retry backoff delay for a dead letter queue cannot be negative."); - } + EnsureDelayIsSupported(initialDelay, "The retry backoff delay for a dead letter queue"); if (factor <= 0) { throw new InvalidConfigurationException("The retry backoff factor for a dead letter queue must be greater than zero."); } - if (maxDelay.HasValue && maxDelay.Value < TimeSpan.Zero) + if (maxDelay.HasValue) { - throw new InvalidConfigurationException("The maximum retry backoff delay for a dead letter queue cannot be negative."); + EnsureDelayIsSupported(maxDelay.Value, "The maximum retry backoff delay for a dead letter queue"); } RetryBackoff = attempt => CalculateExponentialDelay(initialDelay, factor, maxDelay, attempt); return this; } + private static void EnsureDelayIsSupported(TimeSpan delay, string subject) + { + if (delay < TimeSpan.Zero) + { + throw new InvalidConfigurationException($"{subject} cannot be negative."); + } + + if (delay > MaxSupportedRetryDelay) + { + throw new InvalidConfigurationException($"{subject} cannot exceed {MaxSupportedRetryDelay}."); + } + } + private static TimeSpan CalculateExponentialDelay(TimeSpan initialDelay, double factor, TimeSpan? maxDelay, int attempt) { var exponent = attempt < 1 ? 0 : attempt - 1; var ticks = initialDelay.Ticks * Math.Pow(factor, exponent); - var delay = ticks >= TimeSpan.MaxValue.Ticks - ? TimeSpan.MaxValue + if (double.IsNaN(ticks) || ticks <= 0) + { + return TimeSpan.Zero; + } + + var delay = ticks >= MaxSupportedRetryDelay.Ticks + ? MaxSupportedRetryDelay : TimeSpan.FromTicks((long)ticks); if (maxDelay.HasValue && delay > maxDelay.Value) { - return maxDelay.Value; + delay = maxDelay.Value; } - return delay; + return delay > MaxSupportedRetryDelay ? MaxSupportedRetryDelay : delay; } /// From 7787faa61417416b8ef2d6c6b8dd8e040686c08a Mon Sep 17 00:00:00 2001 From: Nabi Sobhi Date: Mon, 14 Sep 2026 09:49:32 +0200 Subject: [PATCH 4/4] Reject a NaN retry backoff factor during configuration NaN passed the factor <= 0 guard because every comparison with NaN is false, so an invalid factor was accepted and then silently degraded the configured backoff at runtime rather than being reported as a misconfiguration. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- src/Dafda.Tests/Configuration/TestDeadLetterQueueOptions.cs | 1 + src/Dafda/Configuration/DeadLetterQueueOptions.cs | 2 +- 2 files changed, 2 insertions(+), 1 deletion(-) diff --git a/src/Dafda.Tests/Configuration/TestDeadLetterQueueOptions.cs b/src/Dafda.Tests/Configuration/TestDeadLetterQueueOptions.cs index d2a7faa..ab74c71 100644 --- a/src/Dafda.Tests/Configuration/TestDeadLetterQueueOptions.cs +++ b/src/Dafda.Tests/Configuration/TestDeadLetterQueueOptions.cs @@ -141,6 +141,7 @@ public void throws_when_exponential_retry_backoff_initial_delay_is_negative() [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"); diff --git a/src/Dafda/Configuration/DeadLetterQueueOptions.cs b/src/Dafda/Configuration/DeadLetterQueueOptions.cs index 370622a..b501abf 100644 --- a/src/Dafda/Configuration/DeadLetterQueueOptions.cs +++ b/src/Dafda/Configuration/DeadLetterQueueOptions.cs @@ -93,7 +93,7 @@ public DeadLetterQueueOptions WithExponentialRetryBackoff(TimeSpan initialDelay, { EnsureDelayIsSupported(initialDelay, "The retry backoff delay for a dead letter queue"); - if (factor <= 0) + if (double.IsNaN(factor) || factor <= 0) { throw new InvalidConfigurationException("The retry backoff factor for a dead letter queue must be greater than zero."); }