diff --git a/docs/MxAccessWorkerInstanceDesign.md b/docs/MxAccessWorkerInstanceDesign.md index 5af663e..b648b1a 100644 --- a/docs/MxAccessWorkerInstanceDesign.md +++ b/docs/MxAccessWorkerInstanceDesign.md @@ -351,6 +351,17 @@ native client correlating by handle has the identical ambiguity). The exposure is bounded: `statuses` is an enrichment field, and `Hresult`/`ProtocolStatus` stay acceptance-only regardless of which completion (if any) was attributed. +**The alarm ack writer narrows the window for acknowledgements.** +`AlarmAckWriter` registers its own MXAccess server with its own +`MxAccessWriteCompletionCache` and reuses one advised `AckMsg` item per alarm +(least-recently-used, capped at 256). After an ack ends unconfirmed it marks the +item dirty, and the next ack to that alarm removes and re-adds the item so the +write waits on a fresh item handle the stale completion cannot match. Every +drop of an item (dirty re-add, eviction, or a failed `AddItem`/`Advise`/`Write`) +also removes that handle's completion entry, and tearing down the registration +removes every entry for its server handle, so the cache does not grow by one +entry per dropped handle for the life of the worker. + ## Live alarm watch-list replacement An auto-mode `SubscribeAlarms` with `fallback_enabled = true` always builds a diff --git a/src/ZB.MOM.WW.MxGateway.Worker.Tests/MxAccess/AlarmAckWriterTests.cs b/src/ZB.MOM.WW.MxGateway.Worker.Tests/MxAccess/AlarmAckWriterTests.cs index 0714173..48960bc 100644 --- a/src/ZB.MOM.WW.MxGateway.Worker.Tests/MxAccess/AlarmAckWriterTests.cs +++ b/src/ZB.MOM.WW.MxGateway.Worker.Tests/MxAccess/AlarmAckWriterTests.cs @@ -333,6 +333,76 @@ public sealed class AlarmAckWriterTests Assert.Empty(server.RemovedItemHandles); } + [Fact] + public void Acknowledge_AfterDirtyReAdd_ForgetsDroppedHandleCompletion() + { + RecordingAckServer server = new RecordingAckServer { AuthenticateResult = 5 }; + MxAccessWriteCompletionCache cache = new MxAccessWriteCompletionCache(); + using AlarmAckWriter writer = new AlarmAckWriter(server, cache, () => { }, TimeSpan.FromMilliseconds(20), "MxGateway"); + + writer.Acknowledge("T.A.AckMsg", "one", Alice()); // handle 100, Unconfirmed + cache.Record(7, 100, Rows(MxStatusCategory.Ok, 0)); // its late completion + Assert.Equal(1, cache.Count); + server.OnWrite = (handle, item) => cache.Record(handle, item, Rows(MxStatusCategory.Ok, 0)); + + AlarmAckResult second = writer.Acknowledge("T.A.AckMsg", "two", Alice()); // drops 100, adds 101 + + Assert.Equal(AlarmAckOutcome.Applied, second.Outcome); + Assert.Equal(1, cache.Count); + Assert.Equal(0UL, cache.CurrentVersion(7, 100)); + Assert.Equal(1UL, cache.CurrentVersion(7, 101)); + } + + [Fact] + public void Acknowledge_PastCacheCap_ForgetsEvictedHandleCompletion() + { + RecordingAckServer server = new RecordingAckServer { AuthenticateResult = 5 }; + MxAccessWriteCompletionCache cache = new MxAccessWriteCompletionCache(); + server.OnWrite = (handle, item) => cache.Record(handle, item, Rows(MxStatusCategory.Ok, 0)); + using AlarmAckWriter writer = new AlarmAckWriter(server, cache, () => { }, TimeSpan.FromSeconds(1), "MxGateway", maxCachedItems: 2); + + writer.Acknowledge("A.X.AckMsg", "1", Alice()); // handle 100 + writer.Acknowledge("B.X.AckMsg", "2", Alice()); // handle 101 + Assert.Equal(2, cache.Count); + writer.Acknowledge("C.X.AckMsg", "3", Alice()); // handle 102, evicts A (100) + + Assert.Equal(2, cache.Count); + Assert.Equal(0UL, cache.CurrentVersion(7, 100)); + } + + [Fact] + public void Dispose_ForgetsServerCompletions() + { + RecordingAckServer server = new RecordingAckServer { AuthenticateResult = 5 }; + MxAccessWriteCompletionCache cache = new MxAccessWriteCompletionCache(); + server.OnWrite = (handle, item) => cache.Record(handle, item, Rows(MxStatusCategory.Ok, 0)); + AlarmAckWriter writer = new AlarmAckWriter(server, cache, () => { }, TimeSpan.FromSeconds(1), "MxGateway"); + writer.Acknowledge("A.X.AckMsg", "1", Alice()); + writer.Acknowledge("B.X.AckMsg", "2", Alice()); + cache.Record(8, 100, Rows(MxStatusCategory.Ok, 0)); // another server's entry survives + + writer.Dispose(); + + Assert.Equal(1, cache.Count); + Assert.Equal(1UL, cache.CurrentVersion(8, 100)); + } + + [Fact] + public void Acknowledge_WhenWriteDisconnected_ForgetsServerCompletions() + { + RecordingAckServer server = new RecordingAckServer { AuthenticateResult = 5 }; + MxAccessWriteCompletionCache cache = new MxAccessWriteCompletionCache(); + server.OnWrite = (handle, item) => cache.Record(handle, item, Rows(MxStatusCategory.Ok, 0)); + using AlarmAckWriter writer = new AlarmAckWriter(server, cache, () => { }, TimeSpan.FromSeconds(1), "MxGateway"); + writer.Acknowledge("A.X.AckMsg", "1", Alice()); // handle 100 recorded + server.WriteException = new COMException("x", unchecked((int)0x80010108)); + + writer.Acknowledge("A.X.AckMsg", "2", Alice()); + + Assert.Contains("Unregister", server.Operations); + Assert.Equal(0, cache.Count); + } + [Fact] public void Acknowledge_WhenNonOkCompletionHasNoDetail_ReturnsMinusOne() { diff --git a/src/ZB.MOM.WW.MxGateway.Worker.Tests/MxAccess/MxAccessWriteCompletionCacheTests.cs b/src/ZB.MOM.WW.MxGateway.Worker.Tests/MxAccess/MxAccessWriteCompletionCacheTests.cs index 8d4b1c8..c086c10 100644 --- a/src/ZB.MOM.WW.MxGateway.Worker.Tests/MxAccess/MxAccessWriteCompletionCacheTests.cs +++ b/src/ZB.MOM.WW.MxGateway.Worker.Tests/MxAccess/MxAccessWriteCompletionCacheTests.cs @@ -183,6 +183,39 @@ public sealed class MxAccessWriteCompletionCacheTests Assert.Equal(77, Assert.Single(statuses).Detail); } + /// Verifies that Remove forgets only the given handle pair. + [Fact] + public void Remove_ForgetsOnlyThatHandlePair() + { + MxAccessWriteCompletionCache cache = new(); + cache.Record(7, 21, BuildStatuses(detail: 1)); + cache.Record(7, 22, BuildStatuses(detail: 2)); + + cache.Remove(7, 21); + cache.Remove(7, 99); // unknown pair is a no-op + + Assert.Equal(1, cache.Count); + Assert.Equal(0UL, cache.CurrentVersion(7, 21)); + Assert.Equal(1UL, cache.CurrentVersion(7, 22)); + } + + /// Verifies that RemoveServer forgets every pair of that server and no other. + [Fact] + public void RemoveServer_ForgetsOnlyThatServersPairs() + { + MxAccessWriteCompletionCache cache = new(); + cache.Record(7, 21, BuildStatuses(detail: 1)); + cache.Record(7, -5, BuildStatuses(detail: 2)); + cache.Record(8, 21, BuildStatuses(detail: 3)); + cache.Record(-7, 21, BuildStatuses(detail: 4)); + + cache.RemoveServer(7); + + Assert.Equal(2, cache.Count); + Assert.Equal(1UL, cache.CurrentVersion(8, 21)); + Assert.Equal(1UL, cache.CurrentVersion(-7, 21)); + } + private static RepeatedField BuildStatuses(int detail) { return new RepeatedField diff --git a/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/AlarmAckWriter.cs b/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/AlarmAckWriter.cs index eed231d..1538607 100644 --- a/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/AlarmAckWriter.cs +++ b/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/AlarmAckWriter.cs @@ -303,6 +303,9 @@ public sealed class AlarmAckWriter : IAlarmAckWriter try { mx.Unregister(serverHandle); } catch { /* best effort */ } } + // The registration is gone, so no completion can arrive for its handles. + completionCache.RemoveServer(serverHandle); + itemHandles.Clear(); itemLru.Clear(); server = null; @@ -335,6 +338,10 @@ public sealed class AlarmAckWriter : IAlarmAckWriter itemLru.Remove(node); try { mx.UnAdvise(serverHandle, node.Value.Handle); } catch { /* best effort */ } try { mx.RemoveItem(serverHandle, node.Value.Handle); } catch { /* best effort */ } + + // Without this the completion cache keeps one entry per dropped handle + // (each Unconfirmed re-add and each eviction) for the worker's life. + completionCache.Remove(serverHandle, node.Value.Handle); } private IMxAccessServer EnsureRegistered() diff --git a/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/MxAccessWriteCompletionCache.cs b/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/MxAccessWriteCompletionCache.cs index 9488e8b..54ed955 100644 --- a/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/MxAccessWriteCompletionCache.cs +++ b/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/MxAccessWriteCompletionCache.cs @@ -184,6 +184,57 @@ public sealed class MxAccessWriteCompletionCache } } + /// + /// Forgets the completion entry of a handle pair whose MXAccess item was + /// removed, so a long-lived writer that drops and re-adds items does not + /// keep one entry per dropped handle for the life of the worker. + /// + /// MXAccess server handle. + /// MXAccess item handle that was removed. + public void Remove( + int serverHandle, + int itemHandle) + { + lock (syncRoot) + { + entries.Remove(CreateItemKey(serverHandle, itemHandle)); + } + } + + /// Forgets every completion entry recorded under a server handle that was unregistered. + /// MXAccess server handle that was unregistered. + public void RemoveServer(int serverHandle) + { + lock (syncRoot) + { + List stale = new(); + foreach (long key in entries.Keys) + { + if ((int)(key >> 32) == serverHandle) + { + stale.Add(key); + } + } + + foreach (long key in stale) + { + entries.Remove(key); + } + } + } + + /// Gets the number of handle pairs with a recorded completion (test seam). + internal int Count + { + get + { + lock (syncRoot) + { + return entries.Count; + } + } + } + private static long CreateItemKey( int serverHandle, int itemHandle)