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));
}
///