diff --git a/src/ZB.MOM.WW.MxGateway.Worker.Tests/MxAccess/MxAccessWriteCompletionCacheTests.cs b/src/ZB.MOM.WW.MxGateway.Worker.Tests/MxAccess/MxAccessWriteCompletionCacheTests.cs new file mode 100644 index 0000000..670bc5f --- /dev/null +++ b/src/ZB.MOM.WW.MxGateway.Worker.Tests/MxAccess/MxAccessWriteCompletionCacheTests.cs @@ -0,0 +1,145 @@ +using System; +using Google.Protobuf.Collections; +using ZB.MOM.WW.MxGateway.Contracts.Proto; +using ZB.MOM.WW.MxGateway.Worker.MxAccess; + +namespace ZB.MOM.WW.MxGateway.Worker.Tests.MxAccess; + +/// +/// Unit tests for . The cache is +/// consumed by the write command executor's bounded pump-wait so a +/// WriteSecured/WriteSecured2 reply can carry the correlated +/// OnWriteComplete outcome; its version-baseline contract is exercised in +/// isolation here before the STA / COM plumbing gets layered on top. +/// +public sealed class MxAccessWriteCompletionCacheTests +{ + /// Verifies that Record bumps the version per key and keys stay isolated. + [Fact] + public void Record_IncrementsVersionPerKey() + { + MxAccessWriteCompletionCache cache = new(); + + Assert.Equal(0UL, cache.CurrentVersion(7, 21)); + + cache.Record(7, 21, BuildStatuses(detail: 100)); + Assert.Equal(1UL, cache.CurrentVersion(7, 21)); + + cache.Record(7, 21, BuildStatuses(detail: 200)); + Assert.Equal(2UL, cache.CurrentVersion(7, 21)); + + cache.Record(7, 22, BuildStatuses(detail: 300)); + Assert.Equal(1UL, cache.CurrentVersion(7, 22)); + Assert.Equal(2UL, cache.CurrentVersion(7, 21)); + } + + /// Verifies that a completion newer than the baseline is returned with its status rows. + [Fact] + public void TryWaitForCompletion_WhenCompletionNewerThanBaseline_ReturnsStatuses() + { + MxAccessWriteCompletionCache cache = new(); + cache.Record(7, 21, BuildStatuses(detail: 4321)); + + bool found = cache.TryWaitForCompletion( + 7, + 21, + sinceVersion: 0UL, + deadlineUtc: DateTime.UtcNow.AddSeconds(5), + pumpStep: static () => { }, + out RepeatedField statuses); + + Assert.True(found); + MxStatusProxy row = Assert.Single(statuses); + Assert.Equal(4321, row.Detail); + Assert.Equal(MxStatusCategory.Ok, row.Category); + } + + /// + /// Verifies that a completion recorded before the baseline was captured is + /// never misattributed to the waiting write: only a strictly newer version + /// satisfies the wait, so a stale row times the wait out. + /// + [Fact] + public void TryWaitForCompletion_WhenOnlyStaleCompletion_TimesOut() + { + MxAccessWriteCompletionCache cache = new(); + cache.Record(7, 21, BuildStatuses(detail: 4321)); + ulong baseline = cache.CurrentVersion(7, 21); + + bool found = cache.TryWaitForCompletion( + 7, + 21, + sinceVersion: baseline, + deadlineUtc: DateTime.UtcNow.AddMilliseconds(50), + pumpStep: static () => { }, + out RepeatedField statuses); + + Assert.False(found); + Assert.Empty(statuses); + } + + /// + /// Verifies the pump loop is what lets a completion land: the completion is + /// recorded from inside a later pump step (standing in for the STA + /// dispatching the OnWriteComplete message) and the wait then succeeds. + /// + [Fact] + public void TryWaitForCompletion_InvokesPumpStepEachIteration() + { + MxAccessWriteCompletionCache cache = new(); + int pumpCalls = 0; + + bool found = cache.TryWaitForCompletion( + 7, + 21, + sinceVersion: 0UL, + deadlineUtc: DateTime.UtcNow.AddSeconds(5), + pumpStep: () => + { + pumpCalls++; + if (pumpCalls == 2) + { + cache.Record(7, 21, BuildStatuses(detail: 55)); + } + }, + out RepeatedField statuses); + + Assert.True(found); + Assert.True(pumpCalls >= 2); + Assert.Equal(55, Assert.Single(statuses).Detail); + } + + /// Verifies that Record stores an independent clone of the caller's rows. + [Fact] + public void Record_ClonesStatuses() + { + MxAccessWriteCompletionCache cache = new(); + RepeatedField callerRows = BuildStatuses(detail: 77); + + cache.Record(7, 21, callerRows); + callerRows[0].Detail = 999; + callerRows.Add(new MxStatusProxy()); + + Assert.True(cache.TryWaitForCompletion( + 7, + 21, + sinceVersion: 0UL, + deadlineUtc: DateTime.UtcNow.AddSeconds(5), + pumpStep: static () => { }, + out RepeatedField statuses)); + Assert.Equal(77, Assert.Single(statuses).Detail); + } + + private static RepeatedField BuildStatuses(int detail) + { + return new RepeatedField + { + new MxStatusProxy + { + Success = 1, + Category = MxStatusCategory.Ok, + Detail = detail, + }, + }; + } +} diff --git a/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/MxAccessWriteCompletionCache.cs b/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/MxAccessWriteCompletionCache.cs new file mode 100644 index 0000000..1bb5e35 --- /dev/null +++ b/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/MxAccessWriteCompletionCache.cs @@ -0,0 +1,152 @@ +using System; +using System.Collections.Generic; +using System.Threading; +using Google.Protobuf.Collections; +using ZB.MOM.WW.MxGateway.Contracts.Proto; + +namespace ZB.MOM.WW.MxGateway.Worker.MxAccess; + +/// +/// Per-session cache of the most recent OnWriteComplete status rows +/// for each (server handle, item handle) pair. Written by the MXAccess +/// event sink as completion callbacks arrive; read by the write command +/// executor so a WriteSecured/WriteSecured2 reply can carry the correlated +/// MXAccess outcome instead of proving command acceptance only. +/// +/// +/// Same threading posture as : writers and +/// readers run on the worker's STA thread (COM dispatches events on the +/// apartment thread; commands also execute on the STA), so no internal +/// locking is required. A single sync root keeps it nominally thread-safe +/// for tests that drive it from a non-STA thread. +/// +public sealed class MxAccessWriteCompletionCache +{ + private readonly Dictionary entries = new(); + private readonly object syncRoot = new(); + + /// Records the status rows of a fresh OnWriteComplete callback for the given handle pair. + /// MXAccess server handle. + /// MXAccess item handle. + /// Status rows from the mapped OnWriteComplete event; cloned before storing. + public void Record( + int serverHandle, + int itemHandle, + RepeatedField statuses) + { + if (statuses is null) + { + throw new ArgumentNullException(nameof(statuses)); + } + + lock (syncRoot) + { + long key = CreateItemKey(serverHandle, itemHandle); + ulong version = entries.TryGetValue(key, out CompletionEntry existing) + ? existing.Version + 1 + : 1UL; + entries[key] = new CompletionEntry(version, statuses.Clone()); + } + } + + /// Returns the current completion version for a handle pair, or 0 if none was recorded. + /// MXAccess server handle. + /// MXAccess item handle. + /// The current completion version, or 0 if no completion was recorded. + public ulong CurrentVersion( + int serverHandle, + int itemHandle) + { + lock (syncRoot) + { + return entries.TryGetValue(CreateItemKey(serverHandle, itemHandle), out CompletionEntry existing) + ? existing.Version + : 0UL; + } + } + + /// + /// Polls for a completion newer than until it + /// arrives or the deadline elapses, calling on every + /// poll iteration so the worker's STA can dispatch the inbound MXAccess + /// OnWriteComplete message. Same loop shape as + /// . + /// + /// MXAccess server handle. + /// MXAccess item handle. + /// Version snapshot captured before the write COM call. + /// Absolute UTC deadline. + /// Action that pumps any pending Windows messages. + /// The recorded status rows if a completion arrived before the deadline; empty otherwise. + /// How long to sleep between pump cycles. Default 5 ms. + /// if a completion newer than arrived before the deadline; otherwise . + public bool TryWaitForCompletion( + int serverHandle, + int itemHandle, + ulong sinceVersion, + DateTime deadlineUtc, + Action pumpStep, + out RepeatedField statuses, + int pollIntervalMs = 5) + { + if (pumpStep is null) + { + throw new ArgumentNullException(nameof(pumpStep)); + } + + while (true) + { + pumpStep(); + + lock (syncRoot) + { + if (entries.TryGetValue(CreateItemKey(serverHandle, itemHandle), out CompletionEntry entry) + && entry.Version > sinceVersion) + { + statuses = entry.Statuses; + return true; + } + } + + if (DateTime.UtcNow >= deadlineUtc) + { + statuses = new RepeatedField(); + return false; + } + + Thread.Sleep(pollIntervalMs); + } + } + + private static long CreateItemKey( + int serverHandle, + int itemHandle) + { + return ((long)serverHandle << 32) | (uint)itemHandle; + } + + /// + /// Snapshot of the most recent OnWriteComplete status rows for a handle + /// pair. increments by one on every + /// call so the write executor can detect "a new + /// completion arrived since I captured my baseline". + /// + /// + /// Plain readonly struct (not a record) so this compiles under the + /// worker's net48 target, which lacks IsExternalInit. + /// + private readonly struct CompletionEntry + { + public CompletionEntry( + ulong version, + RepeatedField statuses) + { + Version = version; + Statuses = statuses; + } + + public ulong Version { get; } + + public RepeatedField Statuses { get; } + } +}