diff --git a/docs/guide/messaging/transports/rabbitmq/deadletterqueues.md b/docs/guide/messaging/transports/rabbitmq/deadletterqueues.md index 0681230a6..805055685 100644 --- a/docs/guide/messaging/transports/rabbitmq/deadletterqueues.md +++ b/docs/guide/messaging/transports/rabbitmq/deadletterqueues.md @@ -69,6 +69,36 @@ using var host = await Host.CreateDefaultBuilder() snippet source | anchor +## Enhanced Dead Lettering with Exception Metadata + +By default, Wolverine uses RabbitMQ's native NACK mechanism to move failed messages to the dead letter exchange. While simple, this approach does not include any information about *why* the message failed. + +With `EnableEnhancedDeadLettering()`, Wolverine will instead publish failed messages directly to the dead letter queue with exception metadata headers, then ACK the original message. This gives you structured failure information on each dead-lettered message: + +| Header | Description | +|--------|-------------| +| `exception-type` | Full type name of the exception | +| `exception-message` | The exception message | +| `exception-stack` | The exception stack trace | +| `failed-at` | Unix timestamp (milliseconds) when the failure occurred | + +```cs +using var host = await Host.CreateDefaultBuilder() + .UseWolverine(opts => + { + opts.UseRabbitMq() + .EnableEnhancedDeadLettering(); + }).StartAsync(); +``` + +::: tip +These same metadata headers are automatically included for all other Wolverine transports (SQS, Azure Service Bus, GCP Pub/Sub, NATS, Kafka, Redis, Pulsar) when messages are moved to dead letter queues. +::: + +::: warning +Enhanced dead lettering bypasses RabbitMQ's native dead letter exchange (DLX) mechanism. Messages are published to the DLQ by Wolverine rather than being NACK'd. If you rely on native DLX routing or policies, this mode may not be appropriate. +::: + And lastly, if you don't particularly want to have any Rabbit MQ dead letter queues and you quite like the [database backed dead letter queues](/guide/durability/dead-letter-storage) you get with Wolverine's message durability, you can use the `WolverineStorage` option: diff --git a/src/Testing/CoreTests/Transports/DeadLetterQueueConstantsTests.cs b/src/Testing/CoreTests/Transports/DeadLetterQueueConstantsTests.cs new file mode 100644 index 000000000..2f93bb256 --- /dev/null +++ b/src/Testing/CoreTests/Transports/DeadLetterQueueConstantsTests.cs @@ -0,0 +1,75 @@ +using Shouldly; +using Wolverine.Transports; +using Xunit; + +namespace CoreTests.Transports; + +public class DeadLetterQueueConstantsTests +{ + [Fact] + public void stamp_failure_metadata_sets_all_headers() + { + var envelope = new Envelope(); + var exception = new InvalidOperationException("something went wrong"); + + DeadLetterQueueConstants.StampFailureMetadata(envelope, exception); + + envelope.Headers[DeadLetterQueueConstants.ExceptionTypeHeader] + .ShouldBe(typeof(InvalidOperationException).FullName); + envelope.Headers[DeadLetterQueueConstants.ExceptionMessageHeader] + .ShouldBe("something went wrong"); + envelope.Headers[DeadLetterQueueConstants.ExceptionStackHeader] + .ShouldNotBeNull(); + envelope.Headers[DeadLetterQueueConstants.FailedAtHeader] + .ShouldNotBeNull(); + + long.TryParse(envelope.Headers[DeadLetterQueueConstants.FailedAtHeader], out var timestamp) + .ShouldBeTrue(); + timestamp.ShouldBeGreaterThan(0); + } + + [Fact] + public void stamp_failure_metadata_preserves_existing_headers() + { + var envelope = new Envelope(); + envelope.Headers["custom-header"] = "custom-value"; + envelope.Headers["another"] = "one"; + + var exception = new ArgumentException("bad arg"); + + DeadLetterQueueConstants.StampFailureMetadata(envelope, exception); + + envelope.Headers["custom-header"].ShouldBe("custom-value"); + envelope.Headers["another"].ShouldBe("one"); + envelope.Headers[DeadLetterQueueConstants.ExceptionTypeHeader] + .ShouldBe(typeof(ArgumentException).FullName); + } + + [Fact] + public void stamp_failure_metadata_handles_null_stack_trace() + { + var envelope = new Envelope(); + // Exception created without throwing has null StackTrace + var exception = new Exception("test"); + + DeadLetterQueueConstants.StampFailureMetadata(envelope, exception); + + envelope.Headers[DeadLetterQueueConstants.ExceptionStackHeader].ShouldBe(""); + } + + [Fact] + public void stamp_failure_metadata_overwrites_previous_failure_headers() + { + var envelope = new Envelope(); + var firstException = new InvalidOperationException("first"); + DeadLetterQueueConstants.StampFailureMetadata(envelope, firstException); + + var secondException = new ArgumentException("second"); + DeadLetterQueueConstants.StampFailureMetadata(envelope, secondException); + + envelope.Headers[DeadLetterQueueConstants.ExceptionTypeHeader] + .ShouldBe(typeof(ArgumentException).FullName); + envelope.Headers[DeadLetterQueueConstants.ExceptionMessageHeader] + .ShouldBe("second"); + } +} diff --git a/src/Transports/AWS/Wolverine.AmazonSqs/Internal/SqsListener.cs b/src/Transports/AWS/Wolverine.AmazonSqs/Internal/SqsListener.cs index 4eb529cd1..e690df2f2 100644 --- a/src/Transports/AWS/Wolverine.AmazonSqs/Internal/SqsListener.cs +++ b/src/Transports/AWS/Wolverine.AmazonSqs/Internal/SqsListener.cs @@ -185,6 +185,7 @@ public async Task TryRequeueAsync(Envelope envelope) public Task MoveToErrorsAsync(Envelope envelope, Exception exception) { + DeadLetterQueueConstants.StampFailureMetadata(envelope, exception); return _deadLetterBlock!.PostAsync(envelope); } diff --git a/src/Transports/Azure/Wolverine.AzureServiceBus/Internal/BatchedAzureServiceBusListener.cs b/src/Transports/Azure/Wolverine.AzureServiceBus/Internal/BatchedAzureServiceBusListener.cs index c142c449a..984b9c3f6 100644 --- a/src/Transports/Azure/Wolverine.AzureServiceBus/Internal/BatchedAzureServiceBusListener.cs +++ b/src/Transports/Azure/Wolverine.AzureServiceBus/Internal/BatchedAzureServiceBusListener.cs @@ -101,6 +101,7 @@ public async Task MoveToErrorsAsync(Envelope envelope, Exception exception) { if (envelope is AzureServiceBusEnvelope e) { + DeadLetterQueueConstants.StampFailureMetadata(envelope, exception); e.Exception = exception; await _deadLetter.PostAsync(e); } diff --git a/src/Transports/Azure/Wolverine.AzureServiceBus/Internal/InlineAzureServiceBusListener.cs b/src/Transports/Azure/Wolverine.AzureServiceBus/Internal/InlineAzureServiceBusListener.cs index 5e3c7bf82..983ce3a41 100644 --- a/src/Transports/Azure/Wolverine.AzureServiceBus/Internal/InlineAzureServiceBusListener.cs +++ b/src/Transports/Azure/Wolverine.AzureServiceBus/Internal/InlineAzureServiceBusListener.cs @@ -114,6 +114,7 @@ public async Task MoveToErrorsAsync(Envelope envelope, Exception exception) { if (envelope is AzureServiceBusEnvelope e) { + DeadLetterQueueConstants.StampFailureMetadata(envelope, exception); e.Exception = exception; await _deadLetter.PostAsync(e); } diff --git a/src/Transports/GCP/Wolverine.Pubsub/Internal/PubsubListener.cs b/src/Transports/GCP/Wolverine.Pubsub/Internal/PubsubListener.cs index 3ce12f643..ee8e91ca6 100644 --- a/src/Transports/GCP/Wolverine.Pubsub/Internal/PubsubListener.cs +++ b/src/Transports/GCP/Wolverine.Pubsub/Internal/PubsubListener.cs @@ -111,6 +111,7 @@ public ValueTask DisposeAsync() public Task MoveToErrorsAsync(Envelope envelope, Exception exception) { + DeadLetterQueueConstants.StampFailureMetadata(envelope, exception); return _deadLetter.PostAsync(envelope); } diff --git a/src/Transports/NATS/Wolverine.Nats/Internal/NatsListener.cs b/src/Transports/NATS/Wolverine.Nats/Internal/NatsListener.cs index ed40ec6c7..39a9d4706 100644 --- a/src/Transports/NATS/Wolverine.Nats/Internal/NatsListener.cs +++ b/src/Transports/NATS/Wolverine.Nats/Internal/NatsListener.cs @@ -97,12 +97,8 @@ await natsEnvelope.JetStreamMsg.AckAsync( { envelope.Attempts = (int)(metadata?.NumDelivered ?? 1); - envelope.Headers["x-dlq-reason"] = exception.Message; - envelope.Headers["x-dlq-timestamp"] = DateTimeOffset.UtcNow.ToString("O"); + DeadLetterQueueConstants.StampFailureMetadata(envelope, exception); envelope.Headers["x-dlq-original-subject"] = _endpoint.Subject; - envelope.Headers["x-dlq-attempts"] = envelope.Attempts.ToString(); - envelope.Headers["x-dlq-exception-type"] = - exception.GetType().FullName ?? "Unknown"; await _deadLetterSender.SendAsync(envelope); } diff --git a/src/Transports/Pulsar/Wolverine.Pulsar/PulsarListener.cs b/src/Transports/Pulsar/Wolverine.Pulsar/PulsarListener.cs index a98d6268d..8ffeb0062 100644 --- a/src/Transports/Pulsar/Wolverine.Pulsar/PulsarListener.cs +++ b/src/Transports/Pulsar/Wolverine.Pulsar/PulsarListener.cs @@ -348,6 +348,8 @@ private MessageMetadata BuildMessageMetadata(Envelope envelope, PulsarEnvelope e messageMetadata.DeliverAtTimeAsDateTimeOffset = DateTimeOffset.UtcNow; if (exception != null) { + DeadLetterQueueConstants.StampFailureMetadata(envelope, exception); + var exceptionText = exception.ToString(); messageMetadata[PulsarEnvelopeConstants.Exception] = exceptionText; e.Headers[PulsarEnvelopeConstants.Exception] = exceptionText; diff --git a/src/Transports/RabbitMQ/Wolverine.RabbitMQ/Internal/RabbitMqListener.cs b/src/Transports/RabbitMQ/Wolverine.RabbitMQ/Internal/RabbitMqListener.cs index 89c66ed55..092b9d7e7 100644 --- a/src/Transports/RabbitMQ/Wolverine.RabbitMQ/Internal/RabbitMqListener.cs +++ b/src/Transports/RabbitMQ/Wolverine.RabbitMQ/Internal/RabbitMqListener.cs @@ -39,6 +39,7 @@ public ValueTask DeferAsync(Envelope envelope) public async Task MoveToErrorsAsync(Envelope envelope, Exception exception) { + DeadLetterQueueConstants.StampFailureMetadata(envelope, exception); await _sendBlock.PostAsync(envelope); } @@ -73,8 +74,11 @@ public RabbitMqListener(IWolverineRuntime runtime, _transport = transport; _receiver = receiver ?? throw new ArgumentNullException(nameof(receiver)); - _callback = (Queue.DeadLetterQueue != null) & - (Queue.DeadLetterQueue?.Mode == DeadLetterQueueMode.InteropFriendly) + var useEnhancedOrInterop = Queue.DeadLetterQueue != null && + (Queue.DeadLetterQueue.Mode == DeadLetterQueueMode.InteropFriendly || + _transport.UseEnhancedDeadLettering); + + _callback = useEnhancedOrInterop ? new RabbitMqInteropFriendlyCallback(_transport, _transport.Queues[Queue.DeadLetterQueue!.QueueName], _runtime) : _transport.Callback!; diff --git a/src/Transports/RabbitMQ/Wolverine.RabbitMQ/Internal/RabbitMqTransport.cs b/src/Transports/RabbitMQ/Wolverine.RabbitMQ/Internal/RabbitMqTransport.cs index 971ee0686..6ce8dd676 100644 --- a/src/Transports/RabbitMQ/Wolverine.RabbitMQ/Internal/RabbitMqTransport.cs +++ b/src/Transports/RabbitMQ/Wolverine.RabbitMQ/Internal/RabbitMqTransport.cs @@ -317,6 +317,14 @@ internal ConnectionMonitor BuildConnection(ConnectionRole role) public ILogger Logger { get; private set; } = NullLogger.Instance; + /// + /// When true, Wolverine will publish failed messages to the dead letter queue + /// with exception metadata headers instead of using RabbitMQ's native NACK-based + /// dead lettering. This enables richer error information at the cost of not using + /// native RabbitMQ dead letter exchange mechanisms. + /// + public bool UseEnhancedDeadLettering { get; set; } + /// /// Opt into making Wolverine "auto ping" new listeners by trying to send a fake Wolverine "ping" message /// This *might* assist in Wolverine auto-starting rabbit mq connections that have failed on the Rabbit MQ side diff --git a/src/Transports/RabbitMQ/Wolverine.RabbitMQ/Internal/RabbitMqTransportExpression.cs b/src/Transports/RabbitMQ/Wolverine.RabbitMQ/Internal/RabbitMqTransportExpression.cs index d69149367..cbfaab45b 100644 --- a/src/Transports/RabbitMQ/Wolverine.RabbitMQ/Internal/RabbitMqTransportExpression.cs +++ b/src/Transports/RabbitMQ/Wolverine.RabbitMQ/Internal/RabbitMqTransportExpression.cs @@ -12,6 +12,18 @@ public RabbitMqTransportExpression(RabbitMqTransport transport, WolverineOptions { } + /// + /// When enabled, Wolverine will publish failed messages to the dead letter queue + /// with exception metadata headers (exception type, message, stack trace, timestamp) + /// instead of using RabbitMQ's native NACK-based dead lettering. This enables richer + /// error information at the cost of not using native RabbitMQ dead letter exchange mechanisms. + /// + public RabbitMqTransportExpression EnableEnhancedDeadLettering() + { + Transport.UseEnhancedDeadLettering = true; + return this; + } + /// /// Opt into making Wolverine "auto ping" new listeners by trying to send a fake Wolverine "ping" message /// This *might* assist in Wolverine auto-starting rabbit mq connections that have failed on the Rabbit MQ side diff --git a/src/Wolverine/Runtime/WorkerQueues/BufferedReceiver.cs b/src/Wolverine/Runtime/WorkerQueues/BufferedReceiver.cs index 6112976de..3691fb40d 100644 --- a/src/Wolverine/Runtime/WorkerQueues/BufferedReceiver.cs +++ b/src/Wolverine/Runtime/WorkerQueues/BufferedReceiver.cs @@ -219,6 +219,7 @@ public void Dispose() public Task MoveToErrorsAsync(Envelope envelope, Exception exception) { + DeadLetterQueueConstants.StampFailureMetadata(envelope, exception); return _moveToErrors!.PostAsync(envelope); } diff --git a/src/Wolverine/Runtime/WorkerQueues/DurableReceiver.cs b/src/Wolverine/Runtime/WorkerQueues/DurableReceiver.cs index 2b33d32e8..71ae86b20 100644 --- a/src/Wolverine/Runtime/WorkerQueues/DurableReceiver.cs +++ b/src/Wolverine/Runtime/WorkerQueues/DurableReceiver.cs @@ -313,6 +313,7 @@ public void Dispose() public Task MoveToErrorsAsync(Envelope envelope, Exception exception) { envelope.Failure = exception; + DeadLetterQueueConstants.StampFailureMetadata(envelope, exception); return _moveToErrors.PostAsync(envelope); } diff --git a/src/Wolverine/Transports/DeadLetterQueueConstants.cs b/src/Wolverine/Transports/DeadLetterQueueConstants.cs index e52e6b5c1..3ba91a8ec 100644 --- a/src/Wolverine/Transports/DeadLetterQueueConstants.cs +++ b/src/Wolverine/Transports/DeadLetterQueueConstants.cs @@ -2,6 +2,17 @@ namespace Wolverine.Transports; public static class DeadLetterQueueConstants { + /// + /// Stamps the envelope headers with standard failure metadata from the given exception. + /// + public static void StampFailureMetadata(Envelope envelope, Exception exception) + { + envelope.Headers[ExceptionTypeHeader] = exception.GetType().FullName ?? "Unknown"; + envelope.Headers[ExceptionMessageHeader] = exception.Message; + envelope.Headers[ExceptionStackHeader] = exception.StackTrace ?? ""; + envelope.Headers[FailedAtHeader] = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds().ToString(); + } + /// /// The default queue/topic name used for dead letter queues across all transports. ///