perf(worker): signal-driven event drain — removes the 25 ms latency floor and idle wakeups

src/ZB.MOM.WW.MxGateway.Worker.Tests/Ipc/WorkerPipeSessionTests.cs

@@ -311,6 +311,61 @@ public sealed class WorkerPipeSessionTests
         await SendShutdownAndWaitAsync(pipePair, runTask, cancellation.Token);
     }
 
+    /// <summary>
+    ///     The event drain loop waits on the runtime's wake signal instead of sleeping a fixed tick,
+    ///     so an event enqueued at an idle worker is framed as soon as it is enqueued rather than up
+    ///     to <c>EventDrainInterval</c> later. The fake's wait honours only the signal here, so the
+    ///     event reaching the pipe is proof the enqueue woke the loop — a poll-driven loop would
+    ///     never run again, and the test would fail on its cancellation deadline instead of passing
+    ///     on a fallback tick that happened to fire. The loop is left parked on that wait before the
+    ///     enqueue, which also makes the recorded fallback ceiling assertable.
+    /// </summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [Fact]
+    public async Task RunAsync_EventAfterIdle_DrainLoopWakesOnSignalNotOnPollTick()
+    {
+        using CancellationTokenSource cancellation = new(TimeSpan.FromSeconds(10));
+        using PipePair pipePair = await PipePair.CreateAsync(cancellation.Token);
+        FakeRuntimeSession runtime = new()
+        {
+            WaitForEventsOnSignalOnly = true,
+        };
+
+        // A far-off heartbeat interval keeps the drain loop the only thing that can produce a frame
+        // after the first beat, so nothing else can mask a drain loop that never woke.
+        WorkerPipeSession session = CreatePipeSession(
+            pipePair.WorkerStream,
+            runtime,
+            new WorkerPipeSessionOptions
+            {
+                HeartbeatInterval = TimeSpan.FromMinutes(5),
+                HeartbeatGrace = TimeSpan.FromSeconds(30),
+            });
+        Task runTask = session.RunAsync(cancellation.Token);
+        await CompleteGatewayHandshakeAsync(pipePair, cancellation.Token);
+
+        // Park the drain loop: it drains empty once and then waits. Enqueuing before it parks would
+        // let the first drain pass find the event, which proves nothing about the wake.
+        while (runtime.LastWaitForEventsTimeout is null)
+        {
+            await Task.Delay(TimeSpan.FromMilliseconds(5), cancellation.Token);
+        }
+
+        runtime.EnqueueEvent(CreateWorkerEvent(sequence: 7));
+
+        WorkerEnvelope workerEvent = await ReadUntilAsync(
+            pipePair.GatewayReader,
+            WorkerEnvelope.BodyOneofCase.WorkerEvent,
+            cancellation.Token);
+
+        Assert.Equal(7UL, workerEvent.WorkerEvent.Event.WorkerSequence);
+
+        // The 25 ms survives as the ceiling the loop passes to every wait, not as a poll period.
+        Assert.Equal(TimeSpan.FromMilliseconds(25), runtime.LastWaitForEventsTimeout);
+
+        await SendShutdownAndWaitAsync(pipePair, runTask, cancellation.Token);
+    }
+
     /// <summary>
     ///     Verifies that a Ping control command is answered on the worker side
     ///     (not dispatched to the STA) with an OK reply that echoes the ping

src/ZB.MOM.WW.MxGateway.Worker.Tests/MxAccess/MxAccessEventQueueTests.cs

@@ -1,5 +1,7 @@
 using System;
 using System.Collections.Generic;
+using System.Threading;
+using System.Threading.Tasks;
 using ZB.MOM.WW.MxGateway.Contracts.Proto;
 using ZB.MOM.WW.MxGateway.Worker.MxAccess;
 
@@ -300,6 +302,123 @@ public sealed class MxAccessEventQueueTests
         Assert.Equal(WorkerFaultCategory.MxaccessEventConversionFailed, queue.Fault?.Category);
     }
 
