#3867 a batched handler participates in partitioned sequential processing

[jeremydmiller]

Closes #3867. Driver: CritterWatch#949.

A batched handler could not participate in PartitionProcessingByGroupId sequential ordering. Two independent causes, both fixed here.

Stated plainly, because it surprised two separate reviewers: GlobalPartitioned did not give you a single writer per group id if any participating message type was batched. The affected CritterWatch deployment already ran GlobalPartitioned with 5 slots and still saw ~20 stream-concurrency exceptions/min, the batched type accounting for 59% of them.

Part 1 — the batch envelope carried no group id

And a null group id means a random slot, not "no partitioning":

// PartitionedMessagingExtensions.SlotForProcessing
var groupId = rules.DetermineGroupId(envelope);
if (groupId == null) return Random.Shared.Next(1, numberOfSlots) - 1;   // <-- random, not "unpartitioned"

DefaultMessageBatcher<T> groups only by TenantId, so a batch spans group ids and has none of its own. BatchingOptions.GroupByGroupId() opts into a batcher that groups by (TenantId, GroupId) and stamps that id onto every batch envelope.

Deliberate behaviors:

• The key is (tenant, group), never group alone. Members settle against the batch envelope, so merging tenants would lose the tenant each member arrived under.
• Ungroupable envelopes batch together and stay ungrouped. The batcher does not invent an identity.
• Rules are injected, not resolved. MessagePartitioningRules isn't available when BatchingOptions is configured, so ProcessorBuilder.Build supplies it via the internal IRequirePartitioningRules.

Part 2 — the batch executed on the wrong queue

BatchingProcessor.processEnvelopes hardcoded grouped.Destination = Queue.Uri, so the assembled batch always ran on one dedicated local queue — while the unbatched handlers for the same group id ran on a topology slot. Different execution blocks, so the batch raced the very handlers the topology had just sequenced.

This implements shape 3: BatchingOptions targets the partitioned topology its element type already belongs to, and the slot comes from the batch's group id.

