diff --git a/docs/adr/0027-invariant-prose-names-the-oracle-that-falsifies-it.md b/docs/adr/0027-invariant-prose-names-the-oracle-that-falsifies-it.md index 875bf513..dfacf53b 100644 --- a/docs/adr/0027-invariant-prose-names-the-oracle-that-falsifies-it.md +++ b/docs/adr/0027-invariant-prose-names-the-oracle-that-falsifies-it.md @@ -323,6 +323,19 @@ is that audit, not a lint. A tracker entry is now open at #524. +**Amended 2026-09-28: further instances of the trigger shape, after promotion.** A local pre-PR review of the #533 +`Chatter.MessageBrokers.RabbitMQ` branch returned a cluster of findings against the `INVARIANT:` comments on +`RabbitMqReceiver` that described the native quorum counters. The comments asserted a universal negative — that the +Receiver reads `x-acquired-count` on no delivery shape — and named facts that pinned particular delivery shapes; no +finite set of facts pins the negative over all of them. Each pass tightened one clause and the next pass found another +the named oracles did not reach. Each false clause lived on one surface, the code comment itself, so these are the trigger's shape and are not +excluded as restatement drift. A `grep` lint would have passed every one of them, because each named a real, passing, +relevant `[Fact]`. They change nothing recorded above: the trigger is already tripped and #524's scope stands as filed. +The response was again a re-keying, not a check. Consumption of the broker's counters moved to the one point where a +delivery is captured, so the comments now make one claim each about that one site, each naming an oracle and a measured +mutation; the universal negative is no longer claimed, because downstream of that site the counters do not exist. +ADR-0042 records the mechanism, under *Amendment: the native counters are consumed at the receive boundary*. + ## Consequences - An `INVARIANT:` comment is now a claim with an obligation attached, and `NOTE:` is the unobligated form. A diff --git a/docs/adr/0042-quorum-redelivery-uses-basic-reject-because-only-reject-means-the-delivery-failed.md b/docs/adr/0042-quorum-redelivery-uses-basic-reject-because-only-reject-means-the-delivery-failed.md new file mode 100644 index 00000000..dfdd5565 --- /dev/null +++ b/docs/adr/0042-quorum-redelivery-uses-basic-reject-because-only-reject-means-the-delivery-failed.md @@ -0,0 +1,215 @@ +--- +status: accepted +date: 2026-09-28 +--- + +# Quorum redelivery uses basic.reject, because only reject means the delivery failed + +When a handler fails on a quorum-queue delivery and the Receiver returns the message for another attempt, +`RabbitMqReceiver` now settles it with `basic.reject(requeue: true)` instead of `basic.nack(requeue: true)`. On +RabbitMQ 4.3 a quorum queue counts the two verbs differently: only `basic.reject` records a failed delivery attempt, so +under `basic.nack` the native `x-delivery-count` header is never emitted, Chatter's attempt count stays at 1, and a +poison message loops without end. The swap changes nothing on earlier brokers, where the two verbs run the same code. +This ADR records the cause, the options considered, the one chosen and what it leaves as it was. + +Issue #533. + +## Context + +**ADR-0001's quorum strategy reads the broker's own count.** Under `QueueType.Quorum` the Receiver reads the native +`x-delivery-count` header and stamps `MessageContext.ReceiveAttempts` as that count + 1, flooring the first delivery at +attempt 1. The core deadletters once the attempt count reaches `maxReceiveAttempts`. ADR-0001 named the header; it never +named the verb that returns a failed delivery, and `NackMessageAsync` used `basic.nack(requeue: true)`. + +**RabbitMQ 4.3 split the quorum-queue counter in two.** The 4.3 quorum-queue documentation, section *When is delivery +count incremented?* (https://www.rabbitmq.com/docs/quorum-queues; source +`versioned_docs/version-4.3/quorum-queues/index.md` in `rabbitmq/rabbitmq-website`), tabulates two counters. +`acquired-count` counts assignments to a consumer; `delivery-count` counts failed delivery attempts and is the counter +the queue's `delivery-limit` is evaluated against. For AMQP 0.9.1: + +| Trigger | `acquired-count` | `delivery-count` | +| --- | --- | --- | +| `basic.nack` | incremented | not incremented | +| `basic.reject` | incremented | incremented | +| Client crash / connection loss | incremented | incremented | +| Intra-cluster network partition | incremented | not incremented | +| Consumer timeout | incremented | not incremented | + +The same page states that since 4.3 the delivery count "will only be incremented for genuine failures such as +session/channel crashes or when using the modified outcome with delivery-failed=true or basic.reject in AMQP 0.9.1". It +introduces `x-acquired-count` as the header for the number of times a message has been assigned to a consumer, and +records that the quorum-queue `delivery-limit` defaults to 20 since RabbitMQ 4.0. + +**The broker source shows why the verbs now differ.** In `deps/rabbit/src/rabbit_channel.erl` of `rabbitmq-server`: + +- On `main` (4.3), the `basic.nack` handler settles with `Op = requeue`, and the `basic.reject` handler settles with + `settle(true, false)`, which is `{modify, DeliveryFailed = true, ...}`. Only the second is a failed delivery. +- On `v3.13.x`, `v4.0.x` and `v4.2.x`, both handlers call the identical `reject(DeliveryTag, Requeue, Multiple, State)`. + `basic.reject` hardcodes `Multiple = false`, and Chatter already passes `multiple: false` to `basic.nack`. Before 4.3 + the two verbs are the same operation. + +In `deps/rabbit/src/rabbit_fifo_client.erl` on `main`, `add_delivery_count_header` emits `x-delivery-count` only once a +delivery count exists. Under a nack loop it never exists, so the header is absent on every redelivery, the Receiver +reads no count, and the attempt count stays at 1. That is the #533 mechanism. The same function back-fills the +counters for messages carrying pre-4.3 headers, so a message in flight across an upgrade keeps its count. + +**The integration suite reproduced it.** `RabbitMqDeliveryCountingOn43Tests`, on the `RabbitMq43Collection` against a +`rabbitmq:4.3-management` container, drove a quorum queue whose handler always throws. Under `basic.nack` every +redelivery carried `x-acquired-count` values 1, 2, 3 and so on and no `x-delivery-count`; the poison message looped +about 4,200 times in 20 seconds, and the broker's default `delivery-limit` never fired. A classic queue on the same +broker was unaffected, because the classic strategy counts with its own republished `x-chatter-delivery-count` header +and never nacks. + +## Considered Options + +### Option A — settle a failed quorum delivery with `basic.reject(requeue: true)` (ACCEPTED) + +Recorded under *Decision*. One verb is correct on every broker version. It says what happened: the Receiver returns the +message because its handler failed. It re-arms the broker's `delivery-limit` as a safety net, and it changes no public +API, configuration, ordering or throughput. + +### Option B — read `x-acquired-count` as the attempt count (REJECTED) + +`x-acquired-count` counts assignments, not failures. A consumer timeout or an intra-cluster partition increments it +without any handler failing, so each would spend a healthy message's retry budget. It leaves the broker's +`delivery-limit` unarmed, because the Receiver would still settle with a verb the broker does not count as a failure. And +the header does not exist before 4.3, so the Receiver would have to branch on its presence. + +### Option C — apply ADR-0001's classic republish counter to quorum queues (REJECTED) + +It carries every cost ADR-0001 records for the classic strategy: the republish and the ack are not atomic, so a crash +between them leaves a duplicate; the republished message goes to the tail of the queue and loses its position; and each +attempt doubles the traffic. On a quorum queue each attempt also appends a Raft log entry, and the republished copy is a +new message whose broker delivery count starts again from zero, so the queue's `delivery-limit` could never fire. + +### Option D — probe the broker version, or branch on which header is present (REJECTED) + +Option A behaves the same on every version, so a branch buys nothing and adds a path that only some brokers exercise. + +## Decision + +**`RabbitMqReceiver.NackMessageAsync` settles a failed quorum-queue delivery with `basic.reject(requeue: true)`.** On +RabbitMQ 4.3 and later that increments the quorum queue's delivery count, so the broker emits `x-delivery-count` on the +redelivery and the Receiver's attempt count climbs. Before 4.3 the broker runs the same code for `basic.reject` as for +`basic.nack`, so nothing changes there and no version detection is needed. + +**`x-delivery-count` remains the only attempt source on a quorum queue.** The ADR-0001 quorum strategy is unchanged. +`x-acquired-count` is deliberately not read as an attempt count (Option B). + +**A counter header is carried onward only by the party that stamps it.** The adapter stamps `x-chatter-delivery-count`, +so that counter is adapter-owned: `ReceivedMessage.Headers` carries it, the classic redelivery hop re-stamps it on the +Receiver's own republish, and the Receiver removes it from the `MessageContext` it emits, so it does not ride onto a +message the application sends while handling. The broker stamps `x-delivery-count` and `x-acquired-count` about the +delivery it has just made, so those counters are broker-owned and read-only to the Receiver. They are consumed at the +receive boundary: `RabbitMqReceiver.BufferDeliveryAsync`, the single point where a delivery is captured, reads the +prior-delivery count once from the key the queue type selects (`x-delivery-count` on a quorum queue, +`x-chatter-delivery-count` on a classic queue), carries it onward as an `int`, and removes both native counters from the +headers it carries. Neither native counter exists downstream of that point: not in `ReceivedMessage.Headers`, not in +the emitted `MessageContext`, not on a message sent while handling, and not on any republished copy (Dead-letter Queue, +Error Queue or classic redelivery). + +**Amended 2026-09-28: the native counters are consumed, not stripped.** As first accepted, the paragraph above called +`x-acquired-count` "adapter-owned receive state like `x-delivery-count`" and said it was "stripped from outbound +headers". Both halves were wrong. The strip ran where the Receiver emits the core `MessageContext`, but every republish +hop built its headers from the carried `ReceivedMessage.Headers`, which still held both native counters, so they rode +onto the Dead-letter Queue, Error Queue and classic-redelivery copies. Consuming them at the capture point closes that +leak for every hop at once. See *Amendment: the native counters are consumed at the receive boundary* below. + +**The classic strategy and `TransactionMode.None` are untouched.** A classic queue still republishes with +`x-chatter-delivery-count` and acks the original. Under `TransactionMode.None` the consumer auto-acks, so there is no +delivery to settle. + +This ADR refines ADR-0001 by naming the settlement verb its quorum strategy depends on. It does not supersede it. + +## Consequences + +- **A poison message on a RabbitMQ 4.3 quorum queue is deadlettered again.** The attempt count reaches + `maxReceiveAttempts`, and the Receiver deadletters the message: to the Dead-letter Queue when one is configured, and + otherwise to the Error Queue. +- **Nothing changes before 4.3, on classic queues, or under `TransactionMode.None`.** +- **The broker's `delivery-limit` advances again on 4.3.** Keep `maxReceiveAttempts` below the queue's `delivery-limit` + (default 20). At or above it, the broker can drop the message, or dead-letter it through the queue's DLX when one is + configured, before the Receiver reaches `maxReceiveAttempts`. The adapter provisions no topology and cannot read the + effective limit without the management API or a non-passive declare, so this is documented rather than guarded. The + same exposure already existed on RabbitMQ 4.0 to 4.2, where `basic.nack` still advanced the delivery count. +- **CI pins both halves.** A focused `rabbitmq:4.3-management` collection (`RabbitMq43Collection`) pins the 4.3 + semantics. `rabbitmq:3.13-management` remains the broad-surface fixture (`RabbitMqFixture`) and pins the pre-4.3 half. +- **`ReceivedMessage.Headers` narrows.** It no longer holds `x-delivery-count` or `x-acquired-count`. No public API + signature or configuration changes; `MessageContext.ReceiveAttempts` is unaffected. + +## Amendment: the native counters are consumed at the receive boundary + +**Amended 2026-09-28**, from a local pre-PR review of this branch. + +**What the review found.** The first implementation strip-listed the native counters where the Receiver emits the core +`MessageContext`, and the `INVARIANT:` comments that described it asserted a universal negative: that the Receiver reads +`x-acquired-count` on no delivery shape. No oracle could pin a claim of that shape, and each review pass that tightened +one clause found another that the named oracles did not reach. One of those findings was not only a prose defect: the +republish hops built their headers from `ReceivedMessage.Headers`, which the emit-time strip never touched, so +`x-delivery-count` rode onto the Dead-letter Queue, Error Queue and classic-redelivery copies. + +**What changed.** Consumption moved to the one capture point, as stated under *Decision*. The attempt count is read there +once and carried as an `int` on a private buffered-delivery record, and both native counters are removed from the headers +before `ReceivedMessage` is built. The emit-time strip of the natives was deleted, because there is nothing left for it to +strip; the emit-time strip of the adapter-owned `x-chatter-delivery-count` stays, because that counter is carried on +purpose. The `INVARIANT:` comments on `BufferDeliveryAsync` now make one claim each about that one site, each naming its +oracle and a measured mutation. + +**Rejected narrower variant: discard only `x-acquired-count` at the capture point.** It would have answered the +finding as filed, which named `x-acquired-count`, and left `x-delivery-count` carried in `ReceivedMessage.Headers`, +where the republish hops would still forward it. Under the Same-Framing Test the next finding is the same shape with a +different header name, so the variant was rejected in favour of consuming every broker-owned counter at the boundary. + +## Closed-by-Construction Acceptance Test + +> What class of future finding does this make impossible, and why? + +**Eliminated: any read, re-emission or republish of a native broker counter downstream of the capture point.** Every +step after `BufferDeliveryAsync` works from the `ReceivedMessage` and the carried `int`: `ReceiveMessageAsync`, the +translator's `ToCore`, the settlement paths and `ToRepublishAmqp`. None of them has a native counter to read, forward or +strip, so a hop added later inherits the absence instead of needing a strip of its own. What holds the property is two +`Remove` calls at one site rather than one strip per hop. The claim is pinned by the counter-ownership `INVARIANT:` on +`BufferDeliveryAsync`: deleting those two calls reddens exactly six unit facts, the three receive-side facts in +`WhenReceivingMessage` and the three republish-hop facts in `WhenSettlingMessage` +(`MustNotRepublishNativeQuorumCountersOnTheDeadletterHop`, `...OnTheErrorQueueHop` and +`...OnTheClassicRedeliveryHop`). The boundary covers code that runs after the capture point; code added inside the +consumer callback, ahead of it, still sees the raw delivery. + +**Honest residue: the capture point can still select the wrong key.** Which header the attempt count is read from is +decided at that one site by the queue type, and nothing structural stops it choosing wrongly. It is pinned on the quorum +arm by `MustResolveReceiveAttemptsFromDeliveryCountWhenBothQuorumCountersArePresent` (`x-delivery-count` 2 beside +`x-acquired-count` 9 resolves attempt 3) and on the classic arm by +`MustNotRepublishNativeQuorumCountersOnTheClassicRedeliveryHop` (`x-chatter-delivery-count` 1 beside both natives +resolves attempt 2). The measured mutations are recorded in the +attempt-count `INVARIANT:` on `BufferDeliveryAsync`. + +**Not closed: load-bearing claims in hand-maintained `INVARIANT:` prose.** The re-keying replaced a universal negative +with claims about one site that an oracle can pin, but nothing stops the next comment claiming more than its oracle +pins. That meta-class is ADR-0027's, recorded there as further instances of its promotion-trigger shape, and is tracked +by #524. + +## References + +- Issue #533 — quorum-queue attempt counting on RabbitMQ 4.3. The defect this ADR resolves. +- ADR-0001 — *RabbitMQ classic-queue redelivery counting via header-stamped republish*. The quorum strategy this ADR + refines, and the classic republish costs that rule out Option C. +- RabbitMQ documentation, *Quorum Queues*, section *When is delivery count incremented?* — + https://www.rabbitmq.com/docs/quorum-queues (source `versioned_docs/version-4.3/quorum-queues/index.md` in + `rabbitmq/rabbitmq-website`). +- `rabbitmq-server` `deps/rabbit/src/rabbit_channel.erl` — the `basic.nack` and `basic.reject` handlers on `main`, and + on `v3.13.x`, `v4.0.x` and `v4.2.x`. +- `rabbitmq-server` `deps/rabbit/src/rabbit_fifo_client.erl` — `add_delivery_count_header` on `main`. +- `src/Chatter.MessageBrokers.RabbitMQ/src/Chatter.MessageBrokers.RabbitMQ/Receiving/RabbitMqReceiver.cs` — + `NackMessageAsync`, the settlement this ADR changes, and `BufferDeliveryAsync`, the capture point where the native + counters are consumed, with its counter-ownership and attempt-count `INVARIANT:` comments. +- `src/Chatter.MessageBrokers.RabbitMQ/tests/Receiving/UsingRabbitMqReceiver/WhenSettlingMessage.cs` — the settlement + unit oracles, and the three `MustNotRepublishNativeQuorumCounters...` republish-hop facts. +- `src/Chatter.MessageBrokers.RabbitMQ/tests/Receiving/UsingRabbitMqReceiver/WhenReceivingMessage.cs` — the + receive-side counter facts, including `MustResolveReceiveAttemptsFromDeliveryCountWhenBothQuorumCountersArePresent`. +- ADR-0027 — *An `INVARIANT:` comment names the oracle that falsifies it*. The rule the amended `INVARIANT:` comments + follow, and the meta-class this ADR leaves open. +- Issue #524 — the `INVARIANT:` audit ADR-0027's promotion trigger filed. +- `src/Chatter.MessageBrokers.RabbitMQ/tests/Integration/RabbitMqDeliveryCountingOn43Tests.cs` — the RabbitMQ 4.3 + reproduction and its facts. +- `src/Chatter.MessageBrokers.RabbitMQ/tests/Integration/RabbitMqNackRedeliveryTests.cs` — quorum and classic + redelivery on the 3.13 fixture. diff --git a/docs/design/rabbitmq-adapter.md b/docs/design/rabbitmq-adapter.md index 09b67a75..2ee5ac31 100644 --- a/docs/design/rabbitmq-adapter.md +++ b/docs/design/rabbitmq-adapter.md @@ -110,8 +110,9 @@ model: - An `AsyncEventingBasicConsumer` (or equivalent async consumer) is registered on the single serialized receive channel during `InitializeAsync`. Its delivery callback **does not handle** - the message; it wraps each delivery (body + headers + delivery tag + owning channel-epoch) and - **writes it into a bounded `Channel` buffer**. + the message; it wraps each delivery (body + headers + delivery tag + owning channel-epoch + the + prior-delivery count read from the delivery's counter header) and **writes it into a bounded + `Channel` buffer**. The broker's native counter headers are consumed here and not buffered (§6). - `ReceiveMessageAsync` **reads** from that `Channel` with `await reader.ReadAsync(ct)`. When the buffer is empty the read **asynchronously parks** the loop — no CPU, no polling — until the push consumer enqueues the next delivery or cancellation fires. This satisfies the blocking-pull @@ -235,9 +236,11 @@ The adapter maps the core's three terminal operations onto AMQP, all on the gate and all epoch-guarded: - **Ack on success** (`AckMessageAsync`) → `BasicAck(deliveryTag)`. Epoch-guarded. -- **Nack → redelivery on failure** (`NackMessageAsync`) → `BasicNack(deliveryTag, requeue: true)` - (Quorum) — the broker increments native `x-delivery-count` and redelivers. (Classic uses the - header-stamped republish counter; see §6.) Epoch-guarded. +- **Nack → redelivery on failure** (`NackMessageAsync`) → `BasicReject(deliveryTag, requeue: true)` + (Quorum) — the broker counts a failed delivery attempt, increments native `x-delivery-count` and + redelivers. `BasicReject`, not `BasicNack`: from RabbitMQ 4.3 a quorum queue counts only + `basic.reject` as a failed delivery, and before 4.3 the two verbs run the same broker code (ADR 0042). + (Classic uses the header-stamped republish counter; see §6.) Epoch-guarded. - **Deadletter once Max Receives Exceeded** (`DeadletterMessageAsync`) → the adapter **republishes** the body to the **attribute-declared** DeadletterQueueName / ErrorQueueName (an adapter-owned republish, authoritative over any broker-side DLX configuration), then **acks the original**. This @@ -255,7 +258,7 @@ stateDiagram-v2 [*] --> Received: push consumer buffers delivery Received --> Handling: core pulls + dispatches to handler Handling --> Acked: handler success → BasicAck - Handling --> NackedRedelivered: handler failure,
count ≤ limit → BasicNack(requeue) / classic republish + Handling --> NackedRedelivered: handler failure,
count ≤ limit → BasicReject(requeue) / classic republish Handling --> DeadLettered: Max Receives Exceeded →
republish to declared DLQ/Error, then ack original NackedRedelivered --> Received: broker redelivers Acked --> [*] @@ -278,8 +281,15 @@ single value (ADR 0001). A `QueueType` option selects the strategy (default **Quorum**, recommended): - **Quorum strategy** — reads RabbitMQ's **native `x-delivery-count`** header, which the broker - increments per redelivery. Adapter computes attempts = `x-delivery-count + 1` and stamps - `ReceiveAttempts`. Redelivery on failure is a plain `BasicNack(requeue: true)`. + increments per failed delivery attempt. Adapter computes attempts = `x-delivery-count + 1` and + stamps `ReceiveAttempts`. Redelivery on failure is a plain `BasicReject(requeue: true)`, which + RabbitMQ counts as a failed delivery attempt; from RabbitMQ 4.3 `BasicNack(requeue: true)` does not + advance `x-delivery-count` (ADR 0042). The 4.3+ `x-acquired-count` header counts assignments to a + consumer, not failures, so the adapter never reads it as the attempt count. Both native counters are + broker-owned and consumed at the receive boundary: the push consumer reads the prior-delivery count + once, from the key the queue type selects, and removes `x-delivery-count` and `x-acquired-count` + from the headers it buffers, so neither reaches `ReceivedMessage.Headers`, the emitted + `MessageContext`, a send made while handling, or any republished copy (ADR 0042). - **Classic strategy** — classic queues expose no native counter, so the adapter uses a **header-stamped republish counter**: on retry it republishes the message to its own queue with a custom `x-chatter-delivery-count` header incremented by 1 (publisher-confirmed), then acks the @@ -298,7 +308,7 @@ flowchart TD Count --> Stamp["stamp MessageContext.ReceiveAttempts
(MANDATORY — absent or unusable falls to the core's dead-letter sentinel)"] Stamp --> Decide{"attempts > maxReceiveAttempts?"} Decide -- no, handler succeeded --> Ack["BasicAck"] - Decide -- "no, handler failed" --> Redeliver["Quorum: BasicNack(requeue)
Classic: republish w/ incremented header + ack original"] + Decide -- "no, handler failed" --> Redeliver["Quorum: BasicReject(requeue)
Classic: republish w/ incremented header + ack original"] Decide -- yes --> DLQ["republish to declared
Deadletter / Error path, then ack original"] ``` diff --git a/src/Chatter.MessageBrokers.RabbitMQ/CONTEXT.md b/src/Chatter.MessageBrokers.RabbitMQ/CONTEXT.md index 53524849..b238c534 100644 --- a/src/Chatter.MessageBrokers.RabbitMQ/CONTEXT.md +++ b/src/Chatter.MessageBrokers.RabbitMQ/CONTEXT.md @@ -20,14 +20,15 @@ _Avoid_: dispatcher (reserved for the Message Brokers Brokered Message Dispatche **Dead-Letter Exchange (DLX) / Dead-letter Queue**: The destination for messages that exhausted Recovery; the adapter republishes to the attribute-declared DeadletterQueueName / ErrorQueueName rather than relying on broker DLX configuration. -**RabbitMq Settlement**: The RabbitMQ realization of Settlement (Message Brokers context). Under `TransactionMode.None` the AMQP push consumer is registered with `autoAck`, so RabbitMQ removed the delivery at receive time and acknowledge, negative acknowledge and deadletter all report the **Not Required** Settlement Outcome — the message is simply dropped, which is what at-most-once means. Otherwise an ack, a requeue/republish nack, or a deadletter republish-then-ack reports **Settled**, and a settlement whose delivery is absent from the message broker context reports **Failed**. +**RabbitMq Settlement**: The RabbitMQ realization of Settlement (Message Brokers context). Under `TransactionMode.None` the AMQP push consumer is registered with `autoAck`, so RabbitMQ removed the delivery at receive time and acknowledge, negative acknowledge and deadletter all report the **Not Required** Settlement Outcome — the message is simply dropped, which is what at-most-once means. Otherwise an ack, a negative acknowledge (a requeue `basic.reject` on a quorum queue, a republish-then-ack on a classic queue; ADR 0042), or a deadletter republish-then-ack reports **Settled**, and a settlement whose delivery is absent from the message broker context reports **Failed**. **Channel Epoch**: The generation counter of the receive channel, carried on every buffered delivery. A settlement runs under the receive-channel gate and compares the delivery's carried epoch with the current one; on a mismatch the channel was recycled since delivery, so the delivery tag is meaningless on the new channel and RabbitMQ has already redelivered the message. The settlement is skipped and reports the **Failed** Settlement Outcome — it was ATTEMPTED and did not happen — never Not Required. **Error-Queue Write Ownership** (`WritesToErrorQueue`): This receiver owns the Error Queue write exactly when NO Dead-letter Queue is configured — the ERROR-ONLY configuration, where deadlettering republishes the failed delivery to the Error Queue itself (publisher-confirmed) before acking the original. That path truthfully reports **Settled**, and the separate ownership signal is what keeps the Brokered Message Receiver from forwarding a SECOND copy of the same poison message to the SAME Error Queue: the single-copy rule holds because the duplicate is suppressed by ownership, never by misreporting the Settlement Outcome. With a Dead-letter Queue configured the receiver republishes there and never touches the Error Queue, so ownership stays with the Brokered Message Receiver and a copy is forwarded to the Error Queue as well. Configuring neither queue is rejected at startup, except under at-most-once (`TransactionMode.None`), which has no poison target to require. _Avoid_: gating the Error Queue write on the Settlement Outcome (a truthful Settled would then write two copies of every poison message). -**Delivery Count Strategy**: How redeliveries are counted — Quorum (native `x-delivery-count`, recommended) or Classic (header-stamped republish counter). See ADR 0001. +**Delivery Count Strategy**: How redeliveries are counted — Quorum (native `x-delivery-count`, recommended) or Classic (header-stamped republish counter). On a quorum queue the Receiver returns a failed delivery with `basic.reject` and requeue, which RabbitMQ counts as a failed delivery attempt and records in `x-delivery-count`; from RabbitMQ 4.3 a `basic.nack` is not counted. The native `x-acquired-count` (RabbitMQ 4.3+) counts assignments to a consumer, not failures, and is never the attempt count. See ADR 0001 and ADR 0042. +_Avoid_: reading `x-acquired-count` as the attempt count (a consumer timeout or a partition advances it with no handler failure). **RabbitMq Options**: Configuration for the connection, prefetch, queue type, TLS, and body settings, supplied via the options builder. TLS (`UseTls` / `TlsServerName` / `WithTls(...)`) applies only to the discrete host/credential connection path — an `amqps://` connection URI already enables TLS and takes precedence — and offers no surface to weaken or disable certificate validation. @@ -45,11 +46,11 @@ _Avoid_: gating the Error Queue write on the Settlement Outcome (a truthful Sett ## Example dialogue > **Dev:** "On a classic queue, how does it know a message is poison if RabbitMQ won't count deliveries?" -> **Domain expert:** "The Classic Delivery Count Strategy republishes the message to its own queue with an incremented `x-chatter-delivery-count` header, then acks the original — the count rides in the message. On a quorum queue we just read the native `x-delivery-count` instead, which is why quorum is the recommended default." +> **Domain expert:** "The Classic Delivery Count Strategy republishes the message to its own queue with an incremented `x-chatter-delivery-count` header, then acks the original — the count rides in the message. On a quorum queue the Receiver returns the failed delivery with `basic.reject` and requeue, RabbitMQ counts that as a failed delivery attempt, and we read the native `x-delivery-count` instead, which is why quorum is the recommended default." ## Flagged ambiguities -- **Quorum vs Classic delivery-count semantics**: quorum queues count redeliveries natively; classic queues do not, so the count is carried in a republish header (ADR 0001) with a rare-duplicate trade-off. +- **Quorum vs Classic delivery-count semantics**: quorum queues count failed deliveries natively, and from RabbitMQ 4.3 only a `basic.reject` (or a lost consumer) counts as one, not a `basic.nack` (ADR 0042); classic queues do not, so the count is carried in a republish header (ADR 0001) with a rare-duplicate trade-off. - **Default-exchange-as-queue-name convention**: when no Exchange override is given, publishing uses the default exchange with Routing Key equal to the destination Queue name — Routing Key and Queue name coincide only under this convention. ## Known limitations diff --git a/src/Chatter.MessageBrokers.RabbitMQ/src/Chatter.MessageBrokers.RabbitMQ/CHANGELOG.md b/src/Chatter.MessageBrokers.RabbitMQ/src/Chatter.MessageBrokers.RabbitMQ/CHANGELOG.md index cfff5fdd..0164eec4 100644 --- a/src/Chatter.MessageBrokers.RabbitMQ/src/Chatter.MessageBrokers.RabbitMQ/CHANGELOG.md +++ b/src/Chatter.MessageBrokers.RabbitMQ/src/Chatter.MessageBrokers.RabbitMQ/CHANGELOG.md @@ -6,6 +6,13 @@ This project follows [Keep a Changelog](https://keepachangelog.com/en/1.0.0/) an ## [Unreleased] +## [0.6.2] - 2026-09-28 + +### Fixed + +- **A quorum-queue message whose handler kept failing was redelivered indefinitely and never dead-lettered, on RabbitMQ 4.3 and later.** From RabbitMQ 4.3, a `basic.nack` with requeue no longer advances the queue's native `x-delivery-count`, so the receiver's attempt count never rose and `maxReceiveAttempts` was never reached. The receiver now returns a failed delivery with `basic.reject` and requeue instead, which RabbitMQ counts as a failed delivery attempt on every supported version. When `maxReceiveAttempts` is kept below the queue's `delivery-limit` (20 by default from RabbitMQ 4.0), the message is dead-lettered once `maxReceiveAttempts` is hit (to the Dead-letter Queue when one is configured, otherwise the Error Queue); at or above it, RabbitMQ can drop the message or dead-letter it through the queue's DLX first. No configuration or API change, and no behavior change on RabbitMQ before 4.3 or on classic queues, which already counted redeliveries through their own header. The earlier advice to use classic queues on RabbitMQ 4.3 and later no longer applies — see [Quorum queues](https://github.com/brenpike/Chatter/blob/master/src/Chatter.MessageBrokers.RabbitMQ/src/README.md#quorum-queues). (#533) +- **The native quorum counters `x-delivery-count` and `x-acquired-count` are now consumed when a delivery is received, not carried onto anything downstream.** They are stripped before `ReceivedMessage` is built, so `ReceivedMessage.Headers` never contains them, and they are never copied onto messages sent while handling a delivery or onto the message republished to the Dead-letter Queue, the Error Queue, or a classic redelivery. `MessageContext.ReceiveAttempts` is unaffected. (#533) + ## [0.6.1] - 2026-09-24 ### Changed diff --git a/src/Chatter.MessageBrokers.RabbitMQ/src/Chatter.MessageBrokers.RabbitMQ/Chatter.MessageBrokers.RabbitMQ.csproj b/src/Chatter.MessageBrokers.RabbitMQ/src/Chatter.MessageBrokers.RabbitMQ/Chatter.MessageBrokers.RabbitMQ.csproj index 7623b8f6..0c417853 100644 --- a/src/Chatter.MessageBrokers.RabbitMQ/src/Chatter.MessageBrokers.RabbitMQ/Chatter.MessageBrokers.RabbitMQ.csproj +++ b/src/Chatter.MessageBrokers.RabbitMQ/src/Chatter.MessageBrokers.RabbitMQ/Chatter.MessageBrokers.RabbitMQ.csproj @@ -4,7 +4,7 @@ net10.0 latest Chatter.MessageBrokers.RabbitMQ - 0.6.1 + 0.6.2 Brennan Pike Brennan Pike An implementation of the Chatter.MessageBrokers adapter library for RabbitMQ. diff --git a/src/Chatter.MessageBrokers.RabbitMQ/src/Chatter.MessageBrokers.RabbitMQ/Receiving/RabbitMqReceiver.cs b/src/Chatter.MessageBrokers.RabbitMQ/src/Chatter.MessageBrokers.RabbitMQ/Receiving/RabbitMqReceiver.cs index 49052df6..475acead 100644 --- a/src/Chatter.MessageBrokers.RabbitMQ/src/Chatter.MessageBrokers.RabbitMQ/Receiving/RabbitMqReceiver.cs +++ b/src/Chatter.MessageBrokers.RabbitMQ/src/Chatter.MessageBrokers.RabbitMQ/Receiving/RabbitMqReceiver.cs @@ -52,9 +52,14 @@ namespace Chatter.MessageBrokers.RabbitMQ.Receiving /// internal sealed class RabbitMqReceiver : IMessagingInfrastructureReceiver { - // The native quorum-queue redelivery counter the broker increments per redelivery. + // The native quorum-queue delivery counter. On RabbitMQ 4.3+ the broker advances it per FAILED delivery + // (basic.reject or a lost consumer), not per basic.nack return; before 4.3 it advances per redelivery. private const string _nativeDeliveryCountHeader = "x-delivery-count"; + // The native quorum-queue assignment counter RabbitMQ 4.3+ stamps on each redelivery. Withheld at the capture + // point (BufferDeliveryAsync); not an attempt source (ADR-0042). + private const string _nativeAcquiredCountHeader = "x-acquired-count"; + // The settlement contract requires every unsettled outcome to explain itself; these are the two // delivery-independent explanations this receiver reports. private const string _atMostOnceNothingToSettleReason = @@ -62,12 +67,18 @@ internal sealed class RabbitMqReceiver : IMessagingInfrastructureReceiver private const string _noCarriedDeliveryReason = "the context carries no " + nameof(ReceivedMessage) + ", so the delivery to settle could not be located"; + /// + /// One buffered delivery: the the Receiver carries onward, and the number of + /// prior deliveries read at the capture point from the header the queue type counts attempts with. + /// + private readonly record struct BufferedDelivery(ReceivedMessage Message, int PriorDeliveries); + private readonly IRabbitMqConnectionSource _connectionSource; private readonly RabbitMqOptions _rabbitOptions; private readonly IBodyConverterFactory _bodyConverterFactory; private readonly ILogger _logger; - private Channel _buffer; + private Channel _buffer; private ReceiverOptions _options; private int _prefetch; // TransactionMode.None is at-most-once: the broker must drop the delivery as it is pushed (no crash-window @@ -154,7 +165,7 @@ public async Task InitializeAsync(ReceiverOptions options, CancellationToken can // prefetch (removing all backpressure), and other large values would silently under-prefetch. Saturate // at ushort.MaxValue instead so a misconfiguration degrades to the maximum supported prefetch, never to 0. _prefetch = ClampPrefetch(Math.Max(Math.Max(1, _rabbitOptions.Prefetch), options.MaxConcurrentCalls)); - _buffer = System.Threading.Channels.Channel.CreateBounded( + _buffer = System.Threading.Channels.Channel.CreateBounded( new BoundedChannelOptions(_prefetch) { SingleReader = false, @@ -265,17 +276,55 @@ await channel.BasicQosAsync(prefetchSize: 0, // re-applied through the translator on a republish hop. Without this capture the republish rebuilt // BasicProperties from scratch and dropped every delivered native property (e.g. a classic-queue per-message // TTL lost on the first redelivery). + // COUNTER OWNERSHIP: a counter header is carried onward only by the party that stamps it. The adapter stamps + // x-chatter-delivery-count, so ReceivedMessage.Headers carries it and the classic redelivery hop re-stamps it + // (BuildClassicRedeliveryHeaders). The broker stamps x-delivery-count and x-acquired-count about the delivery + // it has just made, so they are read-only to the Receiver and are consumed HERE, at the receive boundary: + // read once for the attempt count, then withheld from the ReceivedMessage every later step works from. + // INVARIANT: the broker-owned counters do not exist downstream of this capture point, so neither the emitted + // core context nor any republish hop (dead-letter, Error Queue, classic redelivery) can carry them. Pinned by + // WhenReceivingMessage.MustStripNativeDeliveryCountHeaderWhilePreservingOtherHeaders, + // .MustStripNativeAcquiredCountHeaderWhilePreservingOtherHeaders and + // .MustResolveReceiveAttemptsFromDeliveryCountWhenBothQuorumCountersArePresent, and by + // WhenSettlingMessage.MustNotRepublishNativeQuorumCountersOnTheDeadletterHop, .OnTheErrorQueueHop and + // .OnTheClassicRedeliveryHop. Mutation: carry the natives through (delete the two carriedHeaders.Remove + // calls below). Measured: reddens exactly those six of the module's 366 unit facts. + // INVARIANT: the RECEIVE attempt count is derived once, at this capture point, from the key the queue type + // selects (ReadPriorDeliveries: x-delivery-count on a quorum queue, x-chatter-delivery-count on a classic + // queue), and reaches ReceiveMessageAsync as BufferedDelivery.PriorDeliveries. The classic republish stamp is + // computed separately at settlement, in BuildClassicRedeliveryHeaders, from the same adapter-owned + // x-chatter-delivery-count with the same normalization. Pinned on the quorum arm by + // WhenReceivingMessage.MustResolveReceiveAttemptsFromDeliveryCountWhenBothQuorumCountersArePresent + // (x-delivery-count 2 beside x-acquired-count 9 resolves ReceiveAttempts 3), and on the classic arm by + // WhenSettlingMessage.MustNotRepublishNativeQuorumCountersOnTheClassicRedeliveryHop (x-chatter-delivery-count 1 + // beside x-delivery-count 2 and x-acquired-count 7 resolves ReceiveAttempts 2). Mutations, each measured on + // the module's 366 unit facts: selecting x-acquired-count as the quorum key reddens 11 (the both-counters fact + // and every fact that reads a quorum count); reading x-acquired-count as a max beside the selected key on + // every arm reddens 3 (the both-counters fact, the classic redelivery-hop fact and + // MustStripNativeAcquiredCountHeaderWhilePreservingOtherHeaders); reading it as a max on the classic arm only + // reddens the classic redelivery-hop fact alone; selecting x-delivery-count on the classic arm reddens 5 (the + // classic redelivery-hop fact and four classic count-reading facts). + // NOTE: why x-acquired-count is not an attempt source (ADR-0042) — it counts assignments to a consumer, not + // failed deliveries, so a consumer timeout or an intra-cluster partition advances it with no handler failing, + // and reading it would spend a healthy message's retry budget — is broker behavior no test pins. The + // RabbitMqDeliveryCountingOn43Tests facts cannot tell the two counters apart, because both climb under + // basic.reject. private async Task BufferDeliveryAsync(BasicDeliverEventArgs delivery, long epoch) { var properties = delivery.BasicProperties; var facts = RabbitMqMessageTranslator.CaptureFacts(properties); + var carriedHeaders = properties?.Headers is { } deliveredHeaders + ? new Dictionary(deliveredHeaders) + : new Dictionary(); + var priorDeliveries = ReadPriorDeliveries(carriedHeaders); + carriedHeaders.Remove(_nativeDeliveryCountHeader); + carriedHeaders.Remove(_nativeAcquiredCountHeader); + var received = new ReceivedMessage(body: delivery.Body.ToArray(), deliveryTag: delivery.DeliveryTag, channelEpoch: epoch, - headers: properties?.Headers is { } headers - ? new Dictionary(headers) - : new Dictionary(), + headers: carriedHeaders, exchange: delivery.Exchange, routingKey: delivery.RoutingKey, redelivered: delivery.Redelivered, @@ -289,16 +338,17 @@ private async Task BufferDeliveryAsync(BasicDeliverEventArgs delivery, long epoc contentType: facts.ContentType, correlationId: facts.CorrelationId); - await _buffer.Writer.WriteAsync(received, delivery.CancellationToken).ConfigureAwait(false); + await _buffer.Writer.WriteAsync(new BufferedDelivery(received, priorDeliveries), delivery.CancellationToken).ConfigureAwait(false); } public async Task ReceiveMessageAsync(TransactionContext transactionContext, CancellationToken cancellationToken) { // Blocking pull: async-park when the buffer is empty (no poll, no spin) until the push consumer // enqueues a delivery or the loop token cancels. - var received = await _buffer.Reader.ReadAsync(cancellationToken).ConfigureAwait(false); + var buffered = await _buffer.Reader.ReadAsync(cancellationToken).ConfigureAwait(false); + var received = buffered.Message; - var receiveAttempts = ResolveReceiveAttempts(received); + var receiveAttempts = ResolveReceiveAttempts(buffered.PriorDeliveries); // Seed the context through the SINGLE translation contract: the delivered header table is decoded // (string-typed header keys byte[]->string so the core's type-tested string reads see the header's real @@ -321,17 +371,16 @@ public async Task ReceiveMessageAsync(TransactionContext t headers.Remove(RabbitMqMessageContext.TargetExchange); headers.Remove(RabbitMqMessageContext.RoutingKey); - // INVARIANT: the delivery-count headers are adapter-owned receive-attempt state, NEVER outbound payload. - // The core seeds a receive-then-send follow-up's options from THIS inbound context, and RabbitMqSender - // republishes the full context as headers, so a delivery-count header surviving here would ride onto the - // next send. After a classic-queue retry x-chatter-delivery-count is present on the delivery (likewise the - // native x-delivery-count on a quorum delivery); left in place it would be re-stamped onto an outbound - // message and read back by ResolveReceiveAttempts on the next classic queue's first delivery as a stale - // redelivery — inflating ReceiveAttempts and deadlettering a fresh message with too few attempts. Strip - // BOTH counter keys here (alongside the routing-override strip above) so receive-attempt state never leaks - // out of the receive boundary; ResolveReceiveAttempts above already read the value off received.Headers. + // INVARIANT: the adapter-owned x-chatter-delivery-count is receive state, never outbound payload. + // ReceivedMessage carries it so the classic redelivery hop can re-stamp it, but the core seeds a + // receive-then-send follow-up's options from THIS context and RabbitMqSender publishes the full context + // as headers, so a copy left here would ride onto the next send and be read on the next queue's first + // delivery as a stale redelivery count. Pinned by + // WhenReceivingMessage.MustStripClassicDeliveryCountHeaderWhilePreservingOtherHeaders. Mutation: delete + // the Remove below. Measured: reddens that fact alone of the module's 366 unit facts. + // The broker-owned counters need no strip here: they were consumed at the capture point (see the + // counter-ownership INVARIANTs on BufferDeliveryAsync). headers.Remove(RabbitMqMessageContext.DeliveryCountHeader); - headers.Remove(_nativeDeliveryCountHeader); headers[RabbitMqMessageContext.DeliveryTag] = received.DeliveryTag; headers[RabbitMqMessageContext.ChannelEpoch] = received.ChannelEpoch; @@ -375,16 +424,18 @@ public async Task ReceiveMessageAsync(TransactionContext t // cast straight to int, stamp a negative or wrapped ReceiveAttempts that the core compares to // MaxReceiveAttempts — letting a poison message dodge deadlettering or get a bogus retry budget. Saturate // the raw long into a non-negative int before stamping. - private int ResolveReceiveAttempts(ReceivedMessage received) + private int ReadPriorDeliveries(IReadOnlyDictionary deliveredHeaders) { var headerKey = _rabbitOptions.QueueType == QueueType.Quorum ? _nativeDeliveryCountHeader : RabbitMqMessageContext.DeliveryCountHeader; - var priorDeliveries = SaturateToNonNegativeInt(ReadHeaderAsLong(received.Headers, headerKey, 0L)); - return priorDeliveries == int.MaxValue ? int.MaxValue : priorDeliveries + 1; + return SaturateToNonNegativeInt(ReadHeaderAsLong(deliveredHeaders, headerKey, 0L)); } + private static int ResolveReceiveAttempts(int priorDeliveries) + => priorDeliveries == int.MaxValue ? int.MaxValue : priorDeliveries + 1; + // Clamp the resolved prefetch into the AMQP-wire ushort range [1, ushort.MaxValue] so the (ushort) cast at // the BasicQosAsync call site cannot wrap. A configured value above 65,535 saturates at ushort.MaxValue // (maximum supported prefetch) rather than wrapping to 0 (unlimited, no backpressure) or a much smaller @@ -464,9 +515,27 @@ public Task NackMessageAsync(MessageBrokerContext context, Tra cancellationToken: cancellationToken); } - // Quorum queues: a plain requeue lets the broker increment the native x-delivery-count and redeliver. + // INVARIANT: the quorum requeue verb is basic.reject(requeue: true). Pinned by + // WhenSettlingMessage.MustRejectWithRequeueOnNackForQuorum and, on the 4.3 fixture, by + // RabbitMqDeliveryCountingOn43Tests.QuorumQueueDeadlettersOnMaxReceivesOn43 and + // .QuorumQueueReceiveAttemptsClimbAcrossRedeliveriesOn43. Mutation: revert this call to + // BasicNackAsync(multiple: false, requeue: true). Measured: reddens exactly those three — 1 of the + // module's 366 unit facts and 2 of its 23 integration facts. + // INVARIANT: reject still redelivers and advances the receive attempt count before 4.3. Pinned by + // RabbitMqNackRedeliveryTests.ThrowingHandlerCausesRedeliveryAndClimbingReceiveAttempts on the 3.13 + // fixture. Mutation: requeue: true -> requeue: false. Measured: reddens that fact and, beyond it, + // RabbitMqTransactionModeTests.ReceiveOnlyTransactionModeRedeliversOnThrow (3.13, quorum), the two 4.3 + // facts above and MustRejectWithRequeueOnNackForQuorum — 1 of 366 unit facts and 4 of 23 integration + // facts. Of these, only the two 3.13 facts stay green under the BasicNackAsync revert. + // NOTE (ADR-0042): the broker counter semantics behind the verb — on RabbitMQ >= 4.3 basic.nack is a + // "returned" outcome that does not advance x-delivery-count, basic.reject is the "failed" outcome that + // does — are broker behavior; the 4.3 facts pin their consequence, not the counter table itself. + // NOTE: that the two verbs share one server path before 4.3, and are therefore behavior-identical there, + // is broker-internal behavior NO TEST PINS. No shipped fact issues basic.nack against a 3.13 broker; the + // measured BasicNackAsync revert leaving the 3.13 fact green is consistent with the shared path, not + // proof of it. return SettleOnReceiveChannelAsync(received, (channel) => - channel.BasicNackAsync(received.DeliveryTag, multiple: false, requeue: true, cancellationToken), cancellationToken); + channel.BasicRejectAsync(received.DeliveryTag, requeue: true, cancellationToken), cancellationToken); } /// diff --git a/src/Chatter.MessageBrokers.RabbitMQ/src/Chatter.MessageBrokers.RabbitMQ/Receiving/ReceivedMessage.cs b/src/Chatter.MessageBrokers.RabbitMQ/src/Chatter.MessageBrokers.RabbitMQ/Receiving/ReceivedMessage.cs index 267a935d..abb6ad61 100644 --- a/src/Chatter.MessageBrokers.RabbitMQ/src/Chatter.MessageBrokers.RabbitMQ/Receiving/ReceivedMessage.cs +++ b/src/Chatter.MessageBrokers.RabbitMQ/src/Chatter.MessageBrokers.RabbitMQ/Receiving/ReceivedMessage.cs @@ -7,10 +7,14 @@ namespace Chatter.MessageBrokers.RabbitMQ.Receiving /// A single AMQP delivery buffered by between the push consumer and the /// blocking pull of ReceiveMessageAsync. Carries the raw body, the broker-assigned delivery tag, /// the epoch of the receive channel that delivered it (used to detect a stale-channel ack), the delivery - /// headers (including the native x-delivery-count for quorum queues and the adapter's - /// x-chatter-delivery-count for classic queues), the source exchange / routing key, the + /// headers, the source exchange / routing key, the /// broker's redelivered flag, and the curated set of delivered native AMQP properties that must be /// re-applied when the receiver republishes the message on a nack-redelivery or deadletter hop. + /// The delivery headers carry the adapter's own x-chatter-delivery-count when the delivery arrived with + /// it, which on a classic queue is once the Receiver has republished the message for another attempt. The + /// broker's native counters, x-delivery-count and x-acquired-count, are consumed at the receive + /// boundary and are not carried: the Receiver reads the attempt count there, from x-delivery-count on a + /// quorum queue and from x-chatter-delivery-count on a classic queue. /// public sealed class ReceivedMessage { @@ -58,7 +62,10 @@ public ReceivedMessage(byte[] body, /// The epoch of the receive channel that delivered this message. public long ChannelEpoch { get; } - /// The delivery headers, including any native or adapter delivery-count header. + /// + /// The delivery headers, including the adapter's x-chatter-delivery-count when the delivery carried it. + /// The Receiver does not carry the broker's native x-delivery-count or x-acquired-count here. + /// public IReadOnlyDictionary Headers { get; } /// The exchange the message was published to. diff --git a/src/Chatter.MessageBrokers.RabbitMQ/src/README.md b/src/Chatter.MessageBrokers.RabbitMQ/src/README.md index 327bb450..17dbb83b 100644 --- a/src/Chatter.MessageBrokers.RabbitMQ/src/README.md +++ b/src/Chatter.MessageBrokers.RabbitMQ/src/README.md @@ -311,11 +311,11 @@ Each receiver counts deliveries so it can dead-letter a message once `maxReceive ### Quorum queues -Quorum is the default and the recommended choice. A failed message is negatively acknowledged with requeue, which returns the same message to the queue rather than publishing a copy. On its next delivery the receiver reads the attempt number from the native `x-delivery-count` header. RabbitMQ decides where a returned message goes, so it can be delivered after messages that were queued behind it. +Quorum is the default and the recommended choice. When a handler fails, the receiver returns the message with `basic.reject` and requeue, which puts the same message back on the queue rather than publishing a copy. RabbitMQ counts the return as a failed delivery attempt and advances the message's native `x-delivery-count` header, and on the next delivery the receiver reads the attempt number from that header. RabbitMQ decides where a returned message goes, so it can be delivered after messages that were queued behind it. -The count rises only when RabbitMQ counts a `basic.nack` with requeue as a failed delivery, which RabbitMQ versions before 4.3 do. From RabbitMQ 4.3, `basic.nack` does not increment `x-delivery-count` or count toward the queue's `delivery-limit`, so a message that keeps failing is redelivered indefinitely and never reaches `maxReceiveAttempts`. On RabbitMQ 4.3 or later, use classic queues, which count with their own header. +On RabbitMQ 4.3 and later, a quorum-queue delayed retry of type `failed` applies to these returns, because each one advances the delivery count. -> **Note:** From RabbitMQ 4.0, quorum queues have a broker-side `delivery-limit` of 20 by default. RabbitMQ drops or dead-letters a message past that limit itself, so keep `maxReceiveAttempts` below the queue's `delivery-limit`. +> **Note:** From RabbitMQ 4.0, quorum queues have a broker-side `delivery-limit` of 20 by default, and each failed delivery attempt counts toward it. Keep `maxReceiveAttempts` below the queue's `delivery-limit`. At or above it, RabbitMQ can drop the message, or dead-letter it through the queue's DLX, before the receiver reaches `maxReceiveAttempts`. This package provisions no topology and does not read the queue's `delivery-limit`. ### Classic queues @@ -612,10 +612,6 @@ The application's RabbitMQ user needs: A process supports exactly one RabbitMQ queue receiver. The connection owns one receive channel and one AMQP subscription, so a second receiver would displace the first. Registering a second one, whether through `AddQueueReceiver` or a `[BrokeredMessage]` receiver that resolves to RabbitMQ, makes `AddRabbitMq` throw `NotSupportedException` before the host starts. Split receivers across processes or services; multi-receiver support is tracked in [#195](https://github.com/brenpike/Chatter/issues/195). -### Quorum-queue counting on RabbitMQ 4.3 and later - -On RabbitMQ 4.3 and later, a quorum-queue receiver's attempt count does not rise when a delivery fails, so a failing message is redelivered without reaching `maxReceiveAttempts`. See [Quorum queues](#quorum-queues). Tracked in [#533](https://github.com/brenpike/Chatter/issues/533). - ## Diagnostics This package emits no telemetry of its own. Broker spans and metrics come from the Chatter.MessageBrokers `ActivitySource` and `Meter`; see [Chatter.MessageBrokers diagnostics](https://github.com/brenpike/Chatter/blob/master/src/Chatter.MessageBrokers/src/README.md#diagnostics). The W3C `traceparent` and `tracestate` values travel as AMQP headers in both directions, so a trace continues across the broker (see [Trace context propagation](https://github.com/brenpike/Chatter/blob/master/src/Chatter.MessageBrokers/src/README.md#trace-context-propagation)). diff --git a/src/Chatter.MessageBrokers.RabbitMQ/tests/Integration/RabbitMq43Fixture.cs b/src/Chatter.MessageBrokers.RabbitMQ/tests/Integration/RabbitMq43Fixture.cs new file mode 100644 index 00000000..8ec25f29 --- /dev/null +++ b/src/Chatter.MessageBrokers.RabbitMQ/tests/Integration/RabbitMq43Fixture.cs @@ -0,0 +1,98 @@ +using System; +using System.Threading; +using System.Threading.Tasks; +using Testcontainers.RabbitMq; +using Chatter.Testing.Core.Integration; +using Xunit; + +namespace Chatter.MessageBrokers.RabbitMQ.Tests.Integration +{ + // Collection fixture that brings up a RabbitMQ 4.3 container once per test collection, alongside the 3.13 + // container RabbitMqFixture starts. RabbitMQ 4.3 changed how a quorum queue counts a basic.nack with requeue, + // which is what a quorum-queue Receiver relies on to advance its attempt count, so the delivery-counting + // scenarios run against this broker version as well. Topology is declared in-test via RabbitMqTopology, exactly + // as for the 3.13 fixture. + // + // When Docker is unavailable the fixture is a no-op: InitializeAsync starts nothing (the container stays null) + // and GetAmqpConnectionString throws. The RequiresDockerFact attribute SKIPS the tests at discovery time in that + // case, so a no-Docker `dotnet test` stays green and the fixture's throw is never reached. + public sealed class RabbitMq43Fixture : IAsyncLifetime + { + private const string RabbitMqImage = "rabbitmq:4.3-management"; + + // The container-internal management API port. Testcontainers maps this to a RANDOM host port (NOT 15672), + // so the management base URI must be built from GetMappedPublicPort(15672) rather than a hardcoded port. + private const int ManagementContainerPort = 15672; + + // Bounds container start so a wedged Docker pull/start fails fast instead of hanging the test collection. + // Generous because a cold image pull is slow; still finite. + private static readonly TimeSpan StartupTimeout = TimeSpan.FromMinutes(5); + + private RabbitMqContainer _container; + + // The AMQP connection URI (amqp://user:pass@host:mappedPort) for the running container. Throws if the + // container was not started (Docker unavailable). + public string GetAmqpConnectionString() + { + if (_container is null) + { + throw new InvalidOperationException( + "The RabbitMQ 4.3 container was not started (Docker unavailable). " + + "Integration tests must be guarded with [RequiresDockerFact]."); + } + + return _container.GetConnectionString(); + } + + // The management HTTP API base URI (http://host:mappedManagementPort) for the running container, resolved + // from the randomly mapped host port at call time. Throws if the container was not started (Docker + // unavailable). + public Uri GetManagementBaseUri() + { + if (_container is null) + { + throw new InvalidOperationException( + "The RabbitMQ 4.3 container was not started (Docker unavailable). " + + "Integration tests must be guarded with [RequiresDockerFact]."); + } + + return new UriBuilder("http", _container.Hostname, _container.GetMappedPublicPort(ManagementContainerPort)).Uri; + } + + public async Task InitializeAsync() + { + if (!DockerEnvironment.IsAvailable) + { + return; + } + + using var startupCts = new CancellationTokenSource(StartupTimeout); + + // RabbitMqBuilder publishes only the AMQP port (5672) by default, so the management port (15672) is + // published explicitly to a random host port; GetManagementBaseUri resolves that mapped port. + _container = new RabbitMqBuilder(RabbitMqImage) + .WithPortBinding(ManagementContainerPort, assignRandomHostPort: true) + .Build(); + await _container.StartAsync(startupCts.Token).ConfigureAwait(false); + } + + public async Task DisposeAsync() + { + if (_container is null) + { + return; + } + + await _container.DisposeAsync().ConfigureAwait(false); + } + } + + // DisableParallelization keeps this collection from running concurrently with the 3.13 RabbitMqCollection (also + // non-parallel), so the two brokers and their Receivers never compete for the host at the same time, and the + // process-global ActivityListener hazard RabbitMqCollection guards against cannot leak across them. + [CollectionDefinition(Name, DisableParallelization = true)] + public sealed class RabbitMq43Collection : ICollectionFixture + { + public const string Name = "RabbitMq43"; + } +} diff --git a/src/Chatter.MessageBrokers.RabbitMQ/tests/Integration/RabbitMqDeliveryCountingOn43Tests.cs b/src/Chatter.MessageBrokers.RabbitMQ/tests/Integration/RabbitMqDeliveryCountingOn43Tests.cs new file mode 100644 index 00000000..30239072 --- /dev/null +++ b/src/Chatter.MessageBrokers.RabbitMQ/tests/Integration/RabbitMqDeliveryCountingOn43Tests.cs @@ -0,0 +1,226 @@ +using System; +using System.Collections.Generic; +using System.Linq; +using System.Threading; +using System.Threading.Tasks; +using Chatter.CQRS.Commands; +using Chatter.MessageBrokers; +using Chatter.MessageBrokers.Context; +using Chatter.MessageBrokers.RabbitMQ.Configuration; +using Chatter.MessageBrokers.RabbitMQ.Receiving; +using Chatter.Testing.Core.Integration; +using FluentAssertions; +using Xunit; +using Xunit.Abstractions; + +namespace Chatter.MessageBrokers.RabbitMQ.Tests.Integration +{ + // Delivery counting against a RabbitMQ 4.3 broker. From 4.3 a quorum queue no longer counts a basic.nack with + // requeue as a failed delivery, so the native x-delivery-count a quorum-queue Receiver reads its attempt number + // from may never advance. These facts pin the outcome a Receiver owes on 4.3 regardless of how the count is + // sourced: + // + // - a quorum-queue message whose handler always throws reaches the dead-letter queue once maxReceiveAttempts + // is exhausted, republished by the Receiver itself (the FailureDetails header is present), rather than + // being redelivered indefinitely or dropped by the broker's own delivery-limit; + // - the ReceiveAttempts a handler observes climbs across quorum redeliveries (the 4.3 mirror of + // RabbitMqNackRedeliveryTests.ThrowingHandlerCausesRedeliveryAndClimbingReceiveAttempts); + // - a classic-queue message whose handler always throws still dead-letters, since the classic strategy counts + // with the Receiver's own x-chatter-delivery-count header and does not depend on the broker's counter. + // + // Each fact writes the ReceiveAttempts the Receiver resolved and the broker's redelivered flag for the first few + // deliveries to the test output, so a failure shows whether the attempt count advanced across redeliveries. The + // broker's native counters themselves are consumed at the receive boundary and are not observable here. + // + // ANTI-INFINITE-LOOP: every fact clears ThrowOnHandle in its finally block so a message still looping on the + // work queue is acked on its next delivery before DisposeAsync drains the pump. + [Trait("Category", "Integration")] + [Collection(RabbitMq43Collection.Name)] + public class RabbitMqDeliveryCountingOn43Tests + { + // Three attempts: low enough that a working count dead-letters within a second or two, and above 1 so the + // quorum dead-letter needs the count to advance at least twice rather than tripping on the first delivery. + private const int MaxReceiveAttempts = 3; + + // How many delivery snapshots each fact writes to the test output. A looping message can be delivered + // thousands of times inside the wait, so the output is capped. + private const int DeliverySnapshotLimit = 5; + + private static readonly TimeSpan DeadLetterWait = TimeSpan.FromSeconds(20); + private static readonly TimeSpan RedeliveryWait = TimeSpan.FromSeconds(20); + + private readonly RabbitMq43Fixture _fixture; + private readonly ITestOutputHelper _output; + + public RabbitMqDeliveryCountingOn43Tests(RabbitMq43Fixture fixture, ITestOutputHelper output) + { + _fixture = fixture; + _output = output; + } + + // Distinct command type per scenario so each scenario's handler signal is independent. + public sealed class QuorumMaxReceivesOn43Command : ICommand + { + public string Marker { get; set; } + } + + public sealed class QuorumRedeliveryOn43Command : ICommand + { + public string Marker { get; set; } + } + + public sealed class ClassicMaxReceivesOn43Command : ICommand + { + public string Marker { get; set; } + } + + [RequiresDockerFact] + public Task QuorumQueueDeadlettersOnMaxReceivesOn43() + => RunMaxReceivesScenarioAsync( + QueueType.Quorum, + () => new QuorumMaxReceivesOn43Command { Marker = "quorum-maxreceives-43" }); + + [RequiresDockerFact] + public Task ClassicQueueDeadlettersOnMaxReceivesOn43() + => RunMaxReceivesScenarioAsync( + QueueType.Classic, + () => new ClassicMaxReceivesOn43Command { Marker = "classic-maxreceives-43" }); + + [RequiresDockerFact] + public async Task QuorumQueueReceiveAttemptsClimbAcrossRedeliveriesOn43() + { + var set = RabbitMqTopology.CreateSet("v43_redelivery", QueueType.Quorum); + await RabbitMqTopology.DeclareAsync(_fixture.GetAmqpConnectionString(), set, CancellationToken.None); + + var harness = ChatterRabbitMqPipelineHarness.Build( + _fixture.GetAmqpConnectionString(), + QueueType.Quorum, + rmq => rmq.AddQueueReceiver(set.WorkQueueName, deadLetterQueuePath: set.DeadLetterQueueName), + typeof(QuorumRedeliveryOn43Command)); + try + { + await harness.StartAsync(); + + harness.GetSignal().ThrowOnHandle = + () => new InvalidOperationException("v43-redelivery-test forced throw"); + + await harness.SendToQueueAsync(new QuorumRedeliveryOn43Command { Marker = "quorum-redelivery-43" }, set.WorkQueueName); + + var observedCount = await harness.WaitForInvocationCountAsync( + minCount: 2, RedeliveryWait); + + harness.GetSignal().ThrowOnHandle = null; + + var records = harness.GetSignal().Records.ToList(); + WriteDeliveredAttempts(records.Select(record => record.Context)); + + observedCount.Should().BeGreaterThanOrEqualTo(2, + "the throwing handler must cause at least one redelivery of the nacked quorum message"); + + var maxAttempts = records + .Where(record => record.Context?.BrokeredMessage?.MessageContext?.ContainsKey(MessageContext.ReceiveAttempts) == true) + .Select(record => Convert.ToInt32(record.Context.BrokeredMessage.MessageContext[MessageContext.ReceiveAttempts])) + .DefaultIfEmpty(0) + .Max(); + + maxAttempts.Should().BeGreaterThan(1, + "ReceiveAttempts must advance on each quorum redelivery on RabbitMQ 4.3, so it must exceed 1 " + + "after at least one redelivery"); + } + finally + { + harness.GetSignal().ThrowOnHandle = null; + await harness.DisposeAsync(); + } + } + + // Shared body for the quorum and classic max-receives facts: a handler that always throws must drive the + // message to the dead-letter queue once MaxReceiveAttempts is exhausted, with the Receiver's own failure + // headers merged onto the republished envelope. + private async Task RunMaxReceivesScenarioAsync(QueueType queueType, Func commandFactory) + where TMessage : class, ICommand + { + var set = RabbitMqTopology.CreateSet("v43_maxreceives", queueType); + await RabbitMqTopology.DeclareAsync(_fixture.GetAmqpConnectionString(), set, CancellationToken.None); + + var harness = ChatterRabbitMqPipelineHarness.Build( + _fixture.GetAmqpConnectionString(), + queueType, + rmq => rmq.AddQueueReceiver( + set.WorkQueueName, + deadLetterQueuePath: set.DeadLetterQueueName, + maxReceiveAttempts: MaxReceiveAttempts), + typeof(TMessage)); + try + { + await harness.StartAsync(); + + harness.GetSignal().ThrowOnHandle = + () => new InvalidOperationException("v43-maxreceives-test forced throw"); + + await harness.SendToQueueAsync(commandFactory(), set.WorkQueueName); + + await harness.WaitForHandledAsync(DeadLetterWait); + + var deadLettered = await DeadLetterQueueReader.ReceiveAsync( + _fixture.GetAmqpConnectionString(), set.DeadLetterQueueName, DeadLetterWait); + + deadLettered.Should().NotBeNull( + $"a handler that always throws must exhaust maxReceiveAttempts ({MaxReceiveAttempts}) and be " + + $"republished to the dead-letter queue within {DeadLetterWait}"); + + deadLettered.Headers.Should().ContainKey(MessageContext.FailureDetails, + "the Receiver's dead-letter republish merges FailureDetails, which a broker-side drop or " + + "dead-letter would not carry"); + + deadLettered.Headers.Should().ContainKey(MessageContext.FailureDescription); + deadLettered.Headers[MessageContext.FailureDescription] + .Should().Contain("v43-maxreceives-test forced throw", + "the dead-letter comes from the handler-failure path, not the poison path"); + } + finally + { + var signal = harness.GetSignal(); + signal.ThrowOnHandle = null; + + // Written here rather than before the assertions because DeadLetterQueueReader can throw + // TaskCanceledException at its deadline instead of returning null. + _output.WriteLine($"{queueType} queue: handler invoked {signal.InvocationCount} time(s)."); + WriteDeliveredAttempts(signal.Records.Select(record => record.Context)); + + await harness.DisposeAsync(); + } + } + + // Writes the ReceiveAttempts the Receiver resolved and the broker's redelivered flag for the first few + // deliveries. Diagnostic only: no assertion depends on it. + private void WriteDeliveredAttempts(IEnumerable contexts) + { + var deliveryNumber = 0; + foreach (var context in contexts.Take(DeliverySnapshotLimit)) + { + deliveryNumber++; + if (context is null || !context.Container.TryGet(out var received)) + { + _output.WriteLine($"delivery {deliveryNumber}: no ReceivedMessage in the broker context"); + continue; + } + + _output.WriteLine( + $"delivery {deliveryNumber}: redelivered={received.Redelivered} " + + $"{MessageContext.ReceiveAttempts}={DescribeReceiveAttempts(context)}"); + } + } + + private static string DescribeReceiveAttempts(IMessageBrokerContext context) + { + var messageContext = context.BrokeredMessage?.MessageContext; + if (messageContext is null || !messageContext.TryGetValue(MessageContext.ReceiveAttempts, out var value)) + { + return "absent"; + } + + return value is null ? "null" : $"{value}"; + } + } +} diff --git a/src/Chatter.MessageBrokers.RabbitMQ/tests/Integration/RabbitMqFixture.cs b/src/Chatter.MessageBrokers.RabbitMQ/tests/Integration/RabbitMqFixture.cs index 3760e699..92804d80 100644 --- a/src/Chatter.MessageBrokers.RabbitMQ/tests/Integration/RabbitMqFixture.cs +++ b/src/Chatter.MessageBrokers.RabbitMQ/tests/Integration/RabbitMqFixture.cs @@ -20,7 +20,10 @@ public sealed class RabbitMqFixture : IAsyncLifetime // The RabbitMQ image is passed explicitly to RabbitMqBuilder's image ctor (the parameterless ctor is // obsolete and pins an old tag), mirroring how the SQL Service Broker fixture pins its image. A 3.13 // image supports BOTH quorum queues (native x-delivery-count, introduced in 3.8) and classic queues, - // which the deadletter scenario proves on both per ADR-0001. + // which the deadletter scenario proves on both per ADR-0001. This fixture stays pinned to 3.13 as the + // broad-surface broker for the rest of this collection's scenarios; on 3.13, reject and nack advance + // x-delivery-count identically (docs/adr/0042), so it cannot pin 4.3's divergent delivery-count + // semantics — RabbitMq43Fixture and RabbitMqDeliveryCountingOn43Tests cover those separately. private const string RabbitMqImage = "rabbitmq:3.13-management"; // The container-internal management API port. Testcontainers maps this to a RANDOM host port (NOT 15672), diff --git a/src/Chatter.MessageBrokers.RabbitMQ/tests/Integration/RabbitMqNackRedeliveryTests.cs b/src/Chatter.MessageBrokers.RabbitMQ/tests/Integration/RabbitMqNackRedeliveryTests.cs index 1edaefd2..e323e413 100644 --- a/src/Chatter.MessageBrokers.RabbitMQ/tests/Integration/RabbitMqNackRedeliveryTests.cs +++ b/src/Chatter.MessageBrokers.RabbitMQ/tests/Integration/RabbitMqNackRedeliveryTests.cs @@ -13,12 +13,15 @@ namespace Chatter.MessageBrokers.RabbitMQ.Tests.Integration { // Nack→redelivery integration proof for the RabbitMQ integration harness. The SYSTEM UNDER TEST is - // Chatter's nack path on a Quorum queue: when RecordingMessageHandler throws and ReceiveAttempts has NOT - // reached MaxReceiveAttempts, BrokeredMessageReceiver routes to RabbitMqReceiver.NackMessageAsync, which on - // a quorum queue issues BasicNack(requeue: true) so the broker increments the native x-delivery-count and - // redelivers. The test asserts (a) the handler is invoked at least twice (proving redelivery happened), and - // (b) the ReceiveAttempts stamp climbs across deliveries (proving attempts = native x-delivery-count + 1 - // advances on each redelivery). Mirrors the SQL Service Broker SsbNackRedeliveryTests. + // Chatter's failed-delivery path on a Quorum queue: when RecordingMessageHandler throws and ReceiveAttempts + // has NOT reached MaxReceiveAttempts, BrokeredMessageReceiver routes to RabbitMqReceiver.NackMessageAsync, + // which on a quorum queue issues BasicReject(requeue: true) so the broker increments the native + // x-delivery-count and redelivers. The test asserts (a) the handler is invoked at least twice (proving + // redelivery happened), and (b) the ReceiveAttempts stamp climbs across deliveries (proving attempts = native + // x-delivery-count + 1 advances on each redelivery). Runs against RabbitMqFixture's pinned 3.13 broker, where + // reject behaves exactly as nack did before 4.3 (docs/adr/0042), so this proves the pre-4.3 half of the + // reject/nack compatibility claim; RabbitMqDeliveryCountingOn43Tests proves the 4.3 half. Mirrors the SQL + // Service Broker SsbNackRedeliveryTests. // // ANTI-INFINITE-LOOP: ThrowOnHandle is flipped to null as soon as >= 2 invocations are observed so the // message finally acks before DisposeAsync drains the pump. MaxReceiveAttempts is left at the default (10), @@ -50,9 +53,11 @@ public sealed class ClassicTtlRedeliveryCommand : ICommand public string Marker { get; set; } } - // Nack→redelivery: when the handler throws, NackMessageAsync issues BasicNack(requeue:true) on the - // quorum queue so the broker redelivers and increments x-delivery-count. Assert invocation count >= 2 - // (at least one redelivery) and that ReceiveAttempts climbs across successive deliveries. + // Nack→redelivery: when the handler throws, NackMessageAsync issues BasicReject(requeue:true) on the + // quorum queue so the broker redelivers and increments x-delivery-count. On this fixture's pinned 3.13 + // broker, reject behaves exactly as nack did before 4.3, so this pins the pre-4.3 half of the + // reject/nack compatibility claim (docs/adr/0042). Assert invocation count >= 2 (at least one + // redelivery) and that ReceiveAttempts climbs across successive deliveries. [RequiresDockerFact] public async Task ThrowingHandlerCausesRedeliveryAndClimbingReceiveAttempts() { diff --git a/src/Chatter.MessageBrokers.RabbitMQ/tests/Receiving/RecordingChannel.cs b/src/Chatter.MessageBrokers.RabbitMQ/tests/Receiving/RecordingChannel.cs index 2a09f710..c4a5fd42 100644 --- a/src/Chatter.MessageBrokers.RabbitMQ/tests/Receiving/RecordingChannel.cs +++ b/src/Chatter.MessageBrokers.RabbitMQ/tests/Receiving/RecordingChannel.cs @@ -8,7 +8,7 @@ namespace Chatter.MessageBrokers.RabbitMQ.Tests.Receiving { // A recording IChannel the in-memory connection source hands to RunOnReceiveChannelAsync and to publish - // rentals. It implements only the members RabbitMqReceiver/RabbitMqSender exercise — ack, nack, publish, + // rentals. It implements only the members RabbitMqReceiver/RabbitMqSender exercise — ack, nack, reject, publish, // consume-registration, QoS — and records each so tests assert off the recordings. Every other IChannel // member throws NotImplementedException: reaching one is a signal the production code took an untested // path, which a test should surface rather than silently accept. Reports IsOpen == false so a rental's @@ -41,6 +41,7 @@ public RecordingChannel(OperationSequencer sequencer) public List Acks { get; } = new List(); public List Nacks { get; } = new List(); + public List Rejects { get; } = new List(); public List Publishes { get; } = new List(); public IAsyncBasicConsumer RegisteredConsumer { get; private set; } public ushort? LastQosPrefetchCount { get; private set; } @@ -70,6 +71,12 @@ public ValueTask BasicNackAsync(ulong deliveryTag, bool multiple, bool requeue, return default; } + public ValueTask BasicRejectAsync(ulong deliveryTag, bool requeue, CancellationToken cancellationToken = default) + { + Rejects.Add(new RejectRecord(deliveryTag, requeue, _sequencer.Next())); + return default; + } + // Opt-in publish-fault seam: when non-null, BasicPublishAsync records the publish (so recording // assertions hold), then faults the returned task with this exception — modeling the broker's // 7.2.1 confirm-tracking fault on an unroutable mandatory basic.return. Default null = success. @@ -162,7 +169,6 @@ public Task BasicCancelAsync(string consumerTag, bool noWait = false, Cancellati return Task.CompletedTask; } public Task BasicGetAsync(string queue, bool autoAck, CancellationToken cancellationToken = default) => throw new NotImplementedException(); - public ValueTask BasicRejectAsync(ulong deliveryTag, bool requeue, CancellationToken cancellationToken = default) => throw new NotImplementedException(); public Task CloseAsync(ushort replyCode, string replyText, bool abort, CancellationToken cancellationToken = default) => throw new NotImplementedException(); public Task CloseAsync(ShutdownEventArgs reason, bool abort) => throw new NotImplementedException(); public Task CloseAsync(ShutdownEventArgs reason, bool abort, CancellationToken cancellationToken) => throw new NotImplementedException(); @@ -225,6 +231,21 @@ public NackRecord(ulong deliveryTag, bool multiple, bool requeue, long seq) public long Seq { get; } } + internal readonly struct RejectRecord + { + public RejectRecord(ulong deliveryTag, bool requeue, long seq) + { + DeliveryTag = deliveryTag; + Requeue = requeue; + Seq = seq; + } + + public ulong DeliveryTag { get; } + public bool Requeue { get; } + // The shared-sequencer tick at which this reject was recorded, so a test can assert settlement ordering. + public long Seq { get; } + } + internal sealed class PublishRecord { public PublishRecord(string exchange, diff --git a/src/Chatter.MessageBrokers.RabbitMQ/tests/Receiving/UsingRabbitMqReceiver/ReceiverHarness.cs b/src/Chatter.MessageBrokers.RabbitMQ/tests/Receiving/UsingRabbitMqReceiver/ReceiverHarness.cs index 9a56ee83..fbbf687d 100644 --- a/src/Chatter.MessageBrokers.RabbitMQ/tests/Receiving/UsingRabbitMqReceiver/ReceiverHarness.cs +++ b/src/Chatter.MessageBrokers.RabbitMQ/tests/Receiving/UsingRabbitMqReceiver/ReceiverHarness.cs @@ -24,6 +24,9 @@ internal sealed class ReceiverHarness // Native quorum-queue redelivery counter the broker increments per redelivery. public const string NativeDeliveryCountHeader = "x-delivery-count"; + // Native quorum-queue assignment counter RabbitMQ 4.3+ stamps on each redelivery. + public const string NativeAcquiredCountHeader = "x-acquired-count"; + public InMemoryRabbitMqConnectionSource ConnectionSource { get; } = new InMemoryRabbitMqConnectionSource(); public RabbitMqReceiver Receiver { get; } diff --git a/src/Chatter.MessageBrokers.RabbitMQ/tests/Receiving/UsingRabbitMqReceiver/WhenReceivingMessage.cs b/src/Chatter.MessageBrokers.RabbitMQ/tests/Receiving/UsingRabbitMqReceiver/WhenReceivingMessage.cs index 555b08a0..b683319a 100644 --- a/src/Chatter.MessageBrokers.RabbitMQ/tests/Receiving/UsingRabbitMqReceiver/WhenReceivingMessage.cs +++ b/src/Chatter.MessageBrokers.RabbitMQ/tests/Receiving/UsingRabbitMqReceiver/WhenReceivingMessage.cs @@ -242,6 +242,59 @@ public async Task MustStripNativeDeliveryCountHeaderWhilePreservingOtherHeaders( context.BrokeredMessage.MessageContext["x-correlation-id"].Should().Be("corr-456"); } + // REGRESSION: the RabbitMQ 4.3+ quorum assignment counter (x-acquired-count) is the same kind of receive + // state, so it must likewise NOT survive onto the emitted context. It is also not an attempt source: with no + // x-delivery-count on the delivery, ReceiveAttempts stays at the first-delivery floor of 1. + [Fact] + public async Task MustStripNativeAcquiredCountHeaderWhilePreservingOtherHeaders() + { + var harness = ReceiverHarness.Create(QueueType.Quorum); + var headers = new Dictionary + { + [ReceiverHarness.NativeAcquiredCountHeader] = 4L, + ["x-correlation-id"] = "corr-789" + }; + await harness.PushVerbatimAsync(deliveryTag: 1, headers: headers); + + var context = await harness.ReceiveAsync(); + + context.BrokeredMessage.MessageContext.Should().NotContainKey(ReceiverHarness.NativeAcquiredCountHeader, + "the native assignment counter must not ride the inbound context onto an outbound republish"); + context.BrokeredMessage.MessageContext[MessageContext.ReceiveAttempts].Should().Be(1, + "x-acquired-count counts assignments, not failed deliveries, so it is never read as the attempt count"); + context.BrokeredMessage.MessageContext["x-correlation-id"].Should().Be("corr-789"); + } + + // REGRESSION: the ORACLE for "x-acquired-count is never an attempt source". The acquired-count-ONLY fact + // above cannot pin that claim on its own: with x-delivery-count absent, a resolver that consulted + // x-acquired-count only WHEN x-delivery-count is present — the normal RabbitMQ 4.3 redelivery shape, where + // the broker stamps BOTH — still resolves 1 and stays green. This fact delivers both counters at + // DELIBERATELY DIVERGENT values, so ReceiveAttempts can only come out at 3 if the resolver read + // x-delivery-count alone; any read of x-acquired-count (as the source, as a max, or as a fallback) lands on + // 10 instead. Together with the acquired-count-only fact it covers both shapes a 4.3 delivery can take. + [Fact] + public async Task MustResolveReceiveAttemptsFromDeliveryCountWhenBothQuorumCountersArePresent() + { + var harness = ReceiverHarness.Create(QueueType.Quorum); + var headers = new Dictionary + { + [ReceiverHarness.NativeDeliveryCountHeader] = 2L, + [ReceiverHarness.NativeAcquiredCountHeader] = 9L, + ["x-correlation-id"] = "corr-both" + }; + await harness.PushVerbatimAsync(deliveryTag: 1, headers: headers); + + var context = await harness.ReceiveAsync(); + + context.BrokeredMessage.MessageContext[MessageContext.ReceiveAttempts].Should().Be(3, + "attempts derive solely from x-delivery-count (2 prior + 1); reading x-acquired-count would give 10"); + context.BrokeredMessage.MessageContext.Should().NotContainKey(ReceiverHarness.NativeDeliveryCountHeader, + "the native delivery counter must not ride the inbound context onto an outbound republish"); + context.BrokeredMessage.MessageContext.Should().NotContainKey(ReceiverHarness.NativeAcquiredCountHeader, + "the native assignment counter must not ride the inbound context onto an outbound republish"); + context.BrokeredMessage.MessageContext["x-correlation-id"].Should().Be("corr-both"); + } + // P1 REPRODUCTION: a real broker delivers the string CorrelationId application header as an AMQP longstr // (byte[]). The core's InboundBrokeredMessage reads MessageContext.CorrelationId as a string by kind test at // construction (which MessageBrokerContext does inside ReceiveMessageAsync), so an undecoded byte[] reads as diff --git a/src/Chatter.MessageBrokers.RabbitMQ/tests/Receiving/UsingRabbitMqReceiver/WhenSettlingMessage.cs b/src/Chatter.MessageBrokers.RabbitMQ/tests/Receiving/UsingRabbitMqReceiver/WhenSettlingMessage.cs index 6dd3dc66..ae944594 100644 --- a/src/Chatter.MessageBrokers.RabbitMQ/tests/Receiving/UsingRabbitMqReceiver/WhenSettlingMessage.cs +++ b/src/Chatter.MessageBrokers.RabbitMQ/tests/Receiving/UsingRabbitMqReceiver/WhenSettlingMessage.cs @@ -14,8 +14,8 @@ namespace Chatter.MessageBrokers.RabbitMQ.Tests.Receiving.UsingRabbitMqReceiver { // Pins RabbitMqReceiver settlement against the three-valued settlement contract of // IMessagingInfrastructureReceiver (ADR-0010 D7: every in-repo implementation is contract-tested against all - // three outcomes): ack on a matching epoch acks the carried delivery tag and reports Settled; nack requeues - // (Quorum) or republishes-with-incremented-count-then-acks (Classic) and reports Settled; deadletter + // three outcomes): ack on a matching epoch acks the carried delivery tag and reports Settled; nack requeues via + // basic.reject (Quorum) or republishes-with-incremented-count-then-acks (Classic) and reports Settled; deadletter // republishes-confirmed to the attribute-declared deadletter/error path then acks and reports Settled on BOTH // paths. A settlement with nothing to settle (TransactionMode.None at-most-once) reports NotRequired; a // settlement that was attempted and did not happen (no carried delivery, or a stale channel epoch) reports @@ -87,6 +87,7 @@ public async Task MustReportFailedOnNackWhenContextCarriesNoReceivedMessage() nacked.Outcome.Should().Be(SettlementOutcome.Failed, "a negative acknowledgement that could not locate its delivery was attempted and did not happen"); harness.ConnectionSource.ReceiveChannel.Nacks.Should().BeEmpty("no carried delivery means nothing to nack"); + harness.ConnectionSource.ReceiveChannel.Rejects.Should().BeEmpty("no carried delivery means nothing to reject"); harness.ConnectionSource.PublishChannels.Should().BeEmpty("no carried delivery means nothing to republish"); } @@ -107,10 +108,12 @@ public async Task MustReportFailedOnDeadletterWhenContextCarriesNoReceivedMessag harness.ConnectionSource.ReceiveChannel.Acks.Should().BeEmpty("no carried delivery means nothing to ack"); } - // --- nack: Quorum requeues natively --- + // --- nack: Quorum requeues natively via basic.reject --- + // The quorum requeue verb is basic.reject, never basic.nack: on RabbitMQ 4.3+ only reject records a failed + // delivery attempt (advancing x-delivery-count toward the queue's delivery-limit); nack means "returned". [Fact] - public async Task MustRequeueOnNackForQuorum() + public async Task MustRejectWithRequeueOnNackForQuorum() { var harness = ReceiverHarness.Create(QueueType.Quorum); await harness.PushAsync(deliveryTag: 5); @@ -119,14 +122,15 @@ public async Task MustRequeueOnNackForQuorum() var nacked = await harness.Receiver.NackMessageAsync(context, transactionContext: null, CancellationToken.None); nacked.Outcome.Should().Be(SettlementOutcome.Settled); - var nack = harness.ConnectionSource.ReceiveChannel.Nacks.Single(); - nack.DeliveryTag.Should().Be(5UL); - nack.Requeue.Should().BeTrue(); + var reject = harness.ConnectionSource.ReceiveChannel.Rejects.Single(); + reject.DeliveryTag.Should().Be(5UL); + reject.Requeue.Should().BeTrue(); + harness.ConnectionSource.ReceiveChannel.Nacks.Should().BeEmpty("on RabbitMQ 4.3+ basic.nack does not count as a failed delivery attempt"); harness.ConnectionSource.ReceiveChannel.Acks.Should().BeEmpty("a quorum nack requeues; it does not ack"); } [Fact] - public async Task MustNotRequeueOnNackForQuorumWhenEpochStale() + public async Task MustNotRejectOnNackForQuorumWhenEpochStale() { var harness = ReceiverHarness.Create(QueueType.Quorum); await harness.PushAsync(deliveryTag: 5); @@ -137,6 +141,7 @@ public async Task MustNotRequeueOnNackForQuorumWhenEpochStale() var nacked = await harness.Receiver.NackMessageAsync(context, transactionContext: null, CancellationToken.None); nacked.Outcome.Should().Be(SettlementOutcome.Failed); + harness.ConnectionSource.ReceiveChannel.Rejects.Should().BeEmpty(); harness.ConnectionSource.ReceiveChannel.Nacks.Should().BeEmpty(); } @@ -218,6 +223,72 @@ public async Task MustRepublishToDeadLetterPathThenAck() "the confirmed deadletter republish must be recorded before the original ack (ADR-0001 confirm-before-ack)"); } + // --- native quorum counters are broker-owned: consumed at receive, never carried onto a republished copy --- + + // x-delivery-count and x-acquired-count are the broker's own counters for the delivery it just made. The + // Receiver reads the attempt count off them at the receive boundary and carries neither onward, so the copy + // it republishes to the dead-letter queue is a new message that carries no stale broker count. + [Fact] + public async Task MustNotRepublishNativeQuorumCountersOnTheDeadletterHop() + { + var harness = ReceiverHarness.Create(QueueType.Quorum, deadLetterQueuePath: ReceiverHarness.DeadLetterPath); + await harness.PushAsync(deliveryTag: 11, headers: DeliveryHeadersCarryingNativeQuorumCounters()); + var context = await harness.ReceiveAsync(); + + await harness.Receiver.DeadletterMessageAsync(context, transactionContext: null, "poisoned", "bad", CancellationToken.None); + + var republish = harness.ConnectionSource.PublishChannels.Single().Publishes.Single(); + republish.Headers.Should().NotContainKeys(new[] { ReceiverHarness.NativeDeliveryCountHeader, ReceiverHarness.NativeAcquiredCountHeader }, + "the native quorum counters are consumed at the receive boundary and never ride onto the dead-letter copy"); + republish.Headers.Should().ContainKey("app-header", "only the broker-owned counters are withheld; application headers are carried"); + } + + [Fact] + public async Task MustNotRepublishNativeQuorumCountersOnTheErrorQueueHop() + { + var harness = ReceiverHarness.Create(QueueType.Quorum, deadLetterQueuePath: null, errorQueuePath: ReceiverHarness.ErrorPath); + await harness.PushAsync(deliveryTag: 12, headers: DeliveryHeadersCarryingNativeQuorumCounters()); + var context = await harness.ReceiveAsync(); + + await harness.Receiver.DeadletterMessageAsync(context, transactionContext: null, "poisoned", "bad", CancellationToken.None); + + var republish = harness.ConnectionSource.PublishChannels.Single().Publishes.Single(); + republish.Headers.Should().NotContainKeys(new[] { ReceiverHarness.NativeDeliveryCountHeader, ReceiverHarness.NativeAcquiredCountHeader }, + "the native quorum counters are consumed at the receive boundary and never ride onto the error-queue copy"); + } + + // The classic redelivery hop re-stamps the adapter-owned x-chatter-delivery-count and must still withhold any + // native counter the delivery arrived with, so the delivery and its redelivered copy count only by the + // adapter's own header. + [Fact] + public async Task MustNotRepublishNativeQuorumCountersOnTheClassicRedeliveryHop() + { + var harness = ReceiverHarness.Create(QueueType.Classic); + var headers = DeliveryHeadersCarryingNativeQuorumCounters(); + headers[RabbitMqMessageContext.DeliveryCountHeader] = 1L; + await harness.PushAsync(deliveryTag: 9, headers: headers); + var context = await harness.ReceiveAsync(); + + context.BrokeredMessage.MessageContext[MessageContext.ReceiveAttempts].Should().Be(2, + "a classic queue counts attempts from x-chatter-delivery-count (1 prior + 1) and never from a native counter"); + + await harness.Receiver.NackMessageAsync(context, transactionContext: null, CancellationToken.None); + + var republish = harness.ConnectionSource.PublishChannels.Single().Publishes.Single(); + republish.Headers.Should().NotContainKeys(new[] { ReceiverHarness.NativeDeliveryCountHeader, ReceiverHarness.NativeAcquiredCountHeader }, + "the native quorum counters are consumed at the receive boundary and never ride onto the redelivered copy"); + republish.Headers[RabbitMqMessageContext.DeliveryCountHeader].Should().Be(2L, + "the adapter-owned counter is carried and re-stamped on its own republish"); + } + + private static Dictionary DeliveryHeadersCarryingNativeQuorumCounters() + => new Dictionary + { + [ReceiverHarness.NativeDeliveryCountHeader] = 2L, + [ReceiverHarness.NativeAcquiredCountHeader] = 7L, + ["app-header"] = "kept" + }; + // OWNERSHIP (r3408649034): the dead-letter queue is the ADAPTER's responsibility; the ERROR queue is the // CORE's. On max-receives the core runs deadletter FIRST and, ONLY when the deadletter SETTLED and the // infrastructure does NOT write to the error queue itself, ALSO runs its error-recovery action @@ -244,6 +315,7 @@ public async Task MustRepublishToErrorPathThenAckWhenNoDeadLetterPathConfigured( // republish (NOT a requeue/republish-back), so it is not redelivered. harness.ConnectionSource.ReceiveChannel.Acks.Single().DeliveryTag.Should().Be(12UL); harness.ConnectionSource.ReceiveChannel.Nacks.Should().BeEmpty("the error-only deadletter acks; it does not requeue"); + harness.ConnectionSource.ReceiveChannel.Rejects.Should().BeEmpty("the error-only deadletter acks; it does not requeue"); // (c) The outcome is TRUTHFUL: the delivery WAS settled. Suppressing the core's error action is a // separate control signal (WritesToErrorQueue), never a misreported settlement outcome. deadlettered.Outcome.Should().Be(SettlementOutcome.Settled, @@ -439,6 +511,7 @@ public async Task MustNoOpOnNackForQuorumWhenTransactionModeNone() nacked.Outcome.Should().Be(SettlementOutcome.NotRequired, "under None the broker already removed the delivery at receive (autoAck); there is nothing to settle"); harness.ConnectionSource.ReceiveChannel.Acks.Should().BeEmpty("no manual delivery tag exists to ack under autoAck"); harness.ConnectionSource.ReceiveChannel.Nacks.Should().BeEmpty("at-most-once must not requeue on failure"); + harness.ConnectionSource.ReceiveChannel.Rejects.Should().BeEmpty("at-most-once must not requeue on failure"); } // Under None a classic-queue nack likewise is a no-op: no manual ack and no republish-with-incremented-count