docs(worker): record the SemaphoreSlim/STA dispatch invariant + single-waiter contract; harden fake

This commit is contained in:
Joseph Doherty
2026-08-15 17:13:17 -04:00
parent 25cbe5cd3e
commit 896d81e286
4 changed files with 47 additions and 4 deletions
@@ -1,5 +1,6 @@
using System;
using System.Collections.Generic;
using System.Diagnostics;
using System.Threading;
using System.Threading.Tasks;
using ZB.MOM.WW.MxGateway.Contracts.Proto;
@@ -389,20 +390,32 @@ public sealed class MxAccessEventQueueTests
}
/// <summary>
/// Verifies the timeout still bounds an unsignalled wait: the fallback survives as a ceiling
/// on how long a caller may sleep, so a state change reached by some future path that does
/// not signal is still observed on the next pass rather than never.
/// Verifies the timeout still bounds an unsignalled wait, from both sides: the wait ends
/// without a signal (the fallback survives as a ceiling, so a state change reached by some
/// future path that does not signal is still observed on the next pass rather than never)
/// and it does not end early (the wait really is the timeout, not an already-armed permit
/// completing it instantly). A longer-than-production timeout is used so the lower bound
/// carries a wide margin over timer resolution.
/// </summary>
/// <returns>A task that represents the asynchronous operation.</returns>
[Fact]
public async Task WaitForEventsAsync_WithNoSignal_CompletesAtTheFallbackTimeout()
{
MxAccessEventQueue queue = new(capacity: 4);
TimeSpan fallback = TimeSpan.FromMilliseconds(200);
Stopwatch elapsed = Stopwatch.StartNew();
Task wait = queue.WaitForEventsAsync(TimeSpan.FromMilliseconds(25), CancellationToken.None);
Task wait = queue.WaitForEventsAsync(fallback, CancellationToken.None);
Assert.Same(wait, await Task.WhenAny(wait, Task.Delay(WakePatience)));
await wait;
elapsed.Stop();
// Half the timeout: far enough below it to be immune to timer resolution, far enough above
// zero to fail an implementation that returned a completed task instead of waiting.
Assert.True(
elapsed.Elapsed >= TimeSpan.FromMilliseconds(100),
$"Unsignalled wait returned after {elapsed.ElapsedMilliseconds} ms, well inside its {fallback.TotalMilliseconds} ms fallback.");
}
/// <summary>Verifies a cancelled wait unwinds instead of hanging until the fallback expires.</summary>
@@ -303,6 +303,17 @@ internal sealed class FakeRuntimeSession : IWorkerRuntimeSession
if (BackingQueue is not null)
{
if (WaitForEventsOnSignalOnly)
{
// The two are mutually exclusive: a real backing queue owns its own signal and
// always honours the timeout, so silently letting it win would leave a
// signal-only test passing on a fallback tick — exactly the false green the
// option exists to rule out.
throw new InvalidOperationException(
"FakeRuntimeSession cannot combine BackingQueue with WaitForEventsOnSignalOnly: "
+ "the backing queue honours the fallback timeout, which defeats the signal-only wait.");
}
// Tests that drive a real queue enqueue into it directly, so the real queue owns the
// wake signal too.
return BackingQueue.WaitForEventsAsync(timeout, cancellationToken);
@@ -68,6 +68,11 @@ public interface IWorkerRuntimeSession : IDisposable
/// no default interface members, so every implementation supplies it. The wait's outcome is
/// not surfaced: the caller re-drains and re-checks <see cref="DrainFault"/> after every
/// wait, because the signal is coalesced and the timeout is a ceiling, not a poll period.
/// <para>
/// Precondition: <b>at most one waiter</b>. The implementation's wake signal carries a
/// single permit, which is sufficient only because a session has exactly one event
/// drain loop; a second concurrent waiter would degrade to the fallback timeout.
/// </para>
/// </remarks>
/// <param name="timeout">Maximum time to wait before the wait completes unsignalled.</param>
/// <param name="cancellationToken">Token cancelling the wait.</param>
@@ -75,6 +75,13 @@ public sealed class MxAccessEventQueue
// waiter, no wait handle is ever materialized (nothing touches AvailableWaitHandle), and making
// the queue IDisposable would only add a lifecycle race between the STA's Enqueue and the
// drain loop's wait. WorkerFrameWriter's _writeLock is held the same way.
//
// It must STAY a SemaphoreSlim: on .NET Framework 4.8, Release() hands an async waiter to the
// thread pool (the waiter is a TaskNode : IThreadPoolWorkItem) rather than completing it
// inline, so the STA thread's Enqueue never runs the drain loop's continuation. Swapping this
// for a TaskCompletionSource — a refactor that looks cosmetic — would complete the waiter
// inline on the caller and put the drain loop's pipe write on the apartment thread, which is a
// real STA/COM reentrancy hazard (and stalls the message pump behind the write).
private readonly SemaphoreSlim eventSignal = new SemaphoreSlim(initialCount: 0, maxCount: 1);
private ulong lastEventSequence;
@@ -257,6 +264,13 @@ public sealed class MxAccessEventQueue
/// enqueues. The timeout is therefore a ceiling on how long the caller may sleep, not a
/// poll period: it bounds how late a state change reaches the caller if some future path
/// mutates the queue without signalling.
/// <para>
/// Precondition: <b>at most one waiter</b>. One permit is sufficient only because the
/// queue has exactly one consumer (the worker's single event drain loop). A second
/// concurrent waiter would find no permit left for it and degrade to the fallback
/// timeout, so a design that needs multiple drainers must widen the signal rather than
/// reuse this one.
/// </para>
/// </remarks>
/// <param name="timeout">Maximum time to wait before the wait completes unsignalled.</param>
/// <param name="cancellationToken">Token cancelling the wait.</param>