• resolveBatchExecutionTopologies() at bootstrap records the slot endpoints for any batched element type matching a GlobalPartitioned (companion local queues) or PublishToPartitionedLocalMessaging topology.
• IBatchExecutionQueues selects the queue per assembled batch, so ProcessorBuilder.Build's single ILocalQueue resolve becomes N. PartitionedBatchExecutionQueues uses SlotForSending — the same hash GlobalPartitionedRoute and PartitionedMessageTopology.SelectSlot use, so the batch agrees with where its group's other messages went. (SlotForSending and SlotForProcessing deliberately use different hashes; getting this wrong would silently reintroduce the race, so there's a test pinning it.)
• The built-in batcher is swapped for GroupIdMessageBatcher, since slotting requires a batch to belong to exactly one group. An application-supplied batcher is left alone, and any batch it emits without a group id falls back to the dedicated queue rather than drawing a random slot.
• Opt out with ExecuteOnDedicatedLocalQueue(), or by naming LocalExecutionQueueName. Wolverine's own default assignment goes through an internal SetDefaultLocalExecutionQueueName so it doesn't read as a user choice.

Automatic rather than opt-in because GlobalPartitionLocalQueueUri is internal — users can't wire this themselves — and because an app that declared GlobalPartitioned for a message type already stated the intent the batch was silently exempting itself from.

The deadlock this had to handle

The note on the issue was right that there's no direct self-deadlock: HandleAsync runs on the slot block but only posts into _batchingBlock, while processEnvelopes runs on a separate _processingBlock. The head-of-line hazard flagged alongside it is real though, and it's worse than a stall — it closes a cycle across three bounded buffers:

slot block (bounded, DOP 1) → BatchingProcessor.HandleAsync
  → BatchingChannel._inner (bounded) → addItem → _processingBlock (bounded)
  → processEnvelopes → queue.EnqueueAsync → back into the same slot block

Saturate all three and every worker in the ring is blocked on the next. This didn't exist before, because the batch's dedicated local queue is unbounded (GH-3287).

Endpoint.HostsBatchExecution is set on every slot endpoint a batch targets, and DurableReceiver gives those an unbounded execution block. EnqueueAsync into an unbounded slot never blocks, so _processingBlock never stalls and the cycle can't close — which also removes the cross-group head-of-line coupling, without the drop risk of a non-blocking Post. Back-pressure is preserved by BatchingPendingCounts, which counts members against the originating external listener. BufferedReceiver already passes unbounded for local queues, so only the durable path needed changing.

The two other checks that were asked for

• Does the batch chain resolve on the companion queues? Yes. ExecutorFor(T[], slot) falls through to the default chain; and for Separated mode with sticky Handle(T[]) handlers, HandlerGraph.HandlerFor already special-cases a local queue with UsedInShardedTopology and builds a fanout.
• Does BatchingPendingCounts.SettleBatch still fire once per batch? Yes, by construction. Both DurableReceiver.CompleteAsync and BufferedReceiver's channel callback settle on envelope.Batch != null, keyed off the batch envelope and independent of which queue it landed on. Each grouped envelope still goes to exactly one queue — this is slot selection, not fan-out — so no double-count and no lost settle.

Also fixed

BatchReplay.EnqueueReducedBatchAsync copied Destination, MessageType and TenantId but not GroupId, so a ProbeIndividuallyAfter or ApplyItemException probe lost the batch's identity and scattered the survivors across slots. Already wrong with part 1 alone.

What this does NOT cover

Configuration	Unbatched handlers run on	Covered?
GlobalPartitioned	companion local queue global-{base}{slot}	yes
PublishToPartitionedLocalMessaging	local queue {base}{slot}	yes
Plain listener + PartitionProcessingByGroupId	the listener receiver's own ShardedExecutionBlock	no

The third row is not a queue and cannot be enqueued to. The docs say so rather than implying the composition is complete.

One more edge, noted rather than solved: under MultipleHandlerBehavior.Separated with multiple sticky Handle(T[]) handlers, the batch fans out from the slot queue to each sticky handler's own queue, so those handlers execute off-slot.

Tests

src/Testing/CoreTests/Acceptance/batching_with_partitioned_processing.cs is the pure-Wolverine acceptance test the issue asked for, and it needs no broker — ExplicitRouting already sends a batched element type to its topology slots, so PublishToPartitionedLocalMessaging reproduces the whole thing in memory. One batched and one unbatched message type share a group id; it asserts no intra-group overlap while still observing cross-group parallelism, so a fix that merely serialized everything would not pass.

It fails on main — 12 violations of exactly the described shape.

Also batch_execution_topology_resolution.cs (bootstrap resolution and both opt-outs), BatchExecutionQueuesTests (slot selection, including agreement with SlotForSending), BatchReplayTests (group id survives a probe), and the 5 existing GroupIdMessageBatcherTests.

Full CoreTests, net9.0:  2315 passed / 0 failed / 2 pre-existing skips
dotnet build wolverine.slnx -c Release -f net9.0:  clean, 0 warnings

Not yet run: the broker-backed global_partitioned_sharded_processing suites (RabbitMQ, Kafka, SQS, Postgres, SqlServer), which need docker compose up -d. The CritterWatch soak harness (./build.sh IngestSoak, many_services_many_nodes) remains the field acceptance test and has not been re-run.

Docs

New sections in both guides, which previously didn't mention each other: "Batching by group id" and "Batching inside a partitioned topology" in docs/guide/handlers/batching.md, and "Partitioned Processing with Batched Handlers" in docs/guide/messaging/partitioning.md. Both spell out the random-slot behavior explicitly and state the uncovered case above.

GH-3867-BATCHING-PARTITIONING-HANDOFF.md is updated to a record of what was built and what is still open.

🤖 Generated with Claude Code

https://claude.ai/code/session_01JKfy5EzLX1i149gjUb3Tfg