Akka.Streams: fix a possible lost wakeup in MergeHub (missing memory fence)

[Aaronontheweb]

The race

MergeHub.HubLogic coordinates one consumer (the hub's stage) with many producers through a ConcurrentQueue and a _needWakeup flag:

• Consumer (TryProcessNext): on an empty queue it stores _needWakeup = true, re-polls the queue once, then stops.
• Producer (Enqueue): _queue.Enqueue(e), then if (_needWakeup) { _needWakeup = false; _wakeupCallback(); }.

This is a Dekker-style store-then-load handshake on both sides, so it needs sequential consistency. JVM Akka gets that from @volatile. In .NET, _needWakeup was a plain bool, and nothing between the store and the re-poll acts as a full fence. I decompiled ConcurrentQueueSegment<T>.TryDequeue from the .NET 10 CoreLib: the empty path does Volatile.Read(Head), Volatile.Read(slot.SequenceNumber), Volatile.Read(Tail) and returns false, with no Interlocked operation. Acquire reads don't stop an earlier store from moving past them, and x86-TSO allows exactly this store-load reordering (ARM64 allows more). That makes this interleaving legal:

1. Consumer: _needWakeup = true (still in its store buffer); re-poll reads Tail == Head, so the queue looks empty and the consumer stops.
2. Producer: the Tail CAS publishes the element (full fence); it reads _needWakeup == false and sends no wakeup.
3. The consumer's store drains. The flag is true and an element is queued, but nobody is scheduled to read it.

The hub stalls until another producer enqueues. With perProducerBufferSize = 1, every producer is waiting on demand from the hub, so the stall is permanent.

PostStop has the same shape: _shuttingDown = true, then drain the queue. A producer in SinkLogic.PreStart enqueues Register and re-checks IsShuttingDown. If the store and the drain reorder, the producer can miss both the drain and the flag and wait forever.

C# volatile alone would not fix this. A volatile store followed by a volatile load can still reorder, so we need a full fence.

Evidence

A local-only stress harness (not committed) ran repeated MergeHub.Source<int>(1) streams for 45 s per case, with a 5 s hang timeout per iteration, on 8-core x86-64 with .NET 10:

	2 producers x 5000	16 producers x 500
dev (3 runs)	8 hangs / 8,429 iterations	1 hang / 18,469 iterations
this PR (4 runs)	0 / 16,577	0 / 24,433

That is 9 hangs in about 26,900 iterations before, and 0 in about 41,000 after. At the old rate we would expect about 14 hangs in the post-fix runs.

Fix (Hub.cs only)

• Interlocked.MemoryBarrier() right after _needWakeup = true, before the re-poll.
• Interlocked.MemoryBarrier() right after _shuttingDown = true in PostStop.
• _needWakeup and _shuttingDown are now volatile, which matches upstream and also safely publishes _wakeupCallback (set in PreStart) to producers that see _needWakeup == true.
• One line added to the existing proof comment: the proof assumes sequentially consistent accesses. In .NET that takes the consumer-side fence; on the producer side, the CAS inside ConcurrentQueue.Enqueue already acts as a fence.

The fence only runs when the queue is empty, not on the per-element path.

Tests

• Un-skipped MergeHub_must_work_with_long_streams_when_buffer_size_is_1 (it was skipped as "Very racy") and gave it an explicit WaitAsync timeout. 30/30 passes locally with the fix. It also passed 30/30 on dev, so it guards against gross regressions; it is not a reliable repro of this race. The window is too small for a single run.
• The full HubSpec passed 5/5 runs.