Fix epoch race condition causing unit tests to hang on net10.0 and net472

The HasUnprocessedMessages guard (previous commit) correctly prevents
spurious workflow_invoke Activity creation on timeout wake-ups, but
exposed a latent race in the epoch-based signal filtering.

The race: when the run loop processes messages quickly and calls
Interlocked.Increment(ref _completionEpoch) before the consumer calls
TakeEventStreamAsync, the consumer reads the already-incremented epoch
and sets myEpoch = epoch + 1. This causes the consumer to skip the
valid InternalHaltSignal (its epoch < myEpoch) and block forever
waiting for a signal that will never arrive (since the guard prevents
spurious signal generation).

Fix: read _completionEpoch without +1. The +1 was originally needed to
filter stale signals from timeout-driven spurious loop iterations, but
those no longer exist thanks to the HasUnprocessedMessages guard.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
This commit is contained in:
Roger Barreto
2026-02-27 20:14:52 +00:00
Unverified
parent 98963e17f2
commit 6ce7f01be8
@@ -206,8 +206,12 @@ internal sealed class StreamingRunEventStream : IRunEventStream
bool blockOnPendingRequest,
[EnumeratorCancellation] CancellationToken cancellationToken = default)
{
// Get the current epoch - we'll only respond to completion signals from this epoch or later
int myEpoch = Volatile.Read(ref this._completionEpoch) + 1;
// Get the current epoch - we'll only respond to completion signals from this epoch or later.
// Note: We read the current value (not +1) because the HasUnprocessedMessages guard in the
// run loop prevents spurious completion signals, so there are no stale signals to filter.
// Using +1 would race with the run loop's Interlocked.Increment, causing the consumer to
// skip the valid signal when the run loop finishes before TakeEventStreamAsync starts.
int myEpoch = Volatile.Read(ref this._completionEpoch);
// Use custom async enumerable to avoid exceptions on cancellation.
NonThrowingChannelReaderAsyncEnumerable<WorkflowEvent> eventStream = new(this._eventChannel.Reader);