perf(worker): message-driven completion waits — the STA pumps continuously while waiting

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

@@ -1,5 +1,6 @@
 using System;
 using System.Collections.Generic;
+using System.Diagnostics;
 using System.Threading;
 using System.Threading.Tasks;
 using Google.Protobuf.WellKnownTypes;
@@ -177,6 +178,77 @@ public sealed class MxAccessValueCacheTests
         Assert.Equal(1UL, value.Version);
     }
 
+    /// <summary>
+    ///     Verifies the wait is message/signal-driven rather than sleep-polled:
+    ///     with a poll interval far longer than the deadline, a value set from
+    ///     another thread still wakes the wait and is returned well inside the
+    ///     deadline. The sleep-polled loop this replaced would have sat blind for
+    ///     the whole 10 s interval — on the STA that means 10 s with no Windows
+    ///     messages dispatched, so the OnDataChange being waited for could not
+    ///     have arrived at all — and then reported a timeout.
+    /// </summary>
+    [Fact]
+    public void TryWaitForUpdate_WakesOnCrossThreadSet_DespiteLongPollInterval()
+    {
+        MxAccessValueCache cache = new();
+        Timestamp sourceTimestamp = Timestamp.FromDateTime(DateTime.UtcNow);
+        using ManualResetEventSlim waitEntered = new(false);
+
+        Task setter = Task.Run(() =>
+        {
+            // Handshake so the value cannot land before the wait starts —
+            // otherwise the first check would satisfy it and prove nothing.
+            waitEntered.Wait(TimeSpan.FromSeconds(5));
+            Thread.Sleep(50);
+            cache.Set(7, 21, BuildEvent(7, 21, intValue: 8080, quality: 192, sourceTimestamp));
+        });
+
+        Stopwatch elapsed = Stopwatch.StartNew();
+        bool found = cache.TryWaitForUpdate(
+            serverHandle: 7,
+            itemHandle: 21,
+            sinceVersion: 0,
+            deadlineUtc: DateTime.UtcNow.AddSeconds(3),
+            pumpStep: () => waitEntered.Set(),
+            out MxAccessValueCache.CachedValue value,
+            pollIntervalMs: 10_000);
+        elapsed.Stop();
+        setter.Wait(TimeSpan.FromSeconds(5));
+
+        Assert.True(found);
+        Assert.Equal(8080, value.Value.Int32Value);
+        Assert.True(
+            elapsed.Elapsed < TimeSpan.FromSeconds(2),
+            $"The wait should have woken on the cached value, not slept out the poll interval; took {elapsed.ElapsedMilliseconds} ms.");
+    }
+
+    /// <summary>
+    ///     Verifies the pump step keeps running throughout a wait that nothing
+    ///     ever signals: the caller's poll interval is capped at the 50 ms
+    ///     fallback tick, so a timing-out wait still pumps repeatedly instead of
+    ///     once. This is what keeps the STA dispatching COM events in a process
+    ///     whose message queue never wakes the wait, and it is the safety net
+    ///     behind the ReadBulk per-tag timeout.
+    /// </summary>
+    [Fact]
+    public void TryWaitForUpdate_KeepsPumping_WhenPollIntervalExceedsTheFallbackTick()
+    {
+        MxAccessValueCache cache = new();
+        int pumpCalls = 0;
+
+        bool found = cache.TryWaitForUpdate(
+            serverHandle: 7,
+            itemHandle: 21,
+            sinceVersion: 0,
+            deadlineUtc: DateTime.UtcNow.AddMilliseconds(400),
+            pumpStep: () => Interlocked.Increment(ref pumpCalls),
+            out _,
+            pollIntervalMs: 10_000);
+
+        Assert.False(found);
+        Assert.True(pumpCalls >= 3, $"Expected repeated pumping across the 400 ms wait, saw {pumpCalls} calls.");
+    }
+
     private static MxEvent BuildEvent(
         int serverHandle,
         int itemHandle,

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

@@ -1,4 +1,7 @@
 using System;
+using System.Diagnostics;
+using System.Threading;
+using System.Threading.Tasks;
 using Google.Protobuf.Collections;
 using ZB.MOM.WW.MxGateway.Contracts.Proto;
 using ZB.MOM.WW.MxGateway.Worker.MxAccess;
@@ -109,6 +112,48 @@ public sealed class MxAccessWriteCompletionCacheTests
         Assert.Equal(55, Assert.Single(statuses).Detail);
     }
 
+    /// <summary>
+    ///     Verifies the wait is message/signal-driven rather than sleep-polled:
+    ///     with a poll interval far longer than the deadline, a completion
+    ///     recorded from another thread still wakes the wait and is returned well
+    ///     inside the deadline. The sleep-polled loop this replaced would have sat
+    ///     blind for the whole 10 s interval — on the STA that means 10 s with no
+    ///     Windows messages dispatched — and then reported a timeout.
+    /// </summary>
+    [Fact]
+    public void TryWaitForCompletion_WakesOnCrossThreadRecord_DespiteLongPollInterval()
+    {
+        MxAccessWriteCompletionCache cache = new();
+        using ManualResetEventSlim waitEntered = new(false);
+
+        Task recorder = Task.Run(() =>
+        {
+            // Handshake so the completion cannot land before the wait starts —
+            // otherwise the first check would satisfy it and prove nothing.
+            waitEntered.Wait(TimeSpan.FromSeconds(5));
+            Thread.Sleep(50);
+            cache.Record(7, 21, BuildStatuses(detail: 8080));
+        });
+
+        Stopwatch elapsed = Stopwatch.StartNew();
+        bool found = cache.TryWaitForCompletion(
+            7,
+            21,
+            sinceVersion: 0UL,
+            deadlineUtc: DateTime.UtcNow.AddSeconds(3),
+            pumpStep: () => waitEntered.Set(),
+            out RepeatedField<MxStatusProxy> statuses,
+            pollIntervalMs: 10_000);
+        elapsed.Stop();
+        recorder.Wait(TimeSpan.FromSeconds(5));
+
+        Assert.True(found);
+        Assert.Equal(8080, Assert.Single(statuses).Detail);
+        Assert.True(
+            elapsed.Elapsed < TimeSpan.FromSeconds(2),
+            $"The wait should have woken on the recorded completion, not slept out the poll interval; took {elapsed.ElapsedMilliseconds} ms.");
+    }
+
     /// <summary>Verifies that Record stores an independent clone of the caller's rows.</summary>
     [Fact]
     public void Record_ClonesStatuses()

src/ZB.MOM.WW.MxGateway.Worker/MxAccess/MxAccessValueCache.cs

@@ -4,6 +4,7 @@ using System.Threading;
 using Google.Protobuf.Collections;
 using Google.Protobuf.WellKnownTypes;
 using ZB.MOM.WW.MxGateway.Contracts.Proto;
+using ZB.MOM.WW.MxGateway.Worker.Sta;
 
 namespace ZB.MOM.WW.MxGateway.Worker.MxAccess;
 
@@ -26,6 +27,18 @@ public sealed class MxAccessValueCache
     private readonly Dictionary<long, CachedValue> entries = new();
     private readonly object syncRoot = new();
 
+    // Set by Set() so a waiter blocked in TryWaitForUpdate wakes the moment a
+    // value lands instead of after the next blind poll tick. In the live worker
+    // the value is recorded from inside the waiter's own pumpStep (COM
+    // dispatches OnDataChange on the STA), so the message wake is what does the
+    // work there; this handle covers the cross-thread producers — the tests, and
+    // any future off-STA writer. Not IDisposable on purpose: the cache is
+    // co-owned by MxAccessBaseEventSink and MxAccessSession with no single
+    // teardown owner, and a late COM callback must never Set a disposed handle.
+    // One kernel event per session (one session per worker process), released by
+    // the handle's finalizer.
+    private readonly AutoResetEvent updateSignal = new(false);
+
     /// <summary>Records a fresh OnDataChange payload for the given handle pair.</summary>
     /// <param name="serverHandle">MXAccess server handle.</param>
     /// <param name="itemHandle">MXAccess item handle.</param>
@@ -68,6 +81,11 @@ public sealed class MxAccessValueCache
                 cachedTimestamp,
                 mxEvent.Statuses.Clone());
         }
+
+        // Signaled outside the lock: the waiter re-takes syncRoot (through
+        // TryGet) the instant it wakes, so signaling while holding it would hand
+        // it guaranteed contention.
+        updateSignal.Set();
     }
 
     /// <summary>Tries to read the most recent cached value for the handle pair.</summary>
@@ -107,16 +125,28 @@ public sealed class MxAccessValueCache
 
     /// <summary>
     ///     Waits until the cache entry's version exceeds <paramref name="sinceVersion"/>
-    ///     or the deadline elapses, calling <paramref name="pumpStep"/> on every poll
+    ///     or the deadline elapses, calling <paramref name="pumpStep"/> on every
     ///     iteration so the worker's STA can dispatch the inbound MXAccess message.
     /// </summary>
+    /// <remarks>
+    ///     The idle part of each iteration blocks in
+    ///     <see cref="StaWaitHelper.WaitForSignalOrMessages"/> rather than
+    ///     <c>Thread.Sleep</c>: a sleeping STA pumps no Windows messages, so a
+    ///     ReadBulk waiting on the first OnDataChange would only dispatch it at
+    ///     poll-tick granularity — and it holds the apartment for up to
+    ///     <c>timeout_ms</c> per tag, stalling every other COM event with it. The
+    ///     Win32 wait returns the instant a message needs pumping (or the value
+    ///     is recorded from another thread), so the apartment keeps dispatching
+    ///     continuously for the whole wait. The per-tag timeout, the deadline
+    ///     math, and both return shapes are unchanged.
+    /// </remarks>
     /// <param name="serverHandle">MXAccess server handle.</param>
     /// <param name="itemHandle">MXAccess item handle.</param>
     /// <param name="sinceVersion">Version snapshot captured before the wait.</param>
     /// <param name="deadlineUtc">Absolute UTC deadline.</param>
     /// <param name="pumpStep">Action that pumps any pending Windows messages.</param>
     /// <param name="value">The cached value if the update was received before the deadline.</param>
-    /// <param name="pollIntervalMs">How long to sleep between pump cycles. Default 5 ms.</param>
+    /// <param name="pollIntervalMs">Upper bound on one idle wait slice — the cadence at which <paramref name="pumpStep"/> runs when nothing wakes the wait. Default 5 ms, itself capped at <see cref="StaWaitHelper.MaxFallbackTickMilliseconds"/>.</param>
     /// <returns><see langword="true"/> if an update newer than <paramref name="sinceVersion"/> arrived before the deadline; otherwise <see langword="false"/>.</returns>
     public bool TryWaitForUpdate(
         int serverHandle,
@@ -134,6 +164,12 @@ public sealed class MxAccessValueCache
 
         while (true)
         {
+            // Pumped unconditionally at the top of every iteration — including
+            // the one that immediately follows a signaled wait, and the very
+            // first before any waiting. Dispatching the message is what fills
+            // the cache in the live worker, and unit tests stand in for the STA
+            // by driving the cache from their fake pumpStep, so this call must
+            // never be conditional on a real message queue.
             pumpStep();
 
             if (TryGet(serverHandle, itemHandle, out value) && value.Version > sinceVersion)
@@ -141,12 +177,18 @@ public sealed class MxAccessValueCache
                 return true;
             }
 
-            if (DateTime.UtcNow >= deadlineUtc)
+            // Same expiry test as the sleep-polled version: remaining <= zero is
+            // exactly DateTime.UtcNow >= deadlineUtc. `value` keeps whatever the
+            // TryGet above left in it on this path, as before.
+            TimeSpan remaining = deadlineUtc - DateTime.UtcNow;
+            if (remaining <= TimeSpan.Zero)
             {
                 return false;
             }
 
-            Thread.Sleep(pollIntervalMs);
+            StaWaitHelper.WaitForSignalOrMessages(
+                updateSignal,
+                StaWaitHelper.ClampWaitMilliseconds(remaining, pollIntervalMs));
         }
     }
 

src/ZB.MOM.WW.MxGateway.Worker/MxAccess/MxAccessWriteCompletionCache.cs

@@ -3,6 +3,7 @@ using System.Collections.Generic;
 using System.Threading;
 using Google.Protobuf.Collections;
 using ZB.MOM.WW.MxGateway.Contracts.Proto;
+using ZB.MOM.WW.MxGateway.Worker.Sta;
 
 namespace ZB.MOM.WW.MxGateway.Worker.MxAccess;
 
@@ -25,6 +26,18 @@ public sealed class MxAccessWriteCompletionCache
     private readonly Dictionary<long, CompletionEntry> entries = new();
     private readonly object syncRoot = new();
 
+    // Set by Record so a waiter blocked in TryWaitForCompletion wakes the moment
+    // a completion lands instead of after the next blind poll tick. In the live
+    // worker the completion is recorded from inside the waiter's own pumpStep
+    // (COM dispatches OnWriteComplete on the STA), so the message wake is what
+    // does the work there; this handle covers the cross-thread producers — the
+    // tests, and any future off-STA recorder. Not IDisposable on purpose: the
+    // cache is co-owned by MxAccessBaseEventSink and MxAccessSession with no
+    // single teardown owner, and a late COM callback must never Set a disposed
+    // handle. One kernel event per session (one session per worker process),
+    // released by the handle's finalizer.
+    private readonly AutoResetEvent completionSignal = new(false);
+
     /// <summary>Records the status rows of a fresh OnWriteComplete callback for the given handle pair.</summary>
     /// <param name="serverHandle">MXAccess server handle.</param>
     /// <param name="itemHandle">MXAccess item handle.</param>
@@ -47,6 +60,10 @@ public sealed class MxAccessWriteCompletionCache
                 : 1UL;
             entries[key] = new CompletionEntry(version, statuses.Clone());
         }
+
+        // Signaled outside the lock: the waiter re-takes syncRoot the instant it
+        // wakes, so signaling while holding it would hand it guaranteed contention.
+        completionSignal.Set();
     }
 
     /// <summary>Returns the current completion version for a handle pair, or 0 if none was recorded.</summary>
@@ -66,19 +83,29 @@ public sealed class MxAccessWriteCompletionCache
     }
 
     /// <summary>
-    ///     Polls for a completion newer than <paramref name="sinceVersion"/> until it
+    ///     Waits for a completion newer than <paramref name="sinceVersion"/> until it
     ///     arrives or the deadline elapses, calling <paramref name="pumpStep"/> on every
-    ///     poll iteration so the worker's STA can dispatch the inbound MXAccess
+    ///     iteration so the worker's STA can dispatch the inbound MXAccess
     ///     OnWriteComplete message. Same loop shape as
     ///     <see cref="MxAccessValueCache.TryWaitForUpdate"/>.
     /// </summary>
+    /// <remarks>
+    ///     The idle part of each iteration blocks in
+    ///     <see cref="StaWaitHelper.WaitForSignalOrMessages"/> rather than
+    ///     <c>Thread.Sleep</c>: a sleeping STA pumps no Windows messages, so the
+    ///     inbound OnWriteComplete could only ever be dispatched at poll-tick
+    ///     granularity. The Win32 wait returns the instant a message needs
+    ///     pumping (or the completion is recorded from another thread), so the
+    ///     apartment keeps dispatching COM events continuously for the whole
+    ///     wait. Timeouts, deadline math, and both return shapes are unchanged.
+    /// </remarks>
     /// <param name="serverHandle">MXAccess server handle.</param>
     /// <param name="itemHandle">MXAccess item handle.</param>
     /// <param name="sinceVersion">Version snapshot captured before the write COM call.</param>
     /// <param name="deadlineUtc">Absolute UTC deadline.</param>
     /// <param name="pumpStep">Action that pumps any pending Windows messages.</param>
     /// <param name="statuses">The recorded status rows if a completion arrived before the deadline; empty otherwise.</param>
-    /// <param name="pollIntervalMs">How long to sleep between pump cycles. Default 5 ms.</param>
+    /// <param name="pollIntervalMs">Upper bound on one idle wait slice — the cadence at which <paramref name="pumpStep"/> runs when nothing wakes the wait. Default 5 ms, itself capped at <see cref="StaWaitHelper.MaxFallbackTickMilliseconds"/>.</param>
     /// <returns><see langword="true"/> if a completion newer than <paramref name="sinceVersion"/> arrived before the deadline; otherwise <see langword="false"/>.</returns>
     public bool TryWaitForCompletion(
         int serverHandle,
@@ -96,6 +123,12 @@ public sealed class MxAccessWriteCompletionCache
 
         while (true)
         {
+            // Pumped unconditionally at the top of every iteration — including
+            // the one that immediately follows a signaled wait, and the very
+            // first before any waiting. Dispatching the message is what fills
+            // the cache in the live worker, and unit tests stand in for the STA
+            // by recording the completion from inside their fake pumpStep, so
+            // this call must never be conditional on a real message queue.
             pumpStep();
 
             lock (syncRoot)
@@ -108,13 +141,18 @@ public sealed class MxAccessWriteCompletionCache
                 }
             }
 
-            if (DateTime.UtcNow >= deadlineUtc)
+            // Same expiry test as the sleep-polled version: remaining <= zero is
+            // exactly DateTime.UtcNow >= deadlineUtc.
+            TimeSpan remaining = deadlineUtc - DateTime.UtcNow;
+            if (remaining <= TimeSpan.Zero)
             {
                 statuses = new RepeatedField<MxStatusProxy>();
                 return false;
             }
 
-            Thread.Sleep(pollIntervalMs);
+            StaWaitHelper.WaitForSignalOrMessages(
+                completionSignal,
+                StaWaitHelper.ClampWaitMilliseconds(remaining, pollIntervalMs));
         }
     }
 

src/ZB.MOM.WW.MxGateway.Worker/Sta/StaWaitHelper.cs

@@ -0,0 +1,118 @@
+using System;
+using System.Runtime.InteropServices;
+using System.Threading;
+using Microsoft.Win32.SafeHandles;
+
+namespace ZB.MOM.WW.MxGateway.Worker.Sta;
+
+/// <summary>
+///     Message-aware blocking wait shared by STA code that must sit idle without
+///     starving the Windows message queue. <see cref="StaMessagePump"/> owns the
+///     runtime's outer idle wait; this helper serves the inner waits performed
+///     by commands that already hold the STA — the write-completion wait and the
+///     ReadBulk first-value wait — which need the same "wake on a signal or on
+///     an inbound message" behavior without owning a pump instance.
+/// </summary>
+/// <remarks>
+///     Deliberately separate from <see cref="StaMessagePump"/> rather than a new
+///     method on it: the two caches that use this are plain per-session objects
+///     with no pump dependency (they take the pump step as an injected delegate
+///     so tests can drive them off the STA). Keeping the helper in this folder
+///     keeps every <c>user32</c> wait signature in the worker reviewable in one
+///     place; the declaration below mirrors the pump's exactly.
+/// </remarks>
+internal static class StaWaitHelper
+{
+    /// <summary>
+    ///     Ceiling on the blind fallback tick. A caller's poll interval is
+    ///     clamped to this so no wait can go longer than 50 ms without calling
+    ///     its pump step, even when the process has no message queue to wake it
+    ///     (unit-test threads) and the wait handle is never signaled.
+    /// </summary>
+    internal const int MaxFallbackTickMilliseconds = 50;
+
+    private const uint MsgWaitFailed = 0xFFFFFFFF;
+    private const uint MwmoInputAvailable = 0x0004;
+    private const uint QsAllInput = 0x04FF;
+
+    /// <summary>
+    ///     Blocks the calling thread until <paramref name="signal"/> is set, a
+    ///     Windows message is available to pump, or the timeout elapses —
+    ///     whichever comes first. Unlike <see cref="Thread.Sleep(int)"/> this
+    ///     never blinds the thread to inbound COM event messages: the wait
+    ///     returns the instant one arrives, so the caller can dispatch it.
+    /// </summary>
+    /// <param name="signal">Wait handle set by whoever produces the awaited result.</param>
+    /// <param name="timeoutMilliseconds">Maximum wait; values at or below zero return immediately.</param>
+    internal static void WaitForSignalOrMessages(
+        WaitHandle signal,
+        int timeoutMilliseconds)
+    {
+        if (signal is null)
+        {
+            throw new ArgumentNullException(nameof(signal));
+        }
+
+        if (timeoutMilliseconds <= 0)
+        {
+            return;
+        }
+
+        SafeWaitHandle safeHandle = signal.SafeWaitHandle;
+        IntPtr[] handles = [safeHandle.DangerousGetHandle()];
+        uint result = MsgWaitForMultipleObjectsEx(
+            (uint)handles.Length,
+            handles,
+            (uint)timeoutMilliseconds,
+            QsAllInput,
+            MwmoInputAvailable);
+
+        // The handle is a field of a live caller object, so it cannot be
+        // collected mid-call; KeepAlive states that dependency explicitly
+        // rather than relying on it.
+        GC.KeepAlive(signal);
+
+        if (result == MsgWaitFailed)
+        {
+            // Degrade to the pre-existing blind sleep instead of throwing or
+            // hot-spinning: the caller is a bounded, deadline-driven poll loop,
+            // and a failed wait must not turn a timed-out write into a fault.
+            Thread.Sleep(timeoutMilliseconds);
+        }
+    }
+
+    /// <summary>
+    ///     Computes the wait slice for one iteration of a deadline-driven poll
+    ///     loop: never past the deadline, never longer than the caller's poll
+    ///     interval, and never longer than <see cref="MaxFallbackTickMilliseconds"/>.
+    /// </summary>
+    /// <param name="remaining">Time left before the caller's absolute deadline.</param>
+    /// <param name="pollIntervalMilliseconds">The caller's requested pump cadence.</param>
+    /// <returns>Milliseconds to wait; zero or less when the deadline has passed.</returns>
+    internal static int ClampWaitMilliseconds(
+        TimeSpan remaining,
+        int pollIntervalMilliseconds)
+    {
+        int tick = pollIntervalMilliseconds < 1
+            ? 1
+            : Math.Min(pollIntervalMilliseconds, MaxFallbackTickMilliseconds);
+
+        double remainingMilliseconds = remaining.TotalMilliseconds;
+        if (remainingMilliseconds <= 0d)
+        {
+            return 0;
+        }
+
+        return remainingMilliseconds < tick
+            ? (int)Math.Ceiling(remainingMilliseconds)
+            : tick;
+    }
+
+    [DllImport("user32.dll", SetLastError = true)]
+    private static extern uint MsgWaitForMultipleObjectsEx(
+        uint count,
+        IntPtr[] handles,
+        uint milliseconds,
+        uint wakeMask,
+        uint flags);
+}