Add test demonstrating RabbitMQ DLQ replay bug

[Joelgullander]

Bug: RabbitMQ DLQ Replay Not Working with SQL Server Persistence

Summary

When using RabbitMQ transport with SQL Server persistence and DeadLetterQueueMode.WolverineStorage, replaying messages from the Dead Letter Queue (DLQ) does not work. Messages are successfully moved from the DLQ to the Incoming table, but they are never processed by the handler.

Root Cause

The issue is in the MarkReceived method in src/Wolverine/Envelope.Internals.cs (lines 140-145). When a message is initially received by a RabbitMQ listener, the Destination property is set to the consumer-specific address (e.g., rabbitmq://queue/myqueue?consumer=123) instead of the base queue address (e.g., rabbitmq://queue/myqueue).

Later, when the message is moved from the DLQ back to the Incoming table via MoveReplayableErrorMessagesToIncomingOperation, the original ReceivedAt value (which contains the ?consumer= parameter) is preserved. However, the DurableReceiver looks for messages with ReceivedAt matching the base queue address, causing a mismatch.

Technical Details

Current Behavior

1. Message fails processing → goes to DLQ with ReceivedAt = "rabbitmq://queue/myqueue?consumer=123"
2. Message is marked as replayable → moved to Incoming table with same ReceivedAt value
3. DurableReceiver queries for messages with ReceivedAt = "rabbitmq://queue/myqueue" (base address)
4. No match found → message sits in Incoming table unprocessed

Expected Behavior

1. Message fails processing → goes to DLQ with ReceivedAt = "rabbitmq://queue/myqueue" (base address)
2. Message is marked as replayable → moved to Incoming table with same ReceivedAt value
3. DurableReceiver queries for messages with ReceivedAt = "rabbitmq://queue/myqueue" (base address)
4. Match found → message is processed

Evidence

Test Case

Created isolated test src/Transports/RabbitMQ/Wolverine.RabbitMQ.Tests/Bugs/Bug_ReplayDeadLetterQueue.cs that demonstrates the issue for both durable and non-durable RabbitMQ queues.

Test Output

[HANDLER] Called. FailFirst=True
[BEFORE REPLAY] DLQ: 1, Incoming: 0
[AFTER REPLAY] DLQ: 0, Incoming: 1
[INCOMING] Id: 08ddc96c-e980-e57a-e6aa-525c34390000, Status: Incoming, OwnerId: 0, ScheduledTime: , Attempts: 0, ReceivedAt: rabbitmq://queue/replay-dlq-9600661c-0a01-4cab-9223-1e8f5942a56e?consumer=

The ReceivedAt value contains the ?consumer= parameter, preventing the DurableReceiver from finding the message.

Code Location

The bug is in src/Wolverine/Envelope.Internals.cs:

internal void MarkReceived(IListener listener, DateTimeOffset now, DurabilitySettings settings)
{
    Listener = listener;

    // If this is a stream with multiple consumers, use the consumer-specific address
    if (listener is ISupportMultipleConsumers multiConsumerListener)
    {
        Destination = multiConsumerListener.ConsumerAddress; // ❌ This sets the consumer-specific address
    }
    else
    {
        Destination = listener.Address; // ✅ This sets the base address
    }
    // ...
}

Proposed Fix

Modify the MarkReceived method to always use the base address for RabbitMQ listeners:

internal void MarkReceived(IListener listener, DateTimeOffset now, DurabilitySettings settings)
{
    Listener = listener;

    // Always use the base address to ensure DLQ replay works correctly
    Destination = listener.Address;

    // ... rest of the method
}

Impact

• Affects: RabbitMQ transport with SQL Server/PostgreSQL persistence and DeadLetterQueueMode.WolverineStorage

Related Files

• src/Wolverine/Envelope.Internals.cs
• src/Transports/RabbitMQ/Wolverine.RabbitMQ/Internal/RabbitMqListener.cs - Defines Address vs ConsumerAddress
• src/Transports/RabbitMQ/Wolverine.RabbitMQ.Tests/Bugs/Bug_ReplayDeadLetterQueue.cs - Test case demonstrating the issue
• src/Persistence/Wolverine.RDBMS/Durability/MoveReplayableErrorMessagesToIncomingOperation.cs - Operation that moves messages from DLQ to Incoming

[Joelgullander]

Related #1594