+    // Wake-signal timings. The fallback is far longer than the patience deliberately: every wait
+    // below asserts "the signal completed this", which is only a faithful claim while the fallback
+    // timeout cannot have completed it within the patience window. The patience itself is generous
+    // so a loaded CI box cannot fail a test that is not about latency.
+    private static readonly TimeSpan WakeFallback = TimeSpan.FromSeconds(30);
+    private static readonly TimeSpan WakePatience = TimeSpan.FromSeconds(5);
+
+    /// <summary>
+    ///     Verifies the queue wakes a parked waiter as soon as an event is enqueued, rather than
+    ///     leaving it to time out. This is what removes the drain loop's latency floor: before the
+    ///     signal existed, an event arriving at an idle queue waited out the loop's whole poll tick.
+    /// </summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [Fact]
+    public async Task WaitForEventsAsync_EnqueueAfterIdle_CompletesWithoutWaitingTheFallback()
+    {
+        MxAccessEventQueue queue = new(capacity: 4);
+
+        Task wait = queue.WaitForEventsAsync(WakeFallback, CancellationToken.None);
+        Assert.False(wait.IsCompleted);
+
+        queue.Enqueue(CreateEvent(MxEventFamily.OnDataChange, itemHandle: 10));
+
+        Assert.Same(wait, await Task.WhenAny(wait, Task.Delay(WakePatience)));
+        await wait;
+        Assert.Equal(1, queue.Count);
+    }
+
+    /// <summary>
+    ///     Verifies a fault recorded while the waiter is parked wakes it too. The drain loop
+    ///     discovers faults by calling <c>DrainFault()</c> at the top of each pass, so without this
+    ///     signal an overflow or conversion fault would not be reported until the loop's fallback
+    ///     tick expired.
+    /// </summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [Fact]
+    public async Task WaitForEventsAsync_RecordFaultWhileParked_WakesTheWaiter()
+    {
+        MxAccessEventQueue queue = new(capacity: 4);
+
+        Task wait = queue.WaitForEventsAsync(WakeFallback, CancellationToken.None);
+        Assert.False(wait.IsCompleted);
+
+        queue.RecordFault(new WorkerFault
+        {
+            Category = WorkerFaultCategory.MxaccessEventConversionFailed,
+        });
+
+        Assert.Same(wait, await Task.WhenAny(wait, Task.Delay(WakePatience)));
+        await wait;
+        Assert.NotNull(queue.DrainFault());
+    }
+
+    /// <summary>
+    ///     Verifies the one-permit cap loses no wakeups. A burst that lands while nobody is waiting
+    ///     leaves exactly one pending wake — the waiter that consumes it drains the whole burst, so
+    ///     coalescing costs nothing — and that consumed wake is not replayed: the next wait parks
+    ///     until a new enqueue signals it.
+    /// </summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [Fact]
+    public async Task WaitForEventsAsync_BurstWhileNoWaiter_CoalescesToOneWakeThatLosesNothing()
+    {
+        MxAccessEventQueue queue = new(capacity: 16);
+        for (int itemHandle = 0; itemHandle < 5; itemHandle++)
+        {
+            queue.Enqueue(CreateEvent(MxEventFamily.OnDataChange, itemHandle));
+        }
+
+        Task firstWait = queue.WaitForEventsAsync(WakeFallback, CancellationToken.None);
+        Assert.Same(firstWait, await Task.WhenAny(firstWait, Task.Delay(WakePatience)));
+        await firstWait;
+
+        // One wake, the whole burst: the waiter re-drains everything queued, which is why capping
+        // the signal at a single permit cannot drop an event.
+        Assert.Equal(5, queue.Drain(maxEvents: 0).Count);
+
+        Task secondWait = queue.WaitForEventsAsync(WakeFallback, CancellationToken.None);
+        Assert.False(secondWait.IsCompleted);
+
+        queue.Enqueue(CreateEvent(MxEventFamily.OnDataChange, itemHandle: 99));
+
+        Assert.Same(secondWait, await Task.WhenAny(secondWait, Task.Delay(WakePatience)));
+        await secondWait;
+    }
+
+    /// <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.
+    /// </summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [Fact]
+    public async Task WaitForEventsAsync_WithNoSignal_CompletesAtTheFallbackTimeout()
+    {
+        MxAccessEventQueue queue = new(capacity: 4);
+
+        Task wait = queue.WaitForEventsAsync(TimeSpan.FromMilliseconds(25), CancellationToken.None);
+
+        Assert.Same(wait, await Task.WhenAny(wait, Task.Delay(WakePatience)));
+        await wait;
+    }
+
+    /// <summary>Verifies a cancelled wait unwinds instead of hanging until the fallback expires.</summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [Fact]
+    public async Task WaitForEventsAsync_WhenCancelled_Throws()
+    {
+        MxAccessEventQueue queue = new(capacity: 4);
+        using CancellationTokenSource cancellation = new();
+
+        Task wait = queue.WaitForEventsAsync(WakeFallback, cancellation.Token);
+        cancellation.Cancel();
+
+        await Assert.ThrowsAnyAsync<OperationCanceledException>(async () => await wait);
+    }
+
     // Mirrors MxAccessEventQueue's per-event repeated-field allowance. Kept local rather than made
     // public on the queue: the byte-budget tests state their budgets in units of that charge, so a
     // change to it should surface here as a failing bound instead of silently moving with the code.

src/ZB.MOM.WW.MxGateway.Worker.Tests/TestSupport/FakeRuntimeSession.cs

@@ -24,6 +24,13 @@ internal sealed class FakeRuntimeSession : IWorkerRuntimeSession
     private readonly object gate = new();
     private readonly Queue<WorkerEvent> events = new();
     private readonly List<string> cancelledCorrelationIds = new();
+
+    // Mirrors MxAccessEventQueue's coalesced wake signal so the drain loop under test is driven the
+    // same way it is in production: EnqueueEvent(s) releases one permit, WaitForEventsAsync consumes
+    // it. Never disposed — the drain loop can still be parked on it while Dispose runs, and a
+    // disposed SemaphoreSlim would turn that shutdown into an ObjectDisposedException.
+    private readonly SemaphoreSlim eventSignal = new(0, 1);
+    private TimeSpan? lastWaitForEventsTimeout;
     private WorkerRuntimeHeartbeatSnapshot snapshot = new(
         DateTimeOffset.UtcNow,
         pendingCommandCount: 0,
@@ -263,6 +270,52 @@ internal sealed class FakeRuntimeSession : IWorkerRuntimeSession
         }
     }
 
+    /// <summary>
+    ///     When set, <see cref="WaitForEventsAsync"/> honours only the wake signal and cancellation,
+    ///     never the fallback timeout. A drain loop that ships an event while this is set can only
+    ///     have been woken by the enqueue signal, which is what makes "the drain is signal-driven,
+    ///     not poll-driven" assertable without racing the 25 ms fallback tick.
+    /// </summary>
+    public bool WaitForEventsOnSignalOnly { get; set; }
+
+    /// <summary>
+    ///     The <c>timeout</c> argument of the most recent <see cref="WaitForEventsAsync"/> call, so
+    ///     a test can assert the drain loop still passes its fallback ceiling.
+    /// </summary>
+    public TimeSpan? LastWaitForEventsTimeout
+    {
+        get
+        {
+            lock (gate)
+            {
+                return lastWaitForEventsTimeout;
+            }
+        }
+    }
+
+    /// <inheritdoc />
+    public Task WaitForEventsAsync(TimeSpan timeout, CancellationToken cancellationToken)
+    {
+        lock (gate)
+        {
+            lastWaitForEventsTimeout = timeout;
+        }
+
+        if (BackingQueue is not null)
+        {
+            // Tests that drive a real queue enqueue into it directly, so the real queue owns the
+            // wake signal too.
+            return BackingQueue.WaitForEventsAsync(timeout, cancellationToken);
+        }
+
+        if (WaitForEventsOnSignalOnly)
+        {
+            return eventSignal.WaitAsync(cancellationToken);
+        }
+
+        return eventSignal.WaitAsync(timeout, cancellationToken);
+    }
+
     /// <inheritdoc />
     public WorkerFault? DrainFault()
     {
@@ -370,6 +423,8 @@ internal sealed class FakeRuntimeSession : IWorkerRuntimeSession
         {
             events.Enqueue(workerEvent);
         }
+
+        SignalWake();
     }
 
     /// <summary>
@@ -387,6 +442,26 @@ internal sealed class FakeRuntimeSession : IWorkerRuntimeSession
                 events.Enqueue(workerEvent);
             }
         }
+
+        SignalWake();
+    }
+
+    // Coalesced wake, released outside the gate exactly as MxAccessEventQueue does.
+    private void SignalWake()
+    {
+        if (eventSignal.CurrentCount > 0)
+        {
+            return;
+        }
+
+        try
+        {
+            eventSignal.Release();
+        }
+        catch (SemaphoreFullException)
+        {
+            // A concurrent enqueue already published the pending wake this call wanted.
+        }
     }
 
     /// <inheritdoc />