RabbitMQ: settle interop-DLQ deliveries and move to per-message acks (multiple: false)

[jeremydmiller]

Context

Every RabbitMQ ack currently goes out as BasicAckAsync(tag, multiple: true) (src/Transports/RabbitMQ/Wolverine.RabbitMQ/Internal/RabbitMqListener.cs:350-368). Cumulative acks are only correct when completions happen in delivery order. Under out-of-order completion — which already exists today with ConsumerDispatchConcurrency > 1, and which becomes the normal case for the planned native-ack parallel endpoint mode (see the dependent issue) — acking tag N silently sweeps up every lower unacked tag on the channel, including deliveries whose handlers are still running. A crash at that moment is silent message loss.

The code comment above the ack says exactly this ("wrong under a concurrent listener … but it is currently load-bearing"). It is load-bearing because one settle path never settles at all: RabbitMqInteropFriendlyCallback.MoveToErrorsAsync (RabbitMqListener.cs:40-44) posts a copy to the DLQ and never acks or nacks the original delivery. The cumulative sweep from later acks is what currently reclaims those orphaned deliveries. Its sibling RabbitMqChannelCallback.moveToErrorQueueAsync (Internal/RabbitMqChannelCallback.cs:112-137) does this correctly: it marks Acknowledged/HasBeenAcked and then nacks without requeue.

Flipping multiple: true → false in isolation was already attempted during the GH-3492 perf wave and DEFERRED: it measured throughput-neutral but leaked 1 message into quorum1 in the full suite — precisely because of the unsettled interop path (RABBITMQ-PERF-DEEP-DIVE-PLAN.md:183-198, ledger row 217).

This issue is the prerequisite for the native-ack parallel endpoint mode. Do NOT resurrect ack coalescing/batching — that was prototyped and rejected at −10% throughput (basic.ack is a fire-and-forget frame, not an RPC; ledger row 216).

Execution plan

1. Fix the unsettled path. Make RabbitMqInteropFriendlyCallback.MoveToErrorsAsync settle the original delivery the same way RabbitMqChannelCallback.moveToErrorQueueAsync does: set RabbitMqEnvelope.Acknowledged and Envelope.HasBeenAcked before issuing BasicNackAsync(tag, multiple: false, requeue: false) against envelope.DeliveredOn, guarded by RabbitMqListener.CanSettle (RabbitMqListener.cs:309-338). Audit for any other path that produces a delivery which is never settled (search all uses of RabbitMqEnvelope where neither ack nor nack fires).
2. Flip the ack to per-message. Change RabbitMqListener.CompleteAsync(RabbitMqEnvelope) to BasicAckAsync(envelope.DeliveryTag, multiple: false, …). Update/remove the large explanatory comment at RabbitMqListener.cs:357-366 — replace it with a comment stating that per-message acks are required for out-of-order completion and pointing at this issue.
3. Verify no other settle path relies on the cumulative sweep. The NullReferenceException acking a RabbitMQ delivery on a torn-down channel #3687 work (5f6b3bbc7) already routed all settles through envelope.DeliveredOn + CanSettle; re-read CanSettle's doc comment — its stale-tag rationale explicitly assumed multiple: true made replay catastrophic. The guard stays (settling on a replaced channel is still wrong), but the comment should be updated.

Tests

• Repro-first: a test with a listener processing deliveries out of order (e.g. ConsumerDispatchConcurrency > 1 or hand-driven consumer) proving that with multiple: true, completing a later delivery acks an earlier in-flight one (assert via redelivery absence after channel close). This is the red baseline; flip and watch it go green.
• A test that a message routed through RabbitMqInteropFriendlyCallback.MoveToErrorsAsync leaves zero unacked deliveries on the channel (broker queue depth / channel unacked count via management API or a fresh consumer).
• Run the full Wolverine.RabbitMQ.Tests suite locally including the quorum-queue tests that caught the leak last time (docker compose up -d rabbitmq first). The previous attempt leaked exactly 1 message into quorum1 — watch for that specific failure.
• Existing regression coverage to keep green: src/Transports/RabbitMQ/Wolverine.RabbitMQ.Tests/Bugs/Bug_3687_settling_a_delivery_from_a_dead_channel.cs.

Documentation

• Update docs/guide/messaging/transports/rabbitmq/ listener docs where ack behavior is described: acks are per-message; note the fixed DLQ-interop settle behavior.

Acceptance criteria

• No settle path leaves a delivery permanently unacked.
• All acks are multiple: false.
• Full RabbitMQ test suite green locally (expect ~10 min); no message leaked into quorum queues.
• Throughput sanity check against the r-max-inline benchmark from the RabbitMQ performance deep dive: inline-by-default consumption, durable inbox batching, ack/mapper hot paths #3492 wave: per-message acks were previously measured neutral — confirm no regression beyond noise.

Dependencies

• Blocks the native-ack parallel endpoint mode issue (filed separately).
• Independent of everything else; can start immediately.

[jeremydmiller]

Done in PR #3737 (draft #3675 closed as superseded). Full local suite is 488 passed / 2 failed, with both failures accounted for — details below — and no message leaked into a quorum queue, which is the specific thing that killed the previous attempt.

One correction to the execution plan, from auditing every RabbitMqEnvelope use for a path where neither ack nor nack fires:

The unsettled path isn't the one this issue names. RabbitMqInteropFriendlyCallback.MoveToErrorsAsync genuinely doesn't settle, but it isn't orphaning deliveries in practice — MoveToErrorQueue.ExecuteAsync calls lifecycle.CompleteAsync() immediately after MoveToDeadLetterQueueAsync, and NoHandlerContinuation does the same. I checked this rather than assuming: disabled the settle I'd added there and re-ran the repro, and it still passed. So "the cumulative sweep from later acks is what currently reclaims those orphaned deliveries" doesn't hold for that path.

The real one is WorkerQueueMessageConsumer's un-mappable-message branch (WorkerQueueMessageConsumer.cs:121). It dead-letters through the receiver and returns without touching the delivery at all — no continuation follows, because the message never reaches the pipeline. That's the delivery the cumulative sweep was reclaiming.

Both are settled at the source now, and both with an ack rather than a nack. The plan suggested mirroring moveToErrorQueueAsync's BasicNackAsync(tag, false, false), but that sibling nacks precisely so the broker's own x-dead-letter-exchange routes the original — it sends no copy of its own. These two paths have already written the message elsewhere, so a nack would either be discarded (InteropFriendly strips the DLX argument from the queue) or, under UseEnhancedDeadLettering against a Native-mode queue that still has its DLX, put a second copy in the dead letter queue. The tests assert exactly one copy lands.

CanSettle stays, comment corrected as you suggested — per-message acks narrow a replayed stale tag's blast radius to one wrong message, but settling on a replaced channel is still wrong.

Ack coalescing stayed dead, as instructed.

On the two suite failures, both pre-existing:

• multi_tenancy_through_virtual_hosts.send_message_to_a_specific_tenant — fails identically on unmodified main, same line. Tracked-session condition gap; the response is visibly Sent in the activity table before the session gives up.
• end_to_end.use_direct_exchange_with_binding_key — passes in isolation on both main and the branch, and the whole end_to_end class passes 20/20. Cross-class order dependence, matching CIRabbitMQ: three tests fail round one on every run and are masked by the flaky-retry harness #3726.

This unblocks the native-ack parallel endpoint mode. Shipping in 6.24.1 tonight.