GH-3708: EndpointMode.NativeAck — partitioned, parallel processing with native broker acks

[jeremydmiller]

Completes #3708 (steps 3-6 of the build order; step 1-2 landed in #4032). Also closes #4011.

Fills the empty cell in the mode matrix: Buffered's throughput and partitioning with Inline's no-loss guarantee, and no database involvement.

	Broker ack timing	Loss window	Parallelism	Group partitioning	DB cost
Inline	after handler success	none	ListenerCount only	none	none
NativeAck	after handler success	none	MaximumParallelMessages	✔️	none
BufferedInMemory	at receipt, before the handler	crash loses buffered messages	MaximumParallelMessages	✔️	none
Durable	after the inbox insert	none	MaximumParallelMessages	✔️	inbox insert + mark-handled

opts.ListenToRabbitQueue("webhooks")
    .ProcessInParallelWithNativeAcks()
    .PartitionProcessingByGroupId(PartitionSlots.Five)
    .MaximumParallelMessages(10);

NativeAckReceiver

BufferedReceiver's execution block with InlineReceiver's channel wiring. The single line that defines the mode is the absence of a _completeBlock.PostAsync() on receipt — the delivery stays unacknowledged until the pipeline settles it.

Three decisions worth review attention:

• The channel is resolved per envelope, via GH-4010: resolve the deserialize stage's channel callback per envelope #4013's DeserializeFirst(pipeline, runtime, channelSource) overload rather than one bound IChannelCallback. With ListenerCount > 1 the receiver is shared across listeners, so a single bound channel would ack the wrong delivery. There is a test for it.
• No ISupportNativeScheduling / ISupportDeadLetterQueue on the receiver. The channel is the listener, so those are the listener's native capabilities, exactly as for InlineReceiver. Implementing them here would route a terminal through in-memory state a crash would lose — which is the guarantee this mode exists to provide.
• _completeBlock survives, but only for already-expired deliveries. Something no handler will ever process still has to be acked so the broker stops redelivering it.

RabbitMQ opts in

RabbitMqQueue is the first transport to override supportsNativeAck, and it qualifies on both counts the mode requires:

• Deliveries settle individually. RabbitMQ: settle interop-DLQ deliveries and move to per-message acks (multiple: false) #3706 made every ack multiple: false, which is load-bearing here — under multiple: true an out-of-order completion would silently ack every lower delivery tag still in flight, turning the no-loss guarantee into silent loss.
• Completion-time acks are already safe from arbitrary threads. NullReferenceException acking a RabbitMQ delivery on a torn-down channel #3687's DeliveredOn / CanSettle plumbing refuses to settle against a stale or closed channel.

Queues only, not exchanges or topics — native acks are a listening concept.

ProcessInParallelWithNativeAcks()

Pulled forward from step 6 into step 5 out of necessity: without it the opt-in is unreachable. There is no fluent Mode() on listener configuration, and the first cut of the RabbitMQ test cast through IRabbitMqQueue to set Mode directly — which documents a missing API rather than a working one.

It throws eagerly for a local queue, matching ProcessInline()'s guard from #4027. The lazily-sourced LocalQueueConfiguration(Func<LocalQueue>) path that guard cannot see is already covered: LocalQueue does not override supportsNativeAck, so the Mode setter rejects it with a real message.

GH-4011

ListeningAgent.EnqueueDirectlyAsync is a type-switch ending in a throwing fallthrough. Every receiver that existed before this PR had a branch, so the fallthrough was unreachable; NativeAckReceiver made it reachable. This is the durability agent's re-entry point — DLQ replay (#1942), scheduled-message firing — so without a branch, a NativeAck endpoint in an application that also has persistence configured threw on any replay targeting it. Durable outbox for sending plus native acks on one flooding listener is a legitimate combination, not an exotic one.

The test was red-baselined against the exact InvalidOperationException("There is no active, local queue ...") before the branch was added, and asserts the receiver really is a NativeAckReceiver so it cannot pass vacuously through the BufferedReceiver branch.

Repaying the GH-3712 debt

#4017 landed a bootstrap validator and a per-mode settings matrix written for a three-mode world; both were wrong the moment a fourth mode existed.

• The validator's fatal message for Inline + PartitionProcessingByGroupId() now points at ProcessInParallelWithNativeAcks() — the mode that exists for exactly the combination the user was reaching for — instead of only offering the two modes that mean giving up native acks. The CoreTests assertion pinning that text is updated; it is precisely the debt GH-3712: validate or warn on silently-ignored listener configuration combos #4017 predicted.
• ServerlessEndpointsMustBeInlinePolicy now warns before downgrading a NativeAck endpoint. Coercion is correct for Serverless, but partitioned processing is a guarantee the user asked for, and dropping it in silence is the exact failure mode Validate or warn on silently-ignored listener configuration combos (Inline + partitioning/parallelism) #3712 exists to prevent.
• The listener docs gain a NativeAck column and a section for the mode.

The guarantee, stated exactly

Protection against intra-group concurrency is the hard guarantee. Strict sequential processing in original delivery order is not. A failed or redelivered message re-enters its lane later, never concurrently. That is the honest contract for native-ack retry semantics on every broker; users needing strict order under failure keep the durable inbox. This is in the docs in those words.

Also documented rather than left to be discovered: a rolling deploy produces duplicate deliveries bounded by prefetch depth. Graceful drain processes what it can within the drain timeout; whatever it cannot is never settled and gets redelivered. Handlers should already be idempotent under at-least-once, but the number is not zero. Quantifying it under chaos is #3713.

Verification

• dotnet build wolverine.slnx -c Release -f net9.0 — clean, 0 warnings, 0 errors
• CoreTests — 2539 total, 0 failed, 2 skipped
• RabbitMQ tests run against a real broker, including the one that matters: a node dies mid-flight having acked nothing, and a fresh node picks up all five messages on redelivery. Under BufferedInMemory those five are gone.

That redelivery test initially "failed" by proving itself — its redelivered messages showed up in a different test's assertion, because the tracking bag is static and all three tests shared one queue while the broker outlives the host. Each test now owns its queue.

Not in this PR

#3715 (NativeAck for other transports — ASB, SQS, Pub/Sub, NATS JetStream, Redis Streams and Pulsar are all natural fits), #3713 (the 5-node chaos reproduction), and #3709, whose premise needs re-scoping against what actually landed here.

🤖 Generated with Claude Code