GH-4167: one BatchingProcessor per batched message type, not one per racing caller

[jeremydmiller]

Found while verifying the JasperFx fix for #4167 (JasperFx/jasperfx#714). This is a live bug on main today — it does not depend on that fix.

The bug

BatchingOptions.BuildHandler is an unguarded lazy init, and it runs on the message-handling path, not at bootstrap:

if (_handler != null) return _handler;
_handler = builder.Build(runtime, Batcher, this);

Callers reach it through HandlerPipeline's LightweightCache<Type, IExecutor>, whose indexer does not lock — two concurrent misses each invoke the factory and each returns its own instance.

For a stateless executor that is harmless, which is why the cache is fine everywhere else. A BatchingProcessor is not stateless: each instance owns a separate BatchingChannel buffer, its own flush Timer, and two Blocks with live worker tasks.

Instrumented — two instances on two threads, and a two-member batch arrives as two batches:

BuildHandler BUILT instance 11865849 for ExpiryItem on tid=24
BuildHandler BUILT instance 34717384 for ExpiryItem on tid=19
TriggerBatch count=1 tid=21
TriggerBatch count=1 tid=6
→ 2 batch(es): [one] | [two]

Two consequences:

1. Batches silently fragment — members of one logical batch are split across buffers and flushed separately.
2. The loser leaks. It is never handed back to any caller, so nothing disposes it; its Timer and worker tasks live for the process lifetime.

The fix

Double-checked lock, with _handler made volatile for the unlocked read.

Tests

concurrent_BuildHandler_yields_one_shared_processor races 8 callers into BuildHandler and asserts reference equality. It fails on every run without the fix, against the currently referenced JasperFx 2.56.0 — deliberately written to exercise the race directly rather than through message publishing, so it does not depend on the JasperFx release to be meaningful.

The end-to-end companion (concurrent_first_messages_still_assemble_a_single_batch) still passes on 2.56.0 even unfixed, because inline Block continuations serialize the first messages. It becomes a real guard after the JasperFx bump; the comment says so.

Also: Bug_3399's wait condition

A latent test race the same inline execution was hiding. A WaitFor* condition replaces quiescence rather than adding to it, so WaitForMessageToBeReceivedAt released the session at receipt, before any handler ran, and the assertion raced it. Probed: Batched=0 immediately, Batched=1 after 1s — the handler runs, the test just asked the wrong question.

Now waits for execution, of both separated chains for the batched array; waiting on one only trades a receive/execute race for a which-chain-won race.

Verification

• CoreTests 2664 passed / 2 skipped / 0 failed against a locally built JasperFx carrying sp_MSdropconstraints is not available on Azure SQL Database #714
• CoreTests green on stock JasperFx 2.56.0
• wolverine.slnx clean under -c Release -f net9.0

Follow-up

The GH-4167 local-queue reproduction test is not in this PR — it asserts the publisher thread never runs the handler, which cannot pass until JasperFx ships #714. It should land with the JasperFx version bump, as that bump's acceptance test.

🤖 Generated with Claude Code