diff --git a/src/ZB.MOM.WW.MxGateway.Worker.Tests/MxAccess/MxAccessCommandExecutorTests.cs b/src/ZB.MOM.WW.MxGateway.Worker.Tests/MxAccess/MxAccessCommandExecutorTests.cs index a8ac61d..7533db1 100644 --- a/src/ZB.MOM.WW.MxGateway.Worker.Tests/MxAccess/MxAccessCommandExecutorTests.cs +++ b/src/ZB.MOM.WW.MxGateway.Worker.Tests/MxAccess/MxAccessCommandExecutorTests.cs @@ -669,6 +669,107 @@ public sealed class MxAccessCommandExecutorTests Assert.Contains("RemoveItem:90:900", fakeComObject.OperationNames); } + /// + /// Verifies ReadBulk's cached fast path end to end — the + /// was_cached = true half of the command, which nothing else in + /// the worker suite exercises. With the tag already added AND advised + /// and a value in the per-session cache, the executor answers from the + /// cache: the result is successful and flagged cached, and no COM call + /// is made for the read at all, so the subscription the caller did not + /// create is left exactly as it was. + /// + /// It also pins the borrow contract through the whole read path: the + /// Value, SourceTimestamp, and status row on the + /// BulkReadResult are the cached event's own instances, not + /// copies (see ). The worker only + /// serializes them onto the IPC pipe, so the alias never escapes the + /// process. + /// + /// Driven through directly rather + /// than because seeding the cache needs + /// a handle on it: only shares the + /// sink's cache for the production MxAccessBaseEventSink, which + /// casts the COM object to LMXProxyServerClass and so cannot take + /// a fake, while CreateForTesting accepts the cache directly. The + /// cached path neither waits nor pumps, so it needs no STA. + /// + [Fact] + public void Execute_ReadBulk_WhenTagIsAdvisedAndCached_ServesTheCachedInstancesWithoutTouchingTheSubscription() + { + FakeMxAccessComObject fakeComObject = new( + registerHandle: 92, + addItemHandle: 920); + MxAccessValueCache valueCache = new(); + using MxAccessSession session = MxAccessSession.CreateForTesting( + mxAccessServer: fakeComObject, + eventSink: new NoopEventSink(), + valueCache: valueCache); + MxAccessCommandExecutor executor = new( + session, + new ZB.MOM.WW.MxGateway.Worker.Conversion.VariantConverter()); + + // Registry half of the fast path: the tag must resolve to a live item + // handle on this server AND carry an advice, or TryGetCachedReadFor + // falls through to the AddItem/Advise snapshot lifecycle. + MxCommandReply registerReply = executor.Execute( + CreateRegisterCommand("register-before-cached-read", "client-a")); + MxCommandReply addItemReply = executor.Execute( + CreateAddItemCommand("add-before-cached-read", 92, "Galaxy.Tag.Value")); + MxCommandReply adviseReply = executor.Execute( + CreateAdviseCommand("advise-before-cached-read", 92, 920)); + Assert.Equal(ProtocolStatusCode.Ok, registerReply.ProtocolStatus.Code); + Assert.Equal(ProtocolStatusCode.Ok, addItemReply.ProtocolStatus.Code); + Assert.Equal(ProtocolStatusCode.Ok, adviseReply.ProtocolStatus.Code); + + // Cache half: stand in for the event sink's post-publish hook, which + // records the event it just enqueued. + MxEvent cachedEvent = new() + { + Family = MxEventFamily.OnDataChange, + ServerHandle = 92, + ItemHandle = 920, + Quality = 192, + SourceTimestamp = Timestamp.FromDateTime(new(2026, 8, 15, 10, 0, 0, DateTimeKind.Utc)), + Value = new MxValue + { + DataType = MxDataType.Integer, + VariantType = "VT_I4", + Int32Value = 7788, + }, + OnDataChange = new OnDataChangeEvent(), + }; + cachedEvent.Statuses.Add(new MxStatusProxy { Category = MxStatusCategory.Ok }); + valueCache.Set(92, 920, cachedEvent); + + MxCommandReply reply = executor.Execute(CreateReadBulkCommand( + "read-bulk-cached", + serverHandle: 92, + tagAddresses: new[] { "Galaxy.Tag.Value" }, + timeoutMs: 80)); + + Assert.Equal(ProtocolStatusCode.Ok, reply.ProtocolStatus.Code); + Assert.Equal(MxCommandKind.ReadBulk, reply.Kind); + BulkReadResult result = Assert.Single(reply.ReadBulk.Results); + Assert.True(result.WasSuccessful); + Assert.True(result.WasCached); + Assert.Equal("Galaxy.Tag.Value", result.TagAddress); + Assert.Equal(920, result.ItemHandle); + Assert.Equal(192, result.Quality); + Assert.Equal(7788, result.Value.Int32Value); + + // Borrowed, not copied — all the way from the event handed to + // MxAccessValueCache.Set out to the reply the worker serializes. + Assert.Same(cachedEvent.Value, result.Value); + Assert.Same(cachedEvent.SourceTimestamp, result.SourceTimestamp); + Assert.Same(cachedEvent.Statuses[0], Assert.Single(result.Statuses)); + + // No second AddItem, and above all no UnAdvise/RemoveItem: only the + // three setup calls ever reached MXAccess. + Assert.Equal( + new[] { "Register:client-a", "AddItem:92:Galaxy.Tag.Value", "Advise:92:920" }, + fakeComObject.OperationNames); + } + /// Verifies that ReadBulk with no payload returns an invalid request error. /// A task that represents the asynchronous operation. [Fact] diff --git a/src/ZB.MOM.WW.MxGateway.Worker.Tests/MxAccess/MxAccessValueCacheTests.cs b/src/ZB.MOM.WW.MxGateway.Worker.Tests/MxAccess/MxAccessValueCacheTests.cs index bad8580..2b9151e 100644 --- a/src/ZB.MOM.WW.MxGateway.Worker.Tests/MxAccess/MxAccessValueCacheTests.cs +++ b/src/ZB.MOM.WW.MxGateway.Worker.Tests/MxAccess/MxAccessValueCacheTests.cs @@ -48,14 +48,28 @@ public sealed class MxAccessValueCacheTests } /// - /// Verifies that Set stores an independent deep-copied snapshot: mutating - /// the source event's protobuf sub-messages after caching does not alter - /// the cached value. WRK-11 stopped the event sink cloning before enqueue, - /// so the same MxEvent instance now flows to the outbound queue; the cache - /// must own its own copy so the two never share mutable state. + /// Pins the ownership contract: Set borrows the event's own + /// protobuf sub-messages instead of deep-copying them, so TryGet + /// hands back the very Value, SourceTimestamp, and + /// MxStatusProxy instances the caller passed in. + /// + /// This is safe only because the event is write-once by the time + /// Set runs: the sink enqueues it (which stamps the worker + /// sequence and timestamp) and only then post-publishes it here, and + /// 's ownership invariant forbids + /// mutating an enqueued event. This test therefore asserts reference + /// identity and deliberately does NOT mutate the event afterwards — + /// doing so is exactly what the contract forbids, and it would also + /// invalidate the serialized size the queue memoized at enqueue. + /// + /// It replaced a test asserting the opposite (an independent deep-copied + /// snapshot). The three clones that test pinned — the MxValue, the + /// Timestamp, and the RepeatedField plus every status row in it — ran on + /// every OnDataChange and bought nothing: the read path already aliased + /// the cache's instances into every BulkReadResult. /// [Fact] - public void Set_StoresIndependentSnapshot_UnaffectedByLaterEventMutation() + public void Set_BorrowsTheEventsOwnInstances_ByOwnershipContract() { MxAccessValueCache cache = new(); Timestamp sourceTimestamp = Timestamp.FromDateTime(new(2026, 5, 19, 9, 0, 0, DateTimeKind.Utc)); @@ -63,19 +77,12 @@ public sealed class MxAccessValueCacheTests cache.Set(7, 21, mxEvent); - // Mutate the event in place after it was cached — as if it kept flowing - // through the (unrelated) outbound path. None of this must reach the cache. - mxEvent.Value.Int32Value = 999; - mxEvent.Quality = 0; - mxEvent.SourceTimestamp = Timestamp.FromDateTime(new(2030, 1, 1, 0, 0, 0, DateTimeKind.Utc)); - mxEvent.Statuses[0].Category = MxStatusCategory.SecurityError; - Assert.True(cache.TryGet(7, 21, out MxAccessValueCache.CachedValue cached)); - Assert.Equal(100, cached.Value.Int32Value); + Assert.Same(mxEvent.Value, cached.Value); + Assert.Same(mxEvent.SourceTimestamp, cached.SourceTimestamp); + Assert.Same(mxEvent.Statuses, cached.Statuses); + Assert.Same(mxEvent.Statuses[0], Assert.Single(cached.Statuses)); Assert.Equal(192, cached.Quality); - Assert.Equal(sourceTimestamp, cached.SourceTimestamp); - Assert.Single(cached.Statuses); - Assert.Equal(MxStatusCategory.Ok, cached.Statuses[0].Category); } /// Verifies that TryGet returns false for unknown handles. diff --git a/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/MxAccessEventQueue.cs b/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/MxAccessEventQueue.cs index dd07094..eb4c471 100644 --- a/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/MxAccessEventQueue.cs +++ b/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/MxAccessEventQueue.cs @@ -20,8 +20,11 @@ namespace ZB.MOM.WW.MxGateway.Worker.MxAccess; /// that it does not retain, reuse, or mutate after the /// call returns. All production callers (MxAccessBaseEventSink, /// MxAccessAlarmEventSink, AlarmCommandHandler) build a new event per -/// Enqueue via the mapper and satisfy this; the value cache stores its own -/// independent snapshot (see ). +/// Enqueue via the mapper and satisfy this. "Does not mutate" is the load- +/// bearing half for the post-publish value cache, which deliberately borrows +/// the enqueued event's own value/timestamp/status instances rather than +/// copying them (see ) — it reads and +/// serializes them, and this invariant is what keeps that safe. /// /// The byte-budgeted relies on the same invariant: each /// event's serialized size is measured once at enqueue and stored beside it, which is diff --git a/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/MxAccessValueCache.cs b/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/MxAccessValueCache.cs index b6ecd3a..3841d5f 100644 --- a/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/MxAccessValueCache.cs +++ b/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/MxAccessValueCache.cs @@ -57,7 +57,12 @@ public sealed class MxAccessValueCache /// Records a fresh OnDataChange payload for the given handle pair. /// MXAccess server handle. /// MXAccess item handle. - /// The protobuf MxEvent created by the event mapper. + /// + /// The protobuf MxEvent created by the event mapper, already handed to + /// the outbound queue. The cache borrows this event's + /// Value/SourceTimestamp/Statuses instances rather + /// than copying them — see the ownership contract in the method body. + /// public void Set( int serverHandle, int itemHandle, @@ -68,20 +73,39 @@ public sealed class MxAccessValueCache throw new ArgumentNullException(nameof(mxEvent)); } - // WRK-11: the event sink no longer clones before enqueue, so the passed - // mxEvent is the very instance handed to the outbound queue. Deep-copy - // the value/timestamp/statuses payload we retain here so the cache's - // snapshot stays independent of the enqueued (and later serialized) - // event — the two must never share mutable protobuf sub-messages. - // Value is always set for OnDataChange; SourceTimestamp may be unset when - // the source timestamp could not be parsed, so both are cloned only when - // present. The null-forgiving result matches CachedValue's non-null- - // annotated parameters, which already accepted a runtime-null value or - // timestamp before WRK-11 (the ternary keeps the compiler's null-state - // from poisoning to maybe-null, which a plain null check would do). - MxValue cachedValue = mxEvent.Value is null ? null! : mxEvent.Value.Clone(); - Timestamp cachedTimestamp = mxEvent.SourceTimestamp is null ? null! : mxEvent.SourceTimestamp.Clone(); - + // Ownership contract: the cache BORROWS, it does not copy. CachedValue + // retains the event's own Value, SourceTimestamp, and Statuses + // instances. + // + // Sound because the event is write-once by the time this runs. + // MxAccessBaseEventSink.EnqueueEvent calls eventQueue.Enqueue first — + // which stamps WorkerSequence/WorkerTimestamp inside the queue lock — + // and only then runs the postPublish hook that lands here; the queue's + // ownership invariant (see MxAccessEventQueue's class remarks) forbids + // mutating an event after it is enqueued. The producer side never + // reuses instances either: the mapper builds a fresh MxEvent, and the + // VariantConverter a fresh MxValue, per COM callback. The three deep + // copies this replaced (the MxValue — recursive for an MxArray — the + // Timestamp, and the RepeatedField container plus every MxStatusProxy + // in it) therefore bought nothing but garbage on the worker's hottest + // path. + // + // Consumers of TryGet / TryWaitForUpdate may read and serialize what + // they get back; they must never mutate it. ReadBulk already depends on + // that: MxAccessSession.SucceededRead puts these very instances on the + // BulkReadResult it returns, which the worker only serializes onto the + // IPC pipe — worker↔gateway is a process boundary, so no gateway-side + // consumer can alias them. Mutating one would corrupt the event still + // queued for the outbound stream AND invalidate QueuedEvent.Size, the + // serialized size memoized at enqueue that the byte-budgeted Drain + // charges against its budget: a message grown after enqueue could + // overshoot the negotiated frame max and fault the session with + // MessageTooLarge (WorkerPipeSession.FaultOnOversizedEventAsync). + // + // Value is always set for OnDataChange; SourceTimestamp can be null when + // the source timestamp could not be parsed. CachedValue's parameters are + // annotated non-null but have always accepted a runtime null for both, + // and the read side null-checks accordingly. long key = CreateItemKey(serverHandle, itemHandle); lock (syncRoot) { @@ -91,10 +115,10 @@ public sealed class MxAccessValueCache entries[key] = new CachedValue( nextVersion, - cachedValue, + mxEvent.Value, mxEvent.Quality, - cachedTimestamp, - mxEvent.Statuses.Clone()); + mxEvent.SourceTimestamp, + mxEvent.Statuses); } // Signaled outside the lock: the waiter re-takes syncRoot (through @@ -228,23 +252,35 @@ public sealed class MxAccessValueCache } /// - /// Snapshot of the most recent OnDataChange payload for a handle pair. + /// The most recent OnDataChange payload for a handle pair. /// increments by one on every /// call so the bulk read executor can detect "a new value arrived /// since I started waiting". /// /// - /// Plain readonly struct (not a record) so this compiles under the - /// worker's net48 target, which lacks IsExternalInit. + /// + /// Borrowed references, not copies. , + /// , and are the + /// very instances hanging off the MxEvent that was enqueued for the + /// outbound stream — carries the full ownership + /// contract. Read them and serialize them; never mutate them and + /// never hand them to something that will. That event is write-once + /// from the moment it is enqueued, and its serialized size is + /// memoized at that point for the byte-budgeted drain. + /// + /// + /// Plain readonly struct (not a record) so this compiles under the + /// worker's net48 target, which lacks IsExternalInit. + /// /// public readonly struct CachedValue { /// Initializes a new cached value snapshot. /// Version counter incremented on each update. - /// The MXAccess value. + /// The MXAccess value, borrowed from the enqueued event. /// The MXAccess quality code. - /// The source timestamp of the value. - /// The MXAccess status codes. + /// The source timestamp of the value, borrowed from the enqueued event. + /// The MXAccess status codes, borrowed from the enqueued event. public CachedValue( ulong version, MxValue value, @@ -262,16 +298,16 @@ public sealed class MxAccessValueCache /// Monotonic per-handle version counter. public ulong Version { get; } - /// The cached MxValue payload. + /// The OnDataChange event's own MxValue payload. Read-only to consumers. public MxValue Value { get; } /// Quality code from the OnDataChange event. public int Quality { get; } - /// Source timestamp from the OnDataChange event. + /// The OnDataChange event's own source timestamp. Read-only to consumers. public Timestamp SourceTimestamp { get; } - /// MxStatusProxy entries from the OnDataChange event. + /// The OnDataChange event's own MxStatusProxy collection. Read-only to consumers. public RepeatedField Statuses { get; } } } diff --git a/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/MxAccessWriteCompletionCache.cs b/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/MxAccessWriteCompletionCache.cs index 78a4114..f846cd2 100644 --- a/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/MxAccessWriteCompletionCache.cs +++ b/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/MxAccessWriteCompletionCache.cs @@ -73,6 +73,19 @@ public sealed class MxAccessWriteCompletionCache ulong version = entries.TryGetValue(key, out CompletionEntry existing) ? existing.Version + 1 : 1UL; + + // Still a defensive copy, deliberately — this is NOT an oversight + // left behind by the borrow that MxAccessValueCache.Set adopted (see + // the ownership contract there). The value cache is handed the whole + // enqueued MxEvent, so the queue's write-once ownership invariant + // covers everything it retains. This method is handed a bare + // RepeatedField whose provenance its signature cannot constrain: + // the production sink does pass an enqueued event's Statuses, but + // callers that build and keep their own rows are equally valid + // against this API, and a borrowed alias would then let a later + // mutation rewrite an already-recorded completion. The write path is + // command-rate, not the per-OnDataChange streaming hot path, so the + // clone costs nothing worth reclaiming. entries[key] = new CompletionEntry(version, statuses.Clone()); }