feat(worker): event sink records OnWriteComplete rows into the completion cache

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

@@ -110,6 +110,47 @@ public sealed class MxAccessBaseEventSinkTests
         Assert.Same(cache, sink.ValueCache);
     }
 
+    /// <summary>
+    /// Verifies that an OnWriteComplete COM callback records the completion into
+    /// the per-session write-completion cache for the executor's bounded reply
+    /// wait AND still enqueues the event unchanged for the outbound stream —
+    /// correlation observes the event, it never consumes it. The cache update
+    /// fires only after the event has cleared the queue (post-publish rule).
+    /// </summary>
+    [Fact]
+    public void OnWriteComplete_ComCallback_RecordsCompletionAndStillEnqueuesEvent()
+    {
+        MxAccessEventQueue queue = new();
+        MxAccessWriteCompletionCache completionCache = new();
+        MxAccessBaseEventSink sink = new(queue, new MxAccessEventMapper(), new MxAccessValueCache(), completionCache);
+        MXSTATUS_PROXY[] statuses = Array.Empty<MXSTATUS_PROXY>();
+
+        sink.OnWriteComplete(hLMXServerHandle: 7, phItemHandle: 21, ref statuses);
+
+        Assert.Equal(1, queue.Count);
+        Assert.True(queue.TryDequeue(out WorkerEvent? workerEvent));
+        MxEvent mxEvent = workerEvent!.Event;
+        Assert.Equal(MxEventFamily.OnWriteComplete, mxEvent.Family);
+        Assert.Equal(7, mxEvent.ServerHandle);
+        Assert.Equal(21, mxEvent.ItemHandle);
+        Assert.Equal(1UL, completionCache.CurrentVersion(7, 21));
+    }
+
+    /// <summary>
+    /// Verifies that the sink-bound write-completion cache is exposed for sharing
+    /// with the owning <see cref="MxAccessSession"/> so the sink's recordings and
+    /// the write executor's waits see the same instance.
+    /// </summary>
+    [Fact]
+    public void WriteCompletionCache_ReturnsTheInstanceBoundAtConstruction()
+    {
+        MxAccessEventQueue queue = new();
+        MxAccessWriteCompletionCache completionCache = new();
+        MxAccessBaseEventSink sink = new(queue, new MxAccessEventMapper(), new MxAccessValueCache(), completionCache);
+
+        Assert.Same(completionCache, sink.WriteCompletionCache);
+    }
+
     /// <summary>
     /// Verifies that consecutive OnDataChange callbacks land in the queue with monotonic sequences.
     /// </summary>

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

@@ -0,0 +1,15 @@
+namespace ZB.MOM.WW.MxGateway.Worker.MxAccess;
+
+/// <summary>
+///     Exposes the per-session <see cref="MxAccessWriteCompletionCache"/> an
+///     event sink populates from OnWriteComplete callbacks, so
+///     <see cref="MxAccessSession.Create"/> can share one instance between the
+///     sink (writer) and the write command executor (reader). Implemented by
+///     <see cref="MxAccessBaseEventSink"/> and by test sinks that cannot
+///     attach to a live MXAccess COM object.
+/// </summary>
+public interface IWriteCompletionCacheProvider
+{
+    /// <summary>The completion cache bound to this sink.</summary>
+    MxAccessWriteCompletionCache WriteCompletionCache { get; }
+}

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

@@ -5,11 +5,12 @@ using Proto = ZB.MOM.WW.MxGateway.Contracts.Proto;
 namespace ZB.MOM.WW.MxGateway.Worker.MxAccess;
 
 /// <summary>Sink for MXAccess COM events that converts them to protobuf format.</summary>
-public sealed class MxAccessBaseEventSink : IMxAccessEventSink
+public sealed class MxAccessBaseEventSink : IMxAccessEventSink, IWriteCompletionCacheProvider
 {
     private readonly MxAccessEventMapper eventMapper;
     private readonly MxAccessEventQueue eventQueue;
     private readonly MxAccessValueCache valueCache;
+    private readonly MxAccessWriteCompletionCache writeCompletionCache;
     private LMXProxyServerClass? server;
     private string sessionId = string.Empty;
 
@@ -50,10 +51,32 @@ public sealed class MxAccessBaseEventSink : IMxAccessEventSink
         MxAccessEventQueue eventQueue,
         MxAccessEventMapper eventMapper,
         MxAccessValueCache valueCache)
+        : this(eventQueue, eventMapper, valueCache, new MxAccessWriteCompletionCache())
+    {
+    }
+
+    /// <summary>
+    ///     Initializes a new instance of the MxAccessBaseEventSink class with
+    ///     provided queue, mapper, value cache, and a shared write-completion
+    ///     cache. The completion cache is populated from every successful
+    ///     <c>OnWriteComplete</c> dispatch so the worker's write executor can
+    ///     correlate a WriteSecured/WriteSecured2 reply with the MXAccess
+    ///     completion outcome.
+    /// </summary>
+    /// <param name="eventQueue">Queue for buffering converted MXAccess events.</param>
+    /// <param name="eventMapper">Converter for MXAccess events to protobuf format.</param>
+    /// <param name="valueCache">Per-session last-value cache shared with the MxAccessSession.</param>
+    /// <param name="writeCompletionCache">Per-session OnWriteComplete cache shared with the MxAccessSession.</param>
+    public MxAccessBaseEventSink(
+        MxAccessEventQueue eventQueue,
+        MxAccessEventMapper eventMapper,
+        MxAccessValueCache valueCache,
+        MxAccessWriteCompletionCache writeCompletionCache)
     {
         this.eventQueue = eventQueue ?? throw new ArgumentNullException(nameof(eventQueue));
         this.eventMapper = eventMapper ?? throw new ArgumentNullException(nameof(eventMapper));
         this.valueCache = valueCache ?? throw new ArgumentNullException(nameof(valueCache));
+        this.writeCompletionCache = writeCompletionCache ?? throw new ArgumentNullException(nameof(writeCompletionCache));
     }
 
     /// <summary>
@@ -62,6 +85,14 @@ public sealed class MxAccessBaseEventSink : IMxAccessEventSink
     /// </summary>
     public MxAccessValueCache ValueCache => valueCache;
 
+    /// <summary>
+    ///     The OnWriteComplete completion cache populated by this sink. Exposed
+    ///     via <see cref="IWriteCompletionCacheProvider"/> so the
+    ///     MxAccessSession can share the same instance with the write command
+    ///     executor's bounded completion wait.
+    /// </summary>
+    public MxAccessWriteCompletionCache WriteCompletionCache => writeCompletionCache;
+
     /// <inheritdoc />
     public void Attach(
         object mxAccessComObject,
@@ -143,11 +174,18 @@ public sealed class MxAccessBaseEventSink : IMxAccessEventSink
         ref MXSTATUS_PROXY[] pVars)
     {
         MXSTATUS_PROXY[] statuses = pVars;
-        EnqueueEvent(() => eventMapper.CreateOnWriteComplete(
-            sessionId,
-            hLMXServerHandle,
-            phItemHandle,
-            statuses));
+        // Record the completion for the write executor's bounded reply wait
+        // only after the event has cleared the queue (same post-publish rule
+        // as the OnDataChange value cache) — an overflow faults the session,
+        // so a dropped event never leaves a "fresher" completion behind than
+        // what shipped to the gateway.
+        EnqueueEvent(
+            () => eventMapper.CreateOnWriteComplete(
+                sessionId,
+                hLMXServerHandle,
+                phItemHandle,
+                statuses),
+            mxEvent => writeCompletionCache.Record(hLMXServerHandle, phItemHandle, mxEvent.Statuses));
     }
 
     /// <summary>