fix(#568): close the WaitForShardState race against an already-published state

[jeremydmiller]

Closes #568.

The bug

ShardStateTracker.WaitForShardState could miss the very state it was waiting for and then burn its full timeout (1 minute by default), even though the state had been published before the wait was ever set up.

Publication is asynchronous: PublishAsync posts to a Block<ShardState> and returns, and the consumer thread later walks the listener list as it stood when that delivery started. So between WaitForShardState's check of the state map and ShardStatusWatcher's subscription, the consumer can deliver the state to the listeners that exist at that moment. The watcher subscribes a moment too late, and since it only ever inspected states arriving at OnNext — never re-reading the map — the state is simply gone. If nothing else is published for that shard, the wait sits until the timeout:

System.TimeoutException : Shard Trip:All did not reach sequence number 45 in the time allowed

The window widens exactly when the machine is busy, which is why it read as a flaky test rather than a bug. This matters beyond our own tests — WaitForShardState / WaitForShardCondition / WaitForHighWaterMark are the documented way user code (and Marten's and Polecat's suites) waits on daemon progress, and "publish then wait" is the natural shape to write.

The fix

ShardStatusWatcher re-checks tracker.CurrentStates() against its condition immediately after subscribing. The state is therefore either delivered to OnNext or found in the snapshot — it can't fall between the two. TrySetResult makes the double hit harmless: whichever path wins, the other is a no-op.

That also closes WaitForShardCondition's wider gap for free — it never consulted current state at all, so a condition already satisfied by every known state still waited on the next publication.

A user condition that throws on some unrelated shard's state has always just meant "no match" (ShardStateTracker.publish swallows and logs whatever a listener throws), so the snapshot re-check preserves that contract rather than throwing out of WaitForShardCondition.

Also tightened the watcher's own bookkeeping while in there: TrySetException on the timeout and error paths, and dispose the timeout CancellationTokenSource once the wait resolves instead of leaving a live timer behind for every wait.

Tests

Seven tests in ShardStateTrackerTests. Each publishes and then waits for the delivery to land in the tracker's snapshot before waiting — the deterministic form of the race, since the state is provably already published and nothing further will be published for that shard.

Two of them isolate the bug and fail on main (each burning its full 5s timeout), pass with the fix:

• shard_status_watcher_completes_from_the_already_published_state — straight at the watcher, since WaitForShardState's own pre-check would otherwise mask it
• wait_for_shard_condition_does_not_hang_on_an_already_published_state

The rest guard the surrounding behavior: WaitForShardState / WaitForHighWaterMark end to end, that a wait whose state/condition hasn't happened yet still waits and then completes on publication, and that a throwing condition is still just a miss.

Full EventTests suite: 628 passed, 0 failed.

🤖 Generated with Claude Code

https://claude.ai/code/session_01QYfvCxHCUHo9MriHQocuoD