From a76ecdd59cd205d4b016f2656d64d08a94655ff7 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Sun, 9 Aug 2026 12:22:54 -0400 Subject: [PATCH] feat(worker): event sink records OnWriteComplete rows into the completion cache --- .../MxAccess/MxAccessBaseEventSinkTests.cs | 41 +++++++++++++++ .../MxAccess/IWriteCompletionCacheProvider.cs | 15 ++++++ .../MxAccess/MxAccessBaseEventSink.cs | 50 ++++++++++++++++--- 3 files changed, 100 insertions(+), 6 deletions(-) create mode 100644 src/ZB.MOM.WW.MxGateway.Worker/MxAccess/IWriteCompletionCacheProvider.cs diff --git a/src/ZB.MOM.WW.MxGateway.Worker.Tests/MxAccess/MxAccessBaseEventSinkTests.cs b/src/ZB.MOM.WW.MxGateway.Worker.Tests/MxAccess/MxAccessBaseEventSinkTests.cs index 935e9d2..a81362e 100644 --- a/src/ZB.MOM.WW.MxGateway.Worker.Tests/MxAccess/MxAccessBaseEventSinkTests.cs +++ b/src/ZB.MOM.WW.MxGateway.Worker.Tests/MxAccess/MxAccessBaseEventSinkTests.cs @@ -110,6 +110,47 @@ public sealed class MxAccessBaseEventSinkTests Assert.Same(cache, sink.ValueCache); } + /// + /// 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). + /// + [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(); + + 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)); + } + + /// + /// Verifies that the sink-bound write-completion cache is exposed for sharing + /// with the owning so the sink's recordings and + /// the write executor's waits see the same instance. + /// + [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); + } + /// /// Verifies that consecutive OnDataChange callbacks land in the queue with monotonic sequences. /// diff --git a/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/IWriteCompletionCacheProvider.cs b/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/IWriteCompletionCacheProvider.cs new file mode 100644 index 0000000..71509e9 --- /dev/null +++ b/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/IWriteCompletionCacheProvider.cs @@ -0,0 +1,15 @@ +namespace ZB.MOM.WW.MxGateway.Worker.MxAccess; + +/// +/// Exposes the per-session an +/// event sink populates from OnWriteComplete callbacks, so +/// can share one instance between the +/// sink (writer) and the write command executor (reader). Implemented by +/// and by test sinks that cannot +/// attach to a live MXAccess COM object. +/// +public interface IWriteCompletionCacheProvider +{ + /// The completion cache bound to this sink. + MxAccessWriteCompletionCache WriteCompletionCache { get; } +} diff --git a/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/MxAccessBaseEventSink.cs b/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/MxAccessBaseEventSink.cs index 33c0bfe..d3903c9 100644 --- a/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/MxAccessBaseEventSink.cs +++ b/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; /// Sink for MXAccess COM events that converts them to protobuf format. -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()) + { + } + + /// + /// 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 + /// OnWriteComplete dispatch so the worker's write executor can + /// correlate a WriteSecured/WriteSecured2 reply with the MXAccess + /// completion outcome. + /// + /// Queue for buffering converted MXAccess events. + /// Converter for MXAccess events to protobuf format. + /// Per-session last-value cache shared with the MxAccessSession. + /// Per-session OnWriteComplete cache shared with the MxAccessSession. + 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)); } /// @@ -62,6 +85,14 @@ public sealed class MxAccessBaseEventSink : IMxAccessEventSink /// public MxAccessValueCache ValueCache => valueCache; + /// + /// The OnWriteComplete completion cache populated by this sink. Exposed + /// via so the + /// MxAccessSession can share the same instance with the write command + /// executor's bounded completion wait. + /// + public MxAccessWriteCompletionCache WriteCompletionCache => writeCompletionCache; + /// 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)); } ///