From 1622d07f2ef27b1670a9f1e83715624438c5871c Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Wed, 11 Nov 2026 06:37:22 -0500 Subject: [PATCH 1/3] fix(opcua): ConditionRefresh on a Server-object item is not overtaken by older deferred condition events (#854) While a folder subscriber's SDK channel is full, a condition's later events wait in its #838 deferral queue and reach i=2253 items only when the drain gets to them, off Lock. A ConditionRefresh in that window replayed the current state, then the older deferred events followed RefreshEnd (stale state, or a condition the refresh had just dropped re-added until the newest drained). - Condition events are reported as ConditionEventSnapshot (an InstanceStateSnapshot carrying a report sequence, condition NodeId and Retain); events queued behind a deferred report are marked Deferred. - Server-object refresh (ConditionRefresh and ConditionRefresh2, flag on AND off) now runs in RefreshServerObjectItems: watermark, "anything deferred?", the replay and its queueing in ONE Lock hold, so the replay is exactly the state at the cut. Per item it queues the replay and installs a RefreshCut under the item's gate. - OnConditionReportEvent holds that gate across "superseded? -> QueueEvent" for deferred events only. Withheld from that item alone: events at/below the cut whose condition the replay re-sent, or Retain=true events of a condition it omitted; an older Retain=false event of an omitted condition is still delivered. Queues, other Server-object items and folder items are untouched; refresh never waits for a drain. - Lock order Lock -> item gate -> SDK item lock, same as a synchronous report. - Meter otopcua.opcua.condition_events.superseded. - Folder items keep a documented residual (SDK channel backlog + memoised per-item verdict leave no per-event hook). Tests: ConditionRefreshDeferralWireTests (deterministic overload by holding the folder item's SDK lock; 6 variants fail before the fix; probe-driven gate test fails with the gate removed; concurrent reports/refreshes/stall stress test fails before the fix). --- .claude/rules/data-plane-auth.md | 6 +- docs/security.md | 31 +- .../Observability/OtOpcUaTelemetry.cs | 9 + .../OtOpcUaNodeManager.cs | 247 ++++++++-- .../ConditionRefreshDeferralWireTests.cs | 430 ++++++++++++++++++ 5 files changed, 680 insertions(+), 43 deletions(-) create mode 100644 tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/ConditionRefreshDeferralWireTests.cs diff --git a/.claude/rules/data-plane-auth.md b/.claude/rules/data-plane-auth.md index e2030b98a..62a185a44 100644 --- a/.claude/rules/data-plane-auth.md +++ b/.claude/rules/data-plane-auth.md @@ -211,8 +211,10 @@ Two things to know before relying on any of this: through `ReportEventWithoutBlocking`: our own copy of the SDK walk with a LIVENESS check before every hop (and in `OnConditionReportEvent`) — a deferred event of a removed/replaced condition is dropped, never a ghost on `i=2253`; per-CONDITION ordered queues on overflow; coalesce past 1024/condition or 65 536 total (oldest not-started gives way - to newest); Warning/Information + `otopcua.opcua.condition_events.{deferred,discarded}`. ConditionRefresh does not - flush the queues (documented residual). Folder events follow Part 3's shape — + to newest); Warning/Information + `otopcua.opcua.condition_events.{deferred,discarded}`. A Server-object ConditionRefresh + (and ConditionRefresh2) takes a sequence CUT in one `Lock` hold and withholds older deferred events from THAT item + only (re-sent condition, or a Retain=true event of an omitted one) under a per-item gate held across check+QueueEvent + (#854, `ConditionRefreshDeferralWireTests`); folder items keep the residual (SDK channel + memoised verdict). Folder events follow Part 3's shape — `ReceiveEvents` on EventType + SourceNode (SDK check, kept first) AND our decision; non-condition events on a folder item fail closed under enforcement. Tests: `FolderEventFloodWireTests`. - **ns=0 standard nodes stay ungated by design, even with `Security:Acl:Enabled`** (#617) — `ServerStatus`, diff --git a/docs/security.md b/docs/security.md index f82f9bd0b..5c6d21880 100644 --- a/docs/security.md +++ b/docs/security.md @@ -882,15 +882,38 @@ Behaviours worth knowing before enabling it: changes which folders a later hop reaches — and the liveness check covers removal); `MonitoredNode2.Dispose` completes its channel writer, which fails a pending `WriteAsync` with `ChannelClosedException` (swallowed by the SDK), so a disposed node cannot stall a condition's queue; and the delivery task catches per-event exceptions, so it keeps draining. - - ⚠️ **Residual (documented, not fixed): `ConditionRefresh` does not flush deferrals.** While a notifier channel is - overloaded, a client that refreshes may receive older deferred events of a condition after `RefreshEndEvent`. They - are still in that condition's order and its newest event arrives last, so the state a client ends up holding is - correct; making refresh wait for the queues would block a service call on the very drain the overload is stalling. + - **`ConditionRefresh` is not overtaken by deferred events on Server-object items (Gitea #854).** A refresh of an + `i=2253` item replays each condition's current state between RefreshStart and RefreshEnd; the condition's deferred + events — all older — used to follow RefreshEnd, briefly showing a stale state or re-adding an alarm the refresh had + just dropped (ScadaBridge's snapshot swap drops absent conditions, then a stale active event re-added it). Now every + condition event carries its report sequence; the refresh takes its **cut** (the sequence, whether anything is + deferred, the replay, and its queueing) in **one hold of `Lock`**, so the replay is exactly the state at the cut, and + installs it per item under that item's gate, which a deferred delivery also holds across its check and + `QueueEvent`. A deferred event at or below the cut is withheld **from that item only** when the refresh re-sent its + condition, or when it is `Retain=true` for a condition the refresh omitted; an older `Retain=false` event of an + omitted condition still arrives (it agrees with the omission, and a client that does not reconcile on RefreshEnd + needs it). The deferral queues, every other Server-object item and every folder item are untouched; the refresh never + waits for a drain. Meter: `otopcua.opcua.condition_events.superseded` (per item and event). Both ConditionRefresh and + ConditionRefresh2, flag on and off. + - ⚠️ **Residual: folder (notifier) items.** A refresh of an item on an equipment/device folder can still be followed + by older events of a condition. Two sources, neither reachable per item: the SDK's own `MonitoredNode2` channel + (up to 4096 events already accepted before the refresh — this exists in the stock SDK too, and in normal operation + for the few events in transit), and our deferral queue, whose folder hop goes through that channel, where the SDK + memoises the per-item permission verdict by (item, event type, source) so there is no per-event, per-item hook. The + events stay in the condition's order and its newest arrives last, so the client converges. Discarding the queue + globally on refresh would drop transitions for every non-refreshing subscriber while leaving the channel backlog; + making refresh wait would block a service call on the stalled drain. ScadaBridge also ignores a transition older + than the one it holds for a condition. Pinned by `FolderEventFloodWireTests`: 6000 events under one `Lock` hold with publishing disabled (flag on and off) complete, drain to zero, coalesce, and the item delivers at most its revised `QueueSize` in increasing order ending with the last event; and a condition **removed** in the same `Lock` hold has none of its deferred events delivered to either the folder or the Server-object item (the flag-off variant fails without the liveness checks). + `ConditionRefreshDeferralWireTests` (#854) stalls a folder item's SDK lock so its channel fills without `Lock` held, + refreshes the Server-object item (ConditionRefresh and ConditionRefresh2, flag on/off, condition retained or cleared) + and pins that nothing older than the replay follows RefreshEnd while the non-refreshing folder item still receives the + whole tail; a probe-driven test pins that a refresh waits for a delivery already past its check (fails without the + item gate); and a stress test runs reports, refreshes and an intermittent stall concurrently. **Model-change and audit events on the Server object are gated too** (Gitea #588). The node manager's own server-wide events — `GeneralModelChangeEvent` (NodeAdded / NodeDeleted / DataTypeChanged) and the diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Commons/Observability/OtOpcUaTelemetry.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Commons/Observability/OtOpcUaTelemetry.cs index e71cec9ef..b46012110 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Commons/Observability/OtOpcUaTelemetry.cs +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Commons/Observability/OtOpcUaTelemetry.cs @@ -172,6 +172,15 @@ public static class OtOpcUaTelemetry Meter.CreateCounter("otopcua.opcua.condition_events.discarded", unit: "{event}", description: "Deferred condition events discarded before delivery (reason=coalesced|not_live)."); + /// + /// Deferred condition events withheld from one Server-object (i=2253) item because a ConditionRefresh of that + /// item had already replayed a newer state (Gitea #854) — one per item and event. Every other subscriber still + /// receives them; this only stops a refresh being overtaken by older events. + /// + public static readonly Counter ConditionEventDeliveriesSuperseded = + Meter.CreateCounter("otopcua.opcua.condition_events.superseded", unit: "{delivery}", + description: "Deferred condition events withheld from a Server-object item whose ConditionRefresh already replayed a newer state."); + // ---------------- Convenience helpers ---------------- /// diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs index 0651c8c4f..4458bb50e 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs @@ -1386,7 +1386,8 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3 // in MaterialiseAlarmCondition), whose fan-outs deliver it to direct notifier subscriptions; the // condition's own OnReportEvent hands the same snapshot to the Server-object (all-events) // subscriptions, per item, through the ACL (see OnConditionReportEvent). - var snapshot = new InstanceStateSnapshot(); + var snapshot = new ConditionEventSnapshot( + Interlocked.Increment(ref _conditionEventSequence), alarm.NodeId, alarm.Retain?.Value == true); snapshot.Initialize(SystemContext, alarm); ReportEventWithoutBlocking(alarm, snapshot); } @@ -1434,6 +1435,14 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3 /// Deferred condition events discarded before delivery, any reason (test observation). internal long DiscardedEventReports => Interlocked.Read(ref _discardedEventReports); + private long _refreshSupersededDeliveries; + + /// + /// Deferred condition events withheld from a Server-object item because a ConditionRefresh on that item already + /// delivered a newer state (Gitea #854) — one per item and event (test observation). + /// + internal long RefreshSupersededDeliveries => Interlocked.Read(ref _refreshSupersededDeliveries); + /// /// Reports a condition event through the condition's notifier chain WITHOUT blocking the caller, in the order /// this condition's events were reported, and never for a condition that is no longer live (Gitea #838). @@ -1517,6 +1526,8 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3 /// Queues a later event behind a condition's deferred one, coalescing past the budget. Under the gate. private void EnqueueDeferred(AlarmConditionState condition, DeferredConditionReports queue, IFilterTarget e) { + // Delivered off Lock from here on, so a ConditionRefresh can overtake it (#854, see OnConditionReportEvent). + if (e is ConditionEventSnapshot snapshot) snapshot.Deferred = true; queue.Pending.Enqueue(e); CountDeferred(+1, condition); while (queue.Pending.Count > 1 @@ -1651,6 +1662,65 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3 public readonly Queue Pending = new(); } + /// + /// The report order of condition events (#854): incremented under Lock by , + /// read under Lock by a Server-object ConditionRefresh as its cut. + /// + private long _conditionEventSequence; + + /// + /// A condition event as reports it: the SDK's frozen snapshot, plus what a + /// Server-object ConditionRefresh needs to recognise it as superseded (Gitea #854). Still an + /// , so the SDK's event identity, filtered-retain and queue handling treat it + /// exactly as before. + /// + /// Its place in report order (). + /// The reporting condition's NodeId (not the instance: a cut must not pin a removed node). + /// The condition's Retain as reported. + private sealed class ConditionEventSnapshot(long sequence, NodeId conditionId, bool retain) : InstanceStateSnapshot + { + public long Sequence { get; } = sequence; + + public NodeId ConditionId { get; } = conditionId; + + public bool Retain { get; } = retain; + + /// + /// Set (under the event-report gate, before the drain can dequeue it) when the event waits in a deferral queue: + /// only such an event reaches Server-object items off Lock, which is the only way a refresh can be overtaken. + /// + public bool Deferred { get; set; } + } + + /// + /// Per Server-object item state (#854): the gate that orders a deferred delivery to the item against a refresh of + /// the item, and the cut of the item's latest refresh while deferred events may still be older than it. + /// + private sealed class AllEventsItemState + { + /// Held across "is this deferred event superseded? → QueueEvent", and across a refresh's + /// "QueueEvent the replay → install the cut". Only ever taken innermost (nothing but the item's own lock under it). + public readonly object Gate = new(); + + /// Under . + public RefreshCut? Cut; + } + + /// What a Server-object item's latest ConditionRefresh conveyed (#854). + /// Every condition event with a sequence at or below this was reported before the replay was read. + /// The conditions the replay re-sent to the item; the others were Retain=false (or not visible). + private sealed record RefreshCut(long Watermark, HashSet Refreshed) + { + /// + /// Whether the refresh already told the item something at least as new as : a re-sent + /// condition's current state supersedes every older event of it; for a condition the replay OMITTED (it was not + /// retained), an older Retain=true event would wrongly re-add it, while an older Retain=false event agrees with + /// the omission and is still delivered (a client that did not reconcile on RefreshEnd needs it). + /// + public bool Supersedes(ConditionEventSnapshot e) => + e.Sequence <= Watermark && (e.Retain || Refreshed.Contains(e.ConditionId)); + } + /// /// The gate-relevant slice of a Part 9 condition's state — exactly the fields that drive a @@ -5292,7 +5362,7 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3 /// unsubscribe arms, which the SDK's MasterNodeManager drives for every create, modify, /// transfer, restore and delete of an all-events item, subscription teardown included. /// - private readonly ConcurrentDictionary _allEventsItems = new(); + private readonly ConcurrentDictionary _allEventsItems = new(); // The SDK's base implementation attaches a Server-object item to the event // fan-out of every one of this manager's root notifiers, so every alarm in the server reached it @@ -5314,7 +5384,7 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3 lock (Lock) { if (unsubscribe) _allEventsItems.TryRemove(monitoredItem, out _); - else _allEventsItems[monitoredItem] = 0; + else _allEventsItems.GetOrAdd(monitoredItem, static _ => new AllEventsItemState()); ReconcileAllEventsMonitoring(); } @@ -5488,8 +5558,9 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3 var scopes = AclEnforced ? ConditionVisibilityScopes(node) : null; var reportingSession = (context as ISessionSystemContext)?.SessionId; + var deferred = e is ConditionEventSnapshot { Deferred: true } d ? d : null; - foreach (var item in _allEventsItems.Keys) + foreach (var (item, state) in _allEventsItems) { if (e is AuditEventState && !IsAuditDeliverableTo(item)) continue; if (scopes is not null && !IsConditionVisibleTo(item, scopes)) continue; @@ -5501,10 +5572,36 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3 continue; } - item.QueueEvent(e); + if (deferred is null) + { + item.QueueEvent(e); + continue; + } + + // #854: a deferred event reaches this item off Lock, so a ConditionRefresh of the item may have replayed a + // newer state since it was reported. Decided and queued under the item's gate, which the refresh holds while + // it queues its replay and installs its cut — so it lands either before the replay or not at all. + lock (state.Gate) + { + if (state.Cut?.Supersedes(deferred) == true) + { + Interlocked.Increment(ref _refreshSupersededDeliveries); + OtOpcUaTelemetry.ConditionEventDeliveriesSuperseded.Add(1); + continue; + } + + DeferredServerObjectDeliveryProbe?.Invoke(); + item.QueueEvent(e); + } } } + /// + /// Test probe (#854): runs on the drain thread between a deferred event's supersession check and its queueing to a + /// Server-object item, inside the item's gate — the window a refresh must not slip into. Null in production. + /// + internal Action? DeferredServerObjectDeliveryProbe { get; set; } + /// /// The SDK's audit-event guard from MonitoredNode2.OnReportEvent, reproduced exactly: an audit /// event reaches an item only while the server is auditing, and only over a SignAndEncrypt or @@ -6068,19 +6165,15 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3 } // A refresh on a Server-object item is the other way it learns about conditions — it is how a - // client connecting after a transition gets the current alarm picture — and the SDK base replays - // every condition under every root notifier to it after only its own role-permission check. With - // enforcement on, those items are refreshed here instead, through the same - // IsConditionVisibleTo decision live delivery uses; an event that names no - // condition is withheld. With it off, the base runs unchanged (its per-notifier repeats of one - // condition share an EventId, which the item's queue already collapses to one copy). - // Items on one of this manager's own notifiers keep the base behaviour: a stood-down item is off - // its notifier's fan-out, which the base already treats as "not refreshed". + // client connecting after a transition gets the current alarm picture. Those items are always refreshed + // here (RefreshServerObjectItems): with enforcement on through the same IsConditionVisibleTo decision live + // delivery uses, and in both modes so the refresh can cut off deferred events it supersedes (#854). + // Items on one of this manager's own notifiers keep the base behaviour with enforcement off (a stood-down + // item is off its notifier's fan-out, which the base already treats as "not refreshed"), and + // RefreshDirectItems with it on. /// public override ServiceResult ConditionRefresh(OperationContext context, IList monitoredItems) { - if (!AclEnforced) return base.ConditionRefresh(context, monitoredItems); - var direct = new List(); var serverItems = new List(); foreach (var item in monitoredItems) @@ -6089,36 +6182,116 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3 (item.NodeId == ObjectIds.Server ? serverItems : direct).Add(item); } + var enforced = AclEnforced; var systemContext = SystemContext.Copy(context); - if (direct.Count > 0) RefreshDirectItems(systemContext, direct); - if (serverItems.Count == 0) return ServiceResult.Good; - - List notifiers; - lock (Lock) notifiers = RootNotifiers?.ToList() ?? []; - - var events = new List(); - foreach (var notifier in notifiers) notifier.ConditionRefresh(systemContext, events, true); - - // One snapshot per condition, with its scopes resolved once. - var conditions = new List<(IFilterTarget Event, List Scopes)>(); - var seen = new HashSet(System.Collections.Generic.ReferenceEqualityComparer.Instance); - foreach (var e in events) + if (direct.Count > 0) { - if (ConditionOf(e) is not { } condition || !seen.Add(condition)) continue; - conditions.Add((e, ConditionVisibilityScopes(condition))); + if (enforced) RefreshDirectItems(systemContext, direct); + else base.ConditionRefresh(context, direct); } - foreach (var item in serverItems) + if (serverItems.Count > 0) RefreshServerObjectItems(systemContext, serverItems, enforced); + return ServiceResult.Good; + } + + /// + /// ConditionRefresh (and ConditionRefresh2) for Server-object (i=2253) items: each condition under a root + /// notifier once, through the item's ACL decision when (#586) — and never overtaken by + /// a deferred event older than what it replays (Gitea #854). + /// + /// The refresh's system context. + /// The Server-object items being refreshed. + /// Whether the data-plane ACL is enforced. + /// + /// + /// The overtaking (#854). While a folder subscriber's channel is full, a condition's later events wait + /// in its deferral queue (#838) and reach Server-object items only when the drain gets to them, off + /// Lock. A refresh in that window replays the condition's CURRENT state between RefreshStart and + /// RefreshEnd; the waiting events, all older, then followed RefreshEnd — a stale state, or a condition the + /// refresh had just dropped re-added, until the newest drained. + /// + /// + /// The cut. Everything — the watermark (), whether any event is + /// deferred, the replay itself and its queueing — happens in ONE hold of Lock, under which every condition + /// event is reported. So the replay is exactly the state at the watermark: an event at or below it is not newer + /// than the replay, one above it was reported after it. Each item's replay is queued and its + /// installed under the item's , which a deferred + /// delivery to the item also holds (): a deferred event therefore lands + /// before the replay, or is judged against the cut — withheld if superseded (). + /// Only events marked are judged: any other was delivered to + /// Server-object items synchronously under Lock, so it cannot follow a replay it predates. + /// + /// + /// Per item, never global. Only the refreshed item stops receiving the superseded events; the deferral + /// queue, every other Server-object item and every folder item still get them, in order. The refresh never + /// waits for a drain. No cut is kept while nothing is deferred (the common case). + /// + /// + /// Lock order is Lock → item gate → the SDK item's own lock, the same as a synchronous report + /// (Lock → ); the drain takes the item gate without Lock, and + /// nothing under an item gate takes Lock or the event-report gate. + /// + /// + /// With enforcement off this replaces the SDK base's Server-object loop with the same result: the base queued + /// one copy per root notifier a condition is wired to, which the item's queue collapsed to one (they share an + /// EventId); here it is one per condition. A replayed event that names no condition is withheld only under + /// enforcement (fail closed), as before. + /// + /// + private void RefreshServerObjectItems(ServerSystemContext systemContext, List items, bool enforced) + { + lock (Lock) { - foreach (var (e, scopes) in conditions) + var watermark = Interlocked.Read(ref _conditionEventSequence); + bool deferring; + lock (_eventReportGate) deferring = _deferredEventReports > 0; + + var events = new List(); + foreach (var notifier in RootNotifiers?.ToList() ?? []) notifier.ConditionRefresh(systemContext, events, true); + + // One event per condition, with its scopes resolved once. + var replay = new List<(IFilterTarget Event, NodeState? Condition, List? Scopes)>(); + var seen = new HashSet(System.Collections.Generic.ReferenceEqualityComparer.Instance); + foreach (var e in events) { - if (!IsConditionVisibleTo(item, scopes)) continue; - if (ServiceResult.IsBad(ValidateEventRolePermissions(item, e))) continue; - item.QueueEvent(e); + if (ConditionOf(e) is not { } condition) + { + if (!enforced) replay.Add((e, null, null)); + continue; + } + + if (!seen.Add(condition)) continue; + replay.Add((e, condition, enforced ? ConditionVisibilityScopes(condition) : null)); + } + + foreach (var item in items) + { + var queued = new List(); + var refreshed = new HashSet(); + foreach (var (e, condition, scopes) in replay) + { + if (scopes is not null && !IsConditionVisibleTo(item, scopes)) continue; + if (ServiceResult.IsBad(ValidateEventRolePermissions(item, e))) continue; + queued.Add(e); + if (condition is not null) refreshed.Add(condition.NodeId); + } + + if (!_allEventsItems.TryGetValue(item, out var state)) + { + // Not (or no longer) subscribed: no deferred delivery can reach it, so there is nothing to cut. + foreach (var e in queued) item.QueueEvent(e); + continue; + } + + lock (state.Gate) + { + foreach (var e in queued) item.QueueEvent(e); + // A cut is only needed while an event at or below the watermark may still be delivered. Refreshes are + // serialised by Lock, so a later one always carries the later watermark and simply replaces it. + state.Cut = deferring ? new RefreshCut(watermark, refreshed) : null; + } } } - - return ServiceResult.Good; } /// diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/ConditionRefreshDeferralWireTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/ConditionRefreshDeferralWireTests.cs new file mode 100644 index 000000000..b1d456e98 --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/ConditionRefreshDeferralWireTests.cs @@ -0,0 +1,430 @@ +using System.Collections; +using System.Diagnostics; +using System.Reflection; +using Opc.Ua; +using Opc.Ua.Client; +using Opc.Ua.Server; +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Commons.OpcUa; +using ZB.MOM.WW.OtOpcUa.Configuration.Entities; +using ZB.MOM.WW.OtOpcUa.Configuration.Enums; +using ZB.MOM.WW.OtOpcUa.Core.Authorization; +using MonitoredItem = Opc.Ua.Client.MonitoredItem; + +namespace ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests; + +/// +/// Gitea #854 (from the #838 review): while a folder subscriber's event channel is full, a condition's later events +/// are deferred in its per-condition queue. A ConditionRefresh on a Server-object (i=2253) item in that +/// window delivers the condition's CURRENT state between RefreshStart and RefreshEnd — and the deferred events, all +/// older than that state, used to reach the same item AFTER RefreshEnd, so a client briefly showed (or, for a +/// condition the refresh had dropped, re-added) a stale alarm. They are now withheld from the refreshed item; +/// every other subscriber still receives them, and the refresh never waits for the stalled drain. +/// +/// +/// The overload is produced deterministically: the folder item's server-side MonitoredItem lock is held, so +/// the SDK's per-node delivery task parks in QueueEvent, its 4096-slot channel fills, and the condition's +/// later events are deferred — without holding the node manager's Lock, which the refresh needs. +/// +public sealed class ConditionRefreshDeferralWireTests +{ + private const string EqA = ServerEventRig.EqA; + + // > 4096 (channel) + 1 (parked in the delivery task) + 1 (the in-flight deferred write), < 1024 pending: nothing coalesces. + private const int FloodSize = 5000; + + /// Per-folder / Server-object ACL on or off (off is the production default). + /// + /// Whether the condition's state at refresh time is still Retain (so the refresh re-sends it) or has cleared + /// (so the refresh omits it and its final Retain=false event must still arrive). + /// + /// ConditionRefresh2 on the single item instead of ConditionRefresh on the subscription. + [Theory] + [InlineData(false, true, false)] + [InlineData(true, true, false)] + [InlineData(false, false, false)] + [InlineData(true, false, false)] + [InlineData(false, true, true)] + [InlineData(true, false, true)] + public async Task Deferred_events_older_than_a_refresh_never_reach_the_refreshed_server_object_item_after_RefreshEnd( + bool enforced, bool finalRetained, bool refresh2) + { + var ct = TestContext.Current.CancellationToken; + await using var rig = await ServerEventRig.StartAsync("Refresh854", encrypted: false, ct); + rig.KeyRoles = []; + rig.MaterialiseNative(); + rig.Enforce(enforced, [ServerEventRig.Key()], KeyEquipmentGrant(EqA)); + + // The folder subscriber whose channel the stall fills, on its own session so its parked item never touches the + // refreshing client's subscription. + using var folderSession = await rig.ConnectAsync(anonymous: false, ct); + var folderEvents = new RecordedEvents(); + var folderSubscription = await RecordedEvents.SubscribeAsync(folderSession, folderEvents, ct); + var onA = RecordedEvents.AddEventItem(folderSubscription, ServerEventRig.UnsNodeId(folderSession, EqA)); + await folderSubscription.ApplyChangesAsync(ct); + onA.Status.Error.ShouldBeNull(); + + using var serverSession = await rig.ConnectAsync(anonymous: false, ct); + var serverEvents = new RecordedEvents(); + var serverSubscription = await RecordedEvents.SubscribeAsync(serverSession, serverEvents, ct); + var onServer = RecordedEvents.AddEventItem(serverSubscription, ObjectIds.Server); + await serverSubscription.ApplyChangesAsync(ct); + onServer.Status.Error.ShouldBeNull(); + + // Warm both paths (the folder item's verdict is cached on its first event). + rig.Project(native: true, Active("seed")); + await RecordedEvents.WaitUntilAsync(() => folderEvents.Has(onA, "seed") && serverEvents.Has(onServer, "seed")); + await folderSubscription.SetPublishingModeAsync(false, ct); + + using (var stall = ItemLockStall.Hold(FindServerItem(rig.Nm, onA.Status.Id))) + { + for (var i = 0; i < FloodSize; i++) rig.Project(native: true, Active($"f{i}")); + rig.Nm.DeferredEventReports.ShouldBeGreaterThan(0, "the flood must actually have deferred events"); + rig.Project(native: true, finalRetained ? Active("final") : Cleared("final")); + + // The refresh is a service call: it must complete while the drain is still stalled. + var refresh = Stopwatch.StartNew(); + if (refresh2) + { + await serverSession.CallAsync( + ObjectTypeIds.ConditionType, MethodIds.ConditionType_ConditionRefresh2, ct, + serverSubscription.Id, onServer.Status.Id); + } + else + { + await serverSession.CallAsync( + ObjectTypeIds.ConditionType, MethodIds.ConditionType_ConditionRefresh, ct, serverSubscription.Id); + } + + refresh.Elapsed.ShouldBeLessThan(TimeSpan.FromSeconds(10), "ConditionRefresh must not wait for the stalled drain"); + await RecordedEvents.WaitUntilAsync(() => serverEvents.HasType(onServer, ObjectTypeIds.RefreshEndEventType)); + rig.Nm.DeferredEventReports.ShouldBeGreaterThan(0, "the deferred events were still waiting at RefreshEnd"); + stall.Release(); + } + + await RecordedEvents.WaitUntilAsync(() => rig.Nm.DeferredEventReports == 0); + // Positive control: a later transition is delivered (and everything deferred before it has been decided). + rig.Project(native: true, Active("after")); + await RecordedEvents.WaitUntilAsync(() => serverEvents.Has(onServer, "after")); + await Task.Delay(TimeSpan.FromMilliseconds(300), ct); + + var stream = serverEvents.Of(onServer); + var start = stream.ToList().FindIndex(f => f[RecordedEvents.EventTypeField].GetNodeId() == ObjectTypeIds.RefreshStartEventType); + var end = stream.ToList().FindIndex(f => f[RecordedEvents.EventTypeField].GetNodeId() == ObjectTypeIds.RefreshEndEventType); + start.ShouldBeGreaterThanOrEqualTo(0); + end.ShouldBeGreaterThan(start); + var replayed = stream.Skip(start + 1).Take(end - start - 1).Select(RecordedEvents.Message).ToList(); + var afterEnd = stream.Skip(end + 1) + .Where(f => f[RecordedEvents.AckedField].TryGetValue(out bool _)) + .Select(RecordedEvents.Message) + .ToList(); + + if (finalRetained) + { + replayed.ShouldBe(["final"], "the refresh re-sends the condition's current state"); + afterEnd.ShouldBe(["after"], "nothing older than the refreshed state may follow RefreshEnd"); + } + else + { + replayed.ShouldBeEmpty("a cleared (Retain=false) condition is not part of a refresh"); + // The stale Retain=true events would re-add an alarm the refresh just dropped; the final Retain=false one + // agrees with the refresh and must still arrive (a client that did not reconcile on RefreshEnd needs it). + afterEnd.ShouldBe(["final", "after"]); + stream.Last(f => RecordedEvents.Message(f) == "final")[RecordedEvents.RetainField].GetBoolean().ShouldBeFalse(); + } + + rig.Nm.RefreshSupersededDeliveries.ShouldBeGreaterThan(0); + + // The folder subscriber did not refresh: it still receives the deferred events, in order, ending with the newest. + await folderSubscription.SetPublishingModeAsync(true, ct); + await RecordedEvents.WaitUntilAsync(() => folderEvents.Has(onA, "after")); + var folderFlood = folderEvents.Of(onA) + .Select(RecordedEvents.Message) + .Where(m => m?.StartsWith('f') == true && m != "final") + .Select(m => int.Parse(m![1..], System.Globalization.CultureInfo.InvariantCulture)) + .ToList(); + folderFlood.ShouldBe(folderFlood.Order().ToList()); + folderFlood[^1].ShouldBe(FloodSize - 1, "the non-refreshing subscriber keeps the deferred tail"); + folderEvents.Has(onA, "final").ShouldBeTrue(); + } + + /// + /// The race the item gate closes, made deterministic: a deferred event is between its supersession check and its + /// queueing to the Server-object item when the refresh arrives. The refresh must wait for that one delivery (so + /// the event lands before the replay), not slip in between and let it follow RefreshEnd. + /// + [Fact] + public async Task A_refresh_waits_for_a_deferred_delivery_already_past_its_check_so_it_lands_before_the_replay() + { + var ct = TestContext.Current.CancellationToken; + await using var rig = await ServerEventRig.StartAsync("Refresh854Gate", encrypted: false, ct); + rig.KeyRoles = []; + rig.MaterialiseNative(); + rig.Enforce(false, [ServerEventRig.Key()], KeyEquipmentGrant(EqA)); + + using var folderSession = await rig.ConnectAsync(anonymous: false, ct); + var folderEvents = new RecordedEvents(); + var folderSubscription = await RecordedEvents.SubscribeAsync(folderSession, folderEvents, ct); + var onA = RecordedEvents.AddEventItem(folderSubscription, ServerEventRig.UnsNodeId(folderSession, EqA)); + await folderSubscription.ApplyChangesAsync(ct); + + using var serverSession = await rig.ConnectAsync(anonymous: false, ct); + var serverEvents = new RecordedEvents(); + var serverSubscription = await RecordedEvents.SubscribeAsync(serverSession, serverEvents, ct); + var onServer = RecordedEvents.AddEventItem(serverSubscription, ObjectIds.Server); + await serverSubscription.ApplyChangesAsync(ct); + + rig.Project(native: true, Active("seed")); + await RecordedEvents.WaitUntilAsync(() => folderEvents.Has(onA, "seed") && serverEvents.Has(onServer, "seed")); + await folderSubscription.SetPublishingModeAsync(false, ct); + + using var entered = new ManualResetEventSlim(); + using var proceed = new ManualResetEventSlim(); + var probed = 0; + rig.Nm.DeferredServerObjectDeliveryProbe = () => + { + if (Interlocked.Increment(ref probed) != 1) return; + entered.Set(); + proceed.Wait(TimeSpan.FromSeconds(30)); + }; + + using (ItemLockStall.Hold(FindServerItem(rig.Nm, onA.Status.Id))) + { + for (var i = 0; i < FloodSize; i++) rig.Project(native: true, Active($"f{i}")); + rig.Project(native: true, Active("final")); + } + + // The drain resumes and parks on its first deferred delivery to the Server-object item, past the check. + entered.Wait(TimeSpan.FromSeconds(10), ct).ShouldBeTrue("the drain must reach the Server-object item"); + var refresh = serverSession.CallAsync( + ObjectTypeIds.ConditionType, MethodIds.ConditionType_ConditionRefresh, ct, serverSubscription.Id); + await Task.Delay(TimeSpan.FromSeconds(1), ct); + // (The SDK may answer the Call before the refresh body runs, so the replay is observed on the event stream.) + serverEvents.HasType(onServer, ObjectTypeIds.RefreshEndEventType) + .ShouldBeFalse("the refresh must wait for the delivery already inside the item's gate"); + proceed.Set(); + await refresh.WaitAsync(TimeSpan.FromSeconds(10), ct); + await RecordedEvents.WaitUntilAsync(() => serverEvents.HasType(onServer, ObjectTypeIds.RefreshEndEventType)); + rig.Nm.DeferredServerObjectDeliveryProbe = null; + + await RecordedEvents.WaitUntilAsync(() => rig.Nm.DeferredEventReports == 0); + rig.Project(native: true, Active("after")); + await RecordedEvents.WaitUntilAsync(() => serverEvents.Has(onServer, "after")); + await Task.Delay(TimeSpan.FromMilliseconds(300), ct); + + var stream = serverEvents.Of(onServer).ToList(); + var start = stream.FindIndex(f => f[RecordedEvents.EventTypeField].GetNodeId() == ObjectTypeIds.RefreshStartEventType); + var end = stream.FindIndex(f => f[RecordedEvents.EventTypeField].GetNodeId() == ObjectTypeIds.RefreshEndEventType); + end.ShouldBeGreaterThan(start); + var window = stream.Skip(start + 1).Take(end - start - 1).Select(RecordedEvents.Message).ToList(); + window.Count.ShouldBe(2, "the parked delivery, then the replay"); + window[0]!.ShouldStartWith("f"); + window[1].ShouldBe("final"); + stream.Skip(end + 1) + .Where(f => f[RecordedEvents.AckedField].TryGetValue(out bool _)) + .Select(RecordedEvents.Message) + .ShouldBe(["after"]); + } + + /// + /// Race / stress: reports, repeated refreshes and an intermittently stalled drain all run at once. For the + /// refreshing Server-object item, no condition event delivered after a RefreshEnd may be older than the state that + /// refresh replayed, and the stream still ends on the newest state. Nothing deadlocks. + /// + /// Per-folder / Server-object ACL on or off. + [Theory] + [InlineData(false)] + [InlineData(true)] + public async Task Concurrent_reports_refreshes_and_drains_never_deliver_a_pre_refresh_state_after_RefreshEnd(bool enforced) + { + var ct = TestContext.Current.CancellationToken; + await using var rig = await ServerEventRig.StartAsync("Refresh854Stress", encrypted: false, ct); + rig.KeyRoles = []; + rig.MaterialiseNative(); + rig.Enforce(enforced, [ServerEventRig.Key()], KeyEquipmentGrant(EqA)); + + using var folderSession = await rig.ConnectAsync(anonymous: false, ct); + var folderEvents = new RecordedEvents(); + var folderSubscription = await RecordedEvents.SubscribeAsync(folderSession, folderEvents, ct); + var onA = RecordedEvents.AddEventItem(folderSubscription, ServerEventRig.UnsNodeId(folderSession, EqA)); + await folderSubscription.ApplyChangesAsync(ct); + + using var serverSession = await rig.ConnectAsync(anonymous: false, ct); + var serverEvents = new RecordedEvents(); + var serverSubscription = await RecordedEvents.SubscribeAsync(serverSession, serverEvents, ct); + var onServer = RecordedEvents.AddEventItem(serverSubscription, ObjectIds.Server); + onServer.QueueSize = 100_000; // keep every event so the ordering check sees the whole stream + await serverSubscription.ApplyChangesAsync(ct); + + rig.Project(native: true, Active("s0")); + await RecordedEvents.WaitUntilAsync(() => folderEvents.Has(onA, "s0") && serverEvents.Has(onServer, "s0")); + await folderSubscription.SetPublishingModeAsync(false, ct); + var item = FindServerItem(rig.Nm, onA.Status.Id); + + using var stop = CancellationTokenSource.CreateLinkedTokenSource(ct); + var counter = 0; + var reporter = Task.Factory.StartNew(() => + { + while (!stop.IsCancellationRequested) + { + // Bursts under one Lock hold, like an apply pass, interleaved with single reports. + lock (rig.Nm.Lock) + { + for (var i = 0; i < 500; i++) rig.Project(native: true, Active($"s{Interlocked.Increment(ref counter)}")); + } + + rig.Project(native: true, Active($"s{Interlocked.Increment(ref counter)}")); + Thread.Sleep(2); + } + }, stop.Token, TaskCreationOptions.LongRunning, TaskScheduler.Default); + + var deferredSeen = 0; + var staller = Task.Factory.StartNew(() => + { + while (!stop.IsCancellationRequested) + { + using (ItemLockStall.Hold(item)) + { + Thread.Sleep(400); + if (rig.Nm.DeferredEventReports > 0) Interlocked.Increment(ref deferredSeen); + } + + Thread.Sleep(100); + } + }, stop.Token, TaskCreationOptions.LongRunning, TaskScheduler.Default); + + for (var r = 0; r < 40; r++) + { + var call = r % 3 == 2 + ? serverSession.CallAsync(ObjectTypeIds.ConditionType, MethodIds.ConditionType_ConditionRefresh2, ct, + serverSubscription.Id, onServer.Status.Id) + : serverSession.CallAsync(ObjectTypeIds.ConditionType, MethodIds.ConditionType_ConditionRefresh, ct, + serverSubscription.Id); + await call.WaitAsync(TimeSpan.FromSeconds(20), ct); + await Task.Delay(TimeSpan.FromMilliseconds(80), ct); + } + + await stop.CancelAsync(); + await reporter.WaitAsync(TimeSpan.FromSeconds(30), CancellationToken.None); + await staller.WaitAsync(TimeSpan.FromSeconds(30), CancellationToken.None); + Volatile.Read(ref deferredSeen).ShouldBeGreaterThan(1, "the stress must actually have run with deferred events"); + + await RecordedEvents.WaitUntilAsync(() => rig.Nm.DeferredEventReports == 0); + var last = $"s{Volatile.Read(ref counter)}"; + await folderSubscription.SetPublishingModeAsync(true, ct); + await RecordedEvents.WaitUntilAsync(() => serverEvents.Has(onServer, last)); + await Task.Delay(TimeSpan.FromMilliseconds(500), ct); + + // Walk the Server-object stream: after each RefreshEnd, a condition event must not be older than what that + // refresh replayed (the refresh replays the then-current state, so the floor is its replayed counter). + var floor = -1; + var replaying = false; + var replayedInThisRefresh = -1; + var violations = new List(); + foreach (var f in serverEvents.Of(onServer)) + { + var type = f[RecordedEvents.EventTypeField].GetNodeId(); + if (type == ObjectTypeIds.RefreshStartEventType) + { + replaying = true; + replayedInThisRefresh = -1; + continue; + } + + if (type == ObjectTypeIds.RefreshEndEventType) + { + replaying = false; + floor = Math.Max(floor, replayedInThisRefresh); + continue; + } + + if (!f[RecordedEvents.AckedField].TryGetValue(out bool _)) continue; + var n = int.Parse(RecordedEvents.Message(f)![1..], System.Globalization.CultureInfo.InvariantCulture); + if (replaying) + { + replayedInThisRefresh = n; + continue; + } + + if (n < floor) violations.Add($"s{n} after a refresh that replayed s{floor}"); + } + + violations.ShouldBeEmpty(); + rig.Nm.RefreshSupersededDeliveries.ShouldBeGreaterThan(0, "some refresh must actually have overtaken deferred events"); + RecordedEvents.Message(serverEvents.Of(onServer).Last(f => f[RecordedEvents.AckedField].TryGetValue(out bool _))) + .ShouldBe(last, "the stream still converges on the newest state"); + await RecordedEvents.WaitUntilAsync(() => folderEvents.Has(onA, last)); + } + + /// The server-side event item behind a client item, from the node manager's direct-item map. + private static IEventMonitoredItem FindServerItem(OtOpcUaNodeManager nm, uint id) + { + var map = (IDictionary)typeof(OtOpcUaNodeManager) + .GetField("_eventItems", BindingFlags.Instance | BindingFlags.NonPublic)! + .GetValue(nm)!; + foreach (var key in map.Keys) + { + if (key is IEventMonitoredItem item && item.Id == id) return item; + } + + throw new InvalidOperationException($"no server-side event item {id}"); + } + + private static AlarmConditionSnapshot Active(string message) => + new(true, false, false, true, AlarmShelvingKind.Unshelved, 700, message); + + private static AlarmConditionSnapshot Cleared(string message) => + new(false, true, true, true, AlarmShelvingKind.Unshelved, 700, message); + + private static NodeAcl KeyEquipmentGrant(string equipment) => new() + { + NodeAclId = $"acl-key-{equipment}", + ClusterId = ServerEventRig.Cluster, + Principal = ServerEventRig.KeyPrincipal, + ScopeKind = NodeAclScopeKind.Equipment, + ScopeId = $"eq-{equipment}", + PermissionFlags = NodePermissions.ReadOnly, + }; + + /// + /// Holds a server-side monitored item's private lock on a dedicated thread, so the SDK's per-node delivery task + /// parks in QueueEvent for that item (a stalled subscriber) without the node manager's Lock held. + /// + private sealed class ItemLockStall : IDisposable + { + private readonly ManualResetEventSlim _release = new(); + private readonly Thread _thread; + + private ItemLockStall(IEventMonitoredItem item) + { + var gate = (Lock)typeof(Opc.Ua.Server.MonitoredItem) + .GetField("m_lock", BindingFlags.Instance | BindingFlags.NonPublic)! + .GetValue(item)!; + using var held = new ManualResetEventSlim(); + _thread = new Thread(() => + { + gate.Enter(); + try + { + held.Set(); + _release.Wait(); + } + finally { gate.Exit(); } + }) { IsBackground = true }; + _thread.Start(); + held.Wait(); + } + + public static ItemLockStall Hold(IEventMonitoredItem item) => new(item); + + public void Release() => _release.Set(); + + public void Dispose() + { + _release.Set(); + _thread.Join(); + _release.Dispose(); + } + } +} From fe48458a414ffd58907a8ffd1f70aef1598e53e9 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Wed, 11 Nov 2026 07:28:08 -0500 Subject: [PATCH 2/3] fix(opcua): Server-object ConditionRefresh off Lock unless deferring; cut path O(conditions) with replay queued off Lock (#854 review) Review of 1622d07f: 1. Perf: every i=2253 refresh ran the whole notifier walk + permission checks + QueueEvent under Lock. Now, with nothing deferred (the common case), the refresh is the SDK-shaped walk off Lock (Lock only to copy the root notifiers), kept only if no deferral episode existed at any moment of it (count 0 before and after, episode unchanged); otherwise the cut path also runs and its later replay supersedes. The cut path takes ACL/role verdicts off Lock, then one short Lock hold: watermark + frozen snapshots of Retain conditions from _alarmConditions (no tree walk) + per-item cut and "replay pending" marker; the replay is queued OFF Lock, one event per hold of the item gate. While pending, a condition event newer than the cut marks its condition so the older frozen entry is dropped rather than following it. A concurrent fast-path refresh cancels a pending cut replay of the same item. Measured, 3000 retained conditions, longest Lock wait seen by another thread during one refresh: 1622d07f 95-106 ms idle / 102-127 ms deferring; now 0-3 ms idle / 23-32 ms deferring. 2. ACL: a retained condition omitted from the replay for want of a grant is never withheld (RefreshCut.RetainedAtCut); only Retain=true events of a condition omitted because it was NOT retained are. 3. One gate/cut state per item for its lifetime (ConditionalWeakTable keyed by the item): an unsubscribe + resubscribe between refresh and drain keeps the cut. 4. Cuts carry their deferral episode; ClearStaleRefreshCuts runs when the count returns to zero, and a stale-episode cut is inert. 5. Dispose guard for the custom refresh (ObjectDisposedException once disposing; Dispose waits for refreshes in flight). 6. Docs: a refresh waits at most for one in-progress QueueEvent on the item. Tests: resubscribe variants, ACL-grant-lands test, pending-replay probe test, dispose test, 3000-condition Lock-hold measurement with a probe pinning that neither path queues its replay under Lock; stress invariant tightened (never backwards outside a refresh; each refresh ends on the newest delivered state) and tolerant of the SDK's own BadRefreshInProgress overlap refusal. Each new test fails against its targeted mutation. --- .claude/rules/data-plane-auth.md | 9 +- docs/security.md | 45 +- .../OtOpcUaNodeManager.cs | 432 ++++++++++++++---- .../ConditionRefreshDeferralWireTests.cs | 324 ++++++++++++- 4 files changed, 688 insertions(+), 122 deletions(-) diff --git a/.claude/rules/data-plane-auth.md b/.claude/rules/data-plane-auth.md index 62a185a44..3268b7f18 100644 --- a/.claude/rules/data-plane-auth.md +++ b/.claude/rules/data-plane-auth.md @@ -212,9 +212,12 @@ Two things to know before relying on any of this: `OnConditionReportEvent`) — a deferred event of a removed/replaced condition is dropped, never a ghost on `i=2253`; per-CONDITION ordered queues on overflow; coalesce past 1024/condition or 65 536 total (oldest not-started gives way to newest); Warning/Information + `otopcua.opcua.condition_events.{deferred,discarded}`. A Server-object ConditionRefresh - (and ConditionRefresh2) takes a sequence CUT in one `Lock` hold and withholds older deferred events from THAT item - only (re-sent condition, or a Retain=true event of an omitted one) under a per-item gate held across check+QueueEvent - (#854, `ConditionRefreshDeferralWireTests`); folder items keep the residual (SDK channel + memoised verdict). Folder events follow Part 3's shape — + (and ConditionRefresh2, #854) with nothing deferred is the plain walk OFF `Lock`; while deferring it takes a sequence + CUT (one short `Lock` hold: watermark + frozen snapshots of Retain conditions from `_alarmConditions`), queues the + replay off `Lock` one event per hold of a per-item gate, and withholds older deferred events from THAT item only + (re-sent condition, or Retain=true of one omitted because NOT retained — never one omitted by ACL). Item state is a + `ConditionalWeakTable` keyed by the item; cuts clear when the deferral count hits 0. Tests: + `ConditionRefreshDeferralWireTests`; folder items keep the residual (SDK channel + memoised verdict). Folder events follow Part 3's shape — `ReceiveEvents` on EventType + SourceNode (SDK check, kept first) AND our decision; non-condition events on a folder item fail closed under enforcement. Tests: `FolderEventFloodWireTests`. - **ns=0 standard nodes stay ungated by design, even with `Security:Acl:Enabled`** (#617) — `ServerStatus`, diff --git a/docs/security.md b/docs/security.md index 5c6d21880..3e15eeb37 100644 --- a/docs/security.md +++ b/docs/security.md @@ -885,16 +885,33 @@ Behaviours worth knowing before enabling it: - **`ConditionRefresh` is not overtaken by deferred events on Server-object items (Gitea #854).** A refresh of an `i=2253` item replays each condition's current state between RefreshStart and RefreshEnd; the condition's deferred events — all older — used to follow RefreshEnd, briefly showing a stale state or re-adding an alarm the refresh had - just dropped (ScadaBridge's snapshot swap drops absent conditions, then a stale active event re-added it). Now every - condition event carries its report sequence; the refresh takes its **cut** (the sequence, whether anything is - deferred, the replay, and its queueing) in **one hold of `Lock`**, so the replay is exactly the state at the cut, and - installs it per item under that item's gate, which a deferred delivery also holds across its check and - `QueueEvent`. A deferred event at or below the cut is withheld **from that item only** when the refresh re-sent its - condition, or when it is `Retain=true` for a condition the refresh omitted; an older `Retain=false` event of an - omitted condition still arrives (it agrees with the omission, and a client that does not reconcile on RefreshEnd - needs it). The deferral queues, every other Server-object item and every folder item are untouched; the refresh never - waits for a drain. Meter: `otopcua.opcua.condition_events.superseded` (per item and event). Both ConditionRefresh and - ConditionRefresh2, flag on and off. + just dropped (ScadaBridge's snapshot swap drops absent conditions, then a stale active event re-added it). Every + condition event now carries its report sequence. Two paths, chosen per refresh: + - **Nothing deferred (the common case): unchanged and off `Lock`.** The SDK-shaped notifier walk, queued as live + condition nodes; `Lock` is taken only to copy the root-notifier list. It stands only if no deferral episode existed + at any moment of it (deferral count zero before and after, episode number unchanged); otherwise the cut path runs + too, and its replay — later in the same RefreshStart..RefreshEnd — supersedes anything that slipped in between. + - **Deferring: the cut.** ACL/role verdicts first, off `Lock`. Then ONE short `Lock` hold takes the sequence + watermark and a frozen snapshot of each Retain condition straight from the condition map (O(conditions), no tree + walk), and installs per item, under that item's gate, the **cut** plus a "replay pending" marker. The replay is + queued **off `Lock`**, one event per hold of the item gate. While the marker is set, every condition event to the + item takes the gate: one newer than the cut marks its condition so the older frozen entry is dropped instead of + following it; a deferred one at or below the cut is withheld **from that item only** when the refresh re-sent its + condition, or when it is `Retain=true` for a condition the refresh omitted **because it was not retained**. An + older `Retain=false` event of such a condition still arrives (it agrees with the omission), and a condition that was + retained but omitted for the item's **grants** is never withheld (a grant added before the drain lets its events + through; the ACL decides them at delivery as ever). + - The deferral queues, every other Server-object item and every folder item are untouched. A refresh waits for the + drain at most for one in-progress `QueueEvent` on that item (the gate), never for the drain itself; a reporter + holding `Lock` likewise waits at most for one. Each item keeps one gate/cut state for its lifetime (keyed by the + item, so an unsubscribe + resubscribe keeps it), and cuts are cleared when the deferral count returns to zero. + The custom refresh is refused (`ObjectDisposedException`) once the node manager is disposing, and disposal waits + for refreshes in flight, as the base's private operation guard does. Meter: + `otopcua.opcua.condition_events.superseded` (per item and event). ConditionRefresh and ConditionRefresh2, flag + on and off. + - **Measured** (3000 retained conditions, longest `Lock` wait another thread saw during one refresh): the first + #854 cut (`1622d07f`, everything under `Lock`) 95–106 ms idle / 102–127 ms deferring; now 0–3 ms idle / + 23–32 ms deferring (the frozen snapshots). - ⚠️ **Residual: folder (notifier) items.** A refresh of an item on an equipment/device folder can still be followed by older events of a condition. Two sources, neither reachable per item: the SDK's own `MonitoredNode2` channel (up to 4096 events already accepted before the refresh — this exists in the stock SDK too, and in normal operation @@ -912,8 +929,12 @@ Behaviours worth knowing before enabling it: `ConditionRefreshDeferralWireTests` (#854) stalls a folder item's SDK lock so its channel fills without `Lock` held, refreshes the Server-object item (ConditionRefresh and ConditionRefresh2, flag on/off, condition retained or cleared) and pins that nothing older than the replay follows RefreshEnd while the non-refreshing folder item still receives the - whole tail; a probe-driven test pins that a refresh waits for a delivery already past its check (fails without the - item gate); and a stress test runs reports, refreshes and an intermittent stall concurrently. + whole tail (also across an unsubscribe + resubscribe of the item, and the cut is cleared after the drain); a + probe-driven test pins that a refresh waits for a delivery already past its check (fails without the item gate); an + ACL-on test pins that a condition omitted for want of a grant is not withheld once the grant lands; a 3000-condition + test measures the `Lock` hold and pins that neither path queues its replay under `Lock`; and a stress test runs + reports, refreshes and an intermittent stall concurrently (its stream never goes backwards outside a refresh, and + every refresh ends on the newest state delivered). **Model-change and audit events on the Server object are gated too** (Gitea #588). The node manager's own server-wide events — `GeneralModelChangeEvent` (NodeAdded / NodeDeleted / DataTypeChanged) and the diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs index 4458bb50e..95ceb77d4 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs @@ -1,5 +1,6 @@ using System.Collections.Concurrent; using System.Globalization; +using System.Runtime.CompilerServices; using Microsoft.Extensions.Logging; using Microsoft.Extensions.Logging.Abstractions; using Opc.Ua; @@ -620,6 +621,10 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3 { if (disposing) { + // #854 review: refuse new Server-object refreshes and let the ones in flight finish (the base's private + // NodeManagerOperation guard does this for its own paths). + _disposing = true; + SpinWait.SpinUntil(() => Volatile.Read(ref _refreshesInFlight) == 0, TimeSpan.FromSeconds(10)); Volatile.Read(ref _dataPlaneAuthority).Changed -= OnAuthorityChanged; _liveItems.Clear(); _eventItems.Clear(); @@ -1554,19 +1559,30 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3 _logger.LogError(ex, "OtOpcUaNodeManager: failed to report Part 9 condition event for {NodeId}", condition.NodeId); } - IFilterTarget next; + IFilterTarget? next = null; + var episodeEnded = false; lock (_eventReportGate) { CountDeferred(-1, condition); if (queue.Pending.Count == 0) { _deferredReports.Remove(condition); - return; + episodeEnded = _deferredEventReports == 0; + } + else + { + next = queue.Pending.Dequeue(); } - - next = queue.Pending.Dequeue(); } + if (episodeEnded) + { + ClearStaleRefreshCuts(); + return; + } + + if (next is null) return; + try { current = DeliverConditionEventAsync(condition, condition, next).AsTask(); @@ -1641,7 +1657,14 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3 condition.NodeId, MaxDeferredEventsPerCondition); } } - else if (_deferredEventReports == 0 && before > 0 && _episodeWarned) + if (_deferredEventReports == 0 && before > 0) + { + // #854: every deferred event has been delivered, so no refresh cut taken during this episode can match + // anything any more; a new number makes them all stale at once (see RefreshCut.Episode). + _deferralEpisode++; + } + + if (_deferredEventReports == 0 && before > 0 && _episodeWarned) { _logger.LogInformation( "OtOpcUaNodeManager: deferred condition events drained ({Deferred} deferred, {Discarded} discarded this episode)", @@ -1704,21 +1727,94 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3 /// Under . public RefreshCut? Cut; + + /// + /// The refresh whose frozen replay is being queued to the item (0 = none). Set under Lock AND the gate + /// at the cut; cleared under the gate once queued. While set, every condition event delivered to the item + /// takes the gate, and one newer than the cut marks its condition in so the replay + /// does not follow it with an older state. + /// + public volatile int PendingReplay; + + /// Under : the watermark of . + public long PendingWatermark; + + /// Under : conditions an event newer than the pending replay already reached. + public readonly HashSet NewerSinceCut = []; } + private int _refreshGeneration; + + /// + /// One per Server-object item for the item's whole lifetime (#854 review): keyed by + /// the item object itself, so an unsubscribe + resubscribe of the same item between a refresh and the drain keeps its + /// cut and its gate, and a drain holding the state always holds the one a refresh installs into. + /// + private readonly ConditionalWeakTable _allEventsItemStates = new(); + + /// Server-object items currently holding a refresh cut (test observation, #854 review). + internal int ActiveRefreshCuts + { + get + { + var n = 0; + foreach (var item in _allEventsItems.Keys) + { + if (!_allEventsItemStates.TryGetValue(item, out var state)) continue; + lock (state.Gate) n += state.Cut is null ? 0 : 1; + } + + return n; + } + } + + private AllEventsItemState AllEventsItemStateOf(IEventMonitoredItem item) => + _allEventsItemStates.GetValue(item, static _ => new AllEventsItemState()); + + /// + /// Advanced (under the event-report gate) each time the deferral count returns to zero: the end of a deferral + /// episode, after which no event a cut could withhold is still undelivered (#854 review). + /// + private long _deferralEpisode; + /// What a Server-object item's latest ConditionRefresh conveyed (#854). /// Every condition event with a sequence at or below this was reported before the replay was read. - /// The conditions the replay re-sent to the item; the others were Retain=false (or not visible). - private sealed record RefreshCut(long Watermark, HashSet Refreshed) + /// The deferral episode the cut was taken in; once it ends the cut is stale and inert. + /// The conditions the replay re-sent to THIS item. + /// Every condition that was Retain at the watermark, whether or not this item was sent it + /// (one set shared by every item of the refresh). + private sealed record RefreshCut(long Watermark, long Episode, HashSet Refreshed, HashSet RetainedAtCut) { /// - /// Whether the refresh already told the item something at least as new as : a re-sent - /// condition's current state supersedes every older event of it; for a condition the replay OMITTED (it was not - /// retained), an older Retain=true event would wrongly re-add it, while an older Retain=false event agrees with - /// the omission and is still delivered (a client that did not reconcile on RefreshEnd needs it). + /// Whether the refresh already told the item something at least as new as . A re-sent + /// condition: its replayed state supersedes every older event of it. A condition the replay omitted because it + /// was NOT retained: an older Retain=true event would wrongly re-add it, so it is withheld; an older Retain=false + /// event agrees with the omission and is delivered (a client that did not reconcile on RefreshEnd needs it). A + /// condition that WAS retained but was omitted for this item (ACL / role at refresh time): never withheld — the + /// refresh told the item nothing about it, so a grant added before the drain still lets its events through + /// (decided by the ACL at delivery, as ever). /// public bool Supersedes(ConditionEventSnapshot e) => - e.Sequence <= Watermark && (e.Retain || Refreshed.Contains(e.ConditionId)); + e.Sequence <= Watermark + && (Refreshed.Contains(e.ConditionId) || (e.Retain && !RetainedAtCut.Contains(e.ConditionId))); + } + + /// + /// Drops every refresh cut taken in an episode that has ended (#854 review), so a cut does not outlive the events it + /// guards. Called by the drain after the deferral count reached zero, holding no lock; a cut installed concurrently + /// for the new episode is kept. + /// + private void ClearStaleRefreshCuts() + { + var current = Interlocked.Read(ref _deferralEpisode); + foreach (var item in _allEventsItems.Keys) + { + if (!_allEventsItemStates.TryGetValue(item, out var state)) continue; + lock (state.Gate) + { + if (state.Cut is { } cut && cut.Episode != current) state.Cut = null; + } + } } @@ -5362,7 +5458,7 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3 /// unsubscribe arms, which the SDK's MasterNodeManager drives for every create, modify, /// transfer, restore and delete of an all-events item, subscription teardown included. /// - private readonly ConcurrentDictionary _allEventsItems = new(); + private readonly ConcurrentDictionary _allEventsItems = new(); // The SDK's base implementation attaches a Server-object item to the event // fan-out of every one of this manager's root notifiers, so every alarm in the server reached it @@ -5384,7 +5480,7 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3 lock (Lock) { if (unsubscribe) _allEventsItems.TryRemove(monitoredItem, out _); - else _allEventsItems.GetOrAdd(monitoredItem, static _ => new AllEventsItemState()); + else _allEventsItems[monitoredItem] = 0; ReconcileAllEventsMonitoring(); } @@ -5560,7 +5656,7 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3 var reportingSession = (context as ISessionSystemContext)?.SessionId; var deferred = e is ConditionEventSnapshot { Deferred: true } d ? d : null; - foreach (var (item, state) in _allEventsItems) + foreach (var item in _allEventsItems.Keys) { if (e is AuditEventState && !IsAuditDeliverableTo(item)) continue; if (scopes is not null && !IsConditionVisibleTo(item, scopes)) continue; @@ -5572,25 +5668,36 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3 continue; } - if (deferred is null) + // #854: a synchronous event (under Lock) needs the gate only while a refresh's replay is pending, and the + // pending marker is installed under Lock, so an item no refresh has touched has nothing to order. A deferred + // event always takes the gate — its state is created here if need be, so a refresh racing it shares it. + AllEventsItemState? state; + if (deferred is not null) state = AllEventsItemStateOf(item); + else if (!_allEventsItemStates.TryGetValue(item, out state) || state.PendingReplay == 0 || e is not ConditionEventSnapshot) { item.QueueEvent(e); continue; } - // #854: a deferred event reaches this item off Lock, so a ConditionRefresh of the item may have replayed a - // newer state since it was reported. Decided and queued under the item's gate, which the refresh holds while - // it queues its replay and installs its cut — so it lands either before the replay or not at all. + // A deferred event reaches this item off Lock, so a ConditionRefresh of the item may have replayed a newer + // state since it was reported; and while a refresh's frozen replay is being queued, an event newer than it must + // not be followed by it. Both are decided and queued under the item's gate, which the refresh also holds for + // each replayed event it queues — so a deferred event lands before the replay or is judged against the cut. lock (state.Gate) { - if (state.Cut?.Supersedes(deferred) == true) + // A cut from an episode that has ended is stale (only waiting for ClearStaleRefreshCuts). + if (deferred is not null + && state.Cut is { } cut && cut.Episode == Interlocked.Read(ref _deferralEpisode) && cut.Supersedes(deferred)) { Interlocked.Increment(ref _refreshSupersededDeliveries); OtOpcUaTelemetry.ConditionEventDeliveriesSuperseded.Add(1); continue; } - DeferredServerObjectDeliveryProbe?.Invoke(); + if (state.PendingReplay != 0 && e is ConditionEventSnapshot { } ce && ce.Sequence > state.PendingWatermark) + state.NewerSinceCut.Add(ce.ConditionId); + + if (deferred is not null) DeferredServerObjectDeliveryProbe?.Invoke(); item.QueueEvent(e); } } @@ -6174,26 +6281,47 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3 /// public override ServiceResult ConditionRefresh(OperationContext context, IList monitoredItems) { - var direct = new List(); - var serverItems = new List(); - foreach (var item in monitoredItems) + // The base's NodeManagerOperation guard is private; this is its equivalent for the custom paths (#854 review): + // refused once disposal starts, and Dispose waits for the ones in flight. + Interlocked.Increment(ref _refreshesInFlight); + try { - if (item is null) continue; - (item.NodeId == ObjectIds.Server ? serverItems : direct).Add(item); - } + if (_disposing) throw new ObjectDisposedException(GetType().Name); - var enforced = AclEnforced; - var systemContext = SystemContext.Copy(context); - if (direct.Count > 0) + var direct = new List(); + var serverItems = new List(); + foreach (var item in monitoredItems) + { + if (item is null) continue; + (item.NodeId == ObjectIds.Server ? serverItems : direct).Add(item); + } + + var enforced = AclEnforced; + var systemContext = SystemContext.Copy(context); + if (direct.Count > 0) + { + if (enforced) RefreshDirectItems(systemContext, direct); + else base.ConditionRefresh(context, direct); + } + + if (serverItems.Count > 0) RefreshServerObjectItems(systemContext, serverItems, enforced); + return ServiceResult.Good; + } + finally { - if (enforced) RefreshDirectItems(systemContext, direct); - else base.ConditionRefresh(context, direct); + Interlocked.Decrement(ref _refreshesInFlight); } - - if (serverItems.Count > 0) RefreshServerObjectItems(systemContext, serverItems, enforced); - return ServiceResult.Good; } + private int _refreshesInFlight; + private volatile bool _disposing; + + /// + /// Test probe (#854 review): called once per Server-object refresh with whether it took the cut path and whether + /// the calling thread held Lock while it queued the replay. Null in production. + /// + internal Action? ServerObjectRefreshProbe { get; set; } + /// /// ConditionRefresh (and ConditionRefresh2) for Server-object (i=2253) items: each condition under a root /// notifier once, through the item's ACL decision when (#586) — and never overtaken by @@ -6211,84 +6339,224 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3 /// refresh had just dropped re-added, until the newest drained. /// /// - /// The cut. Everything — the watermark (), whether any event is - /// deferred, the replay itself and its queueing — happens in ONE hold of Lock, under which every condition - /// event is reported. So the replay is exactly the state at the watermark: an event at or below it is not newer - /// than the replay, one above it was reported after it. Each item's replay is queued and its - /// installed under the item's , which a deferred - /// delivery to the item also holds (): a deferred event therefore lands - /// before the replay, or is judged against the cut — withheld if superseded (). - /// Only events marked are judged: any other was delivered to - /// Server-object items synchronously under Lock, so it cannot follow a replay it predates. + /// Common case — nothing deferred: unchanged, and off Lock. The replay is the notifier walk the SDK + /// base does, queued as live condition nodes, with Lock taken only to copy the root-notifier list. It is + /// kept only if no deferral episode existed at any moment of it: the deferral count is zero before AND after and + /// the episode number () did not move. Otherwise a deferred event may have reached + /// the item after the replay, so the cut path below runs as well — its second replay, queued later inside the + /// same RefreshStart/RefreshEnd, supersedes anything that slipped in between. + /// + /// + /// Deferring — the cut. In ONE hold of Lock (under which every condition event is reported): the + /// watermark (), the episode, and a frozen snapshot of each Retain + /// condition read from (O(conditions), no tree walk), then queued to each item + /// with its installed under the item's . The ACL / + /// role verdicts are computed BEFORE taking Lock, against the live conditions (they depend on the event + /// type and source, not state). The cut and a "replay pending" marker are installed per item under its + /// inside that Lock hold; the replay is then queued OFF + /// Lock, one event per hold of the gate. While the marker is set every condition event to the item takes + /// the gate: one newer than the cut marks its condition so its (older) frozen replay entry is dropped instead of + /// following it; a deferred one at or below the cut is judged against the cut () + /// — it lands before the replay entry or not at all. /// /// /// Per item, never global. Only the refreshed item stops receiving the superseded events; the deferral - /// queue, every other Server-object item and every folder item still get them, in order. The refresh never - /// waits for a drain. No cut is kept while nothing is deferred (the common case). + /// queue, every other Server-object item and every folder item still get them, in order. A refresh waits for a + /// drain at most for one in-progress QueueEvent on that item (the gate), never for the drain itself. + /// Cuts die with their episode (). /// /// /// Lock order is Lock → item gate → the SDK item's own lock, the same as a synchronous report - /// (Lock → ); the drain takes the item gate without Lock, and - /// nothing under an item gate takes Lock or the event-report gate. + /// (Lock → ); the drain and the replay queueing take the item gate + /// without Lock, and nothing under an item gate takes Lock or the event-report gate. A gate is held + /// for one QueueEvent at a time, so a reporter holding Lock waits at most for one. /// /// /// With enforcement off this replaces the SDK base's Server-object loop with the same result: the base queued /// one copy per root notifier a condition is wired to, which the item's queue collapsed to one (they share an /// EventId); here it is one per condition. A replayed event that names no condition is withheld only under - /// enforcement (fail closed), as before. + /// enforcement (fail closed), as before. The cut path replays from , which holds + /// every condition this manager materialises (all of them sit under a root notifier); branches are replayed as + /// the SDK's ConditionState.ConditionRefresh lists them. /// /// private void RefreshServerObjectItems(ServerSystemContext systemContext, List items, bool enforced) { - lock (Lock) + long episodeBefore; + bool deferringBefore; + lock (_eventReportGate) { - var watermark = Interlocked.Read(ref _conditionEventSequence); - bool deferring; - lock (_eventReportGate) deferring = _deferredEventReports > 0; + episodeBefore = _deferralEpisode; + deferringBefore = _deferredEventReports > 0; + } + if (!deferringBefore) + { + List notifiers; + lock (Lock) notifiers = RootNotifiers?.ToList() ?? []; var events = new List(); - foreach (var notifier in RootNotifiers?.ToList() ?? []) notifier.ConditionRefresh(systemContext, events, true); - - // One event per condition, with its scopes resolved once. - var replay = new List<(IFilterTarget Event, NodeState? Condition, List? Scopes)>(); - var seen = new HashSet(System.Collections.Generic.ReferenceEqualityComparer.Instance); - foreach (var e in events) - { - if (ConditionOf(e) is not { } condition) - { - if (!enforced) replay.Add((e, null, null)); - continue; - } - - if (!seen.Add(condition)) continue; - replay.Add((e, condition, enforced ? ConditionVisibilityScopes(condition) : null)); - } + foreach (var notifier in notifiers) notifier.ConditionRefresh(systemContext, events, true); + var replay = DedupeReplay(events, enforced); + ServerObjectRefreshProbe?.Invoke(false, Monitor.IsEntered(Lock)); foreach (var item in items) { - var queued = new List(); - var refreshed = new HashSet(); - foreach (var (e, condition, scopes) in replay) + // A cut-path replay of this item still being queued by a concurrent refresh is older than this live one: + // stop it, so it cannot follow this replay with frozen states. + if (_allEventsItemStates.TryGetValue(item, out var pending)) + { + lock (pending.Gate) + { + pending.PendingReplay = 0; + pending.NewerSinceCut.Clear(); + } + } + + foreach (var (e, _, scopes) in replay) { if (scopes is not null && !IsConditionVisibleTo(item, scopes)) continue; if (ServiceResult.IsBad(ValidateEventRolePermissions(item, e))) continue; - queued.Add(e); - if (condition is not null) refreshed.Add(condition.NodeId); + item.QueueEvent(e); } + } - if (!_allEventsItems.TryGetValue(item, out var state)) + lock (_eventReportGate) + { + if (_deferralEpisode == episodeBefore && _deferredEventReports == 0) return; + } + } + + RefreshServerObjectItemsWithCut(systemContext, items, enforced); + } + + /// One replay entry per condition (the first the walk produced), with its ACL scopes resolved once. + private List<(IFilterTarget Event, NodeState? Condition, List? Scopes)> DedupeReplay( + List events, bool enforced) + { + var replay = new List<(IFilterTarget Event, NodeState? Condition, List? Scopes)>(); + var seen = new HashSet(System.Collections.Generic.ReferenceEqualityComparer.Instance); + foreach (var e in events) + { + if (ConditionOf(e) is not { } condition) + { + if (!enforced) replay.Add((e, null, null)); + continue; + } + + if (!seen.Add(condition)) continue; + replay.Add((e, condition, enforced ? ConditionVisibilityScopes(condition) : null)); + } + + return replay; + } + + /// The cut path of — see there. + private void RefreshServerObjectItemsWithCut(ServerSystemContext systemContext, List items, bool enforced) + { + // ACL / role verdicts per (item, condition) before Lock: they depend on the event type, source and grants only. + List conditions; + lock (Lock) conditions = [.. _alarmConditions.Values]; + var known = new HashSet(conditions, System.Collections.Generic.ReferenceEqualityComparer.Instance); + var allowed = new HashSet<(IEventMonitoredItem, AlarmConditionState)>(); + foreach (var condition in conditions) + { + var scopes = enforced ? ConditionVisibilityScopes(condition) : null; + foreach (var item in items) + { + if (scopes is not null && !IsConditionVisibleTo(item, scopes)) continue; + if (ServiceResult.IsBad(ValidateEventRolePermissions(item, condition))) continue; + allowed.Add((item, condition)); + } + } + + var states = items.Select(AllEventsItemStateOf).ToList(); + var perItem = new List<(IFilterTarget Event, AlarmConditionState Condition)>[items.Count]; + var generation = Interlocked.Increment(ref _refreshGeneration); + lock (Lock) + { + var watermark = Interlocked.Read(ref _conditionEventSequence); + long episode; + bool deferring; + lock (_eventReportGate) + { + episode = _deferralEpisode; + deferring = _deferredEventReports > 0; + } + + // Frozen at the watermark: the replay is exactly the state every event at or below it describes. + var retained = new HashSet(); + var replay = new List<(IFilterTarget Event, AlarmConditionState Condition)>(); + var branches = new List(); + foreach (var condition in _alarmConditions.Values) + { + if (condition.Retain?.Value != true) continue; + retained.Add(condition.NodeId); + branches.Clear(); + condition.ConditionRefresh(SystemContext, branches, true); + foreach (var target in branches) { - // Not (or no longer) subscribed: no deferred delivery can reach it, so there is nothing to cut. - foreach (var e in queued) item.QueueEvent(e); - continue; + if (target is not BaseInstanceState instance) continue; + var snapshot = new InstanceStateSnapshot(); + snapshot.Initialize(SystemContext, instance); + replay.Add((snapshot, condition)); + } + } + + for (var i = 0; i < items.Count; i++) + { + var item = items[i]; + var entries = new List<(IFilterTarget Event, AlarmConditionState Condition)>(); + var refreshed = new HashSet(); + foreach (var (e, condition) in replay) + { + // A condition materialised since the verdicts were taken is decided here (rare). + if (!allowed.Contains((item, condition))) + { + if (known.Contains(condition)) continue; + if (enforced && !IsConditionVisibleTo(item, ConditionVisibilityScopes(condition))) continue; + if (ServiceResult.IsBad(ValidateEventRolePermissions(item, e))) continue; + } + + entries.Add((e, condition)); + refreshed.Add(condition.NodeId); } + perItem[i] = entries; + var state = states[i]; lock (state.Gate) { - foreach (var e in queued) item.QueueEvent(e); - // A cut is only needed while an event at or below the watermark may still be delivered. Refreshes are - // serialised by Lock, so a later one always carries the later watermark and simply replaces it. - state.Cut = deferring ? new RefreshCut(watermark, refreshed) : null; + state.Cut = deferring ? new RefreshCut(watermark, episode, refreshed, retained) : null; + state.PendingWatermark = watermark; + state.NewerSinceCut.Clear(); + state.PendingReplay = generation; + } + } + } + + // Queued off Lock, one replayed event per hold of the item gate: an event that reached the item since the cut is + // newer than the frozen replay of its condition, so that replay entry is dropped rather than queued after it (the + // newer event is already inside RefreshStart..RefreshEnd). A later refresh of the item supersedes this one. + ServerObjectRefreshProbe?.Invoke(true, Monitor.IsEntered(Lock)); + for (var i = 0; i < items.Count; i++) + { + var item = items[i]; + var state = states[i]; + foreach (var (e, condition) in perItem[i]) + { + lock (state.Gate) + { + if (state.PendingReplay != generation) break; + if (state.NewerSinceCut.Contains(condition.NodeId)) continue; + item.QueueEvent(e); + } + } + + lock (state.Gate) + { + if (state.PendingReplay == generation) + { + state.PendingReplay = 0; + state.NewerSinceCut.Clear(); } } } diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/ConditionRefreshDeferralWireTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/ConditionRefreshDeferralWireTests.cs index b1d456e98..29886a6bd 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/ConditionRefreshDeferralWireTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/ConditionRefreshDeferralWireTests.cs @@ -9,6 +9,7 @@ using Xunit; using ZB.MOM.WW.OtOpcUa.Commons.OpcUa; using ZB.MOM.WW.OtOpcUa.Configuration.Entities; using ZB.MOM.WW.OtOpcUa.Configuration.Enums; +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; using ZB.MOM.WW.OtOpcUa.Core.Authorization; using MonitoredItem = Opc.Ua.Client.MonitoredItem; @@ -40,15 +41,21 @@ public sealed class ConditionRefreshDeferralWireTests /// (so the refresh omits it and its final Retain=false event must still arrive). /// /// ConditionRefresh2 on the single item instead of ConditionRefresh on the subscription. + /// + /// The Server-object item is unsubscribed and resubscribed (the SDK drives this on a monitoring-mode change) + /// between the refresh and the drain: its cut must survive (#854 review — it lived in the subscription map entry). + /// [Theory] - [InlineData(false, true, false)] - [InlineData(true, true, false)] - [InlineData(false, false, false)] - [InlineData(true, false, false)] - [InlineData(false, true, true)] - [InlineData(true, false, true)] + [InlineData(false, true, false, false)] + [InlineData(true, true, false, false)] + [InlineData(false, false, false, false)] + [InlineData(true, false, false, false)] + [InlineData(false, true, true, false)] + [InlineData(true, false, true, false)] + [InlineData(false, true, false, true)] + [InlineData(true, false, false, true)] public async Task Deferred_events_older_than_a_refresh_never_reach_the_refreshed_server_object_item_after_RefreshEnd( - bool enforced, bool finalRetained, bool refresh2) + bool enforced, bool finalRetained, bool refresh2, bool resubscribe) { var ct = TestContext.Current.CancellationToken; await using var rig = await ServerEventRig.StartAsync("Refresh854", encrypted: false, ct); @@ -100,10 +107,21 @@ public sealed class ConditionRefreshDeferralWireTests refresh.Elapsed.ShouldBeLessThan(TimeSpan.FromSeconds(10), "ConditionRefresh must not wait for the stalled drain"); await RecordedEvents.WaitUntilAsync(() => serverEvents.HasType(onServer, ObjectTypeIds.RefreshEndEventType)); rig.Nm.DeferredEventReports.ShouldBeGreaterThan(0, "the deferred events were still waiting at RefreshEnd"); + rig.Nm.ActiveRefreshCuts.ShouldBe(1); + if (resubscribe) + { + var serverItem = FindAllEventsItem(rig.Nm, onServer.Status.Id); + rig.Nm.SubscribeToAllEvents(null!, serverSubscription.Id, serverItem, unsubscribe: true); + rig.Nm.SubscribeToAllEvents(null!, serverSubscription.Id, serverItem, unsubscribe: false); + } + stall.Release(); } await RecordedEvents.WaitUntilAsync(() => rig.Nm.DeferredEventReports == 0); + // #854 review: a cut does not outlive its deferral episode. + await RecordedEvents.WaitUntilAsync(() => rig.Nm.ActiveRefreshCuts == 0); + rig.Nm.ActiveRefreshCuts.ShouldBe(0, "the cut is cleared when the deferral count returns to zero"); // Positive control: a later transition is delivered (and everything deferred before it has been decided). rig.Project(native: true, Active("after")); await RecordedEvents.WaitUntilAsync(() => serverEvents.Has(onServer, "after")); @@ -149,6 +167,144 @@ public sealed class ConditionRefreshDeferralWireTests folderEvents.Has(onA, "final").ShouldBeTrue(); } + /// + /// #854 review (ACL on): a condition the refresh omitted for the item because of its grants — not because it was + /// cleared — must not have its deferred events withheld: a grant added before the drain lets them through, and the + /// item must end on the condition's newest state (it would otherwise miss an active alarm entirely). + /// + [Fact] + public async Task Deferred_events_of_a_condition_the_refresh_omitted_for_want_of_a_grant_are_not_withheld() + { + var ct = TestContext.Current.CancellationToken; + await using var rig = await ServerEventRig.StartAsync("Refresh854Acl", encrypted: false, ct); + rig.KeyRoles = []; + rig.MaterialiseNative(); + var anonymousOnA = ServerEventRig.EquipmentGrant(EqA, NodePermissions.ReadOnly); + rig.Enforce(true, [ServerEventRig.Key()], anonymousOnA); + + // The stalled folder subscriber is anonymous (granted EqA); the refreshing key session holds no grant yet. + using var folderSession = await rig.ConnectAsync(anonymous: true, ct); + var folderEvents = new RecordedEvents(); + var folderSubscription = await RecordedEvents.SubscribeAsync(folderSession, folderEvents, ct); + var onA = RecordedEvents.AddEventItem(folderSubscription, ServerEventRig.UnsNodeId(folderSession, EqA)); + await folderSubscription.ApplyChangesAsync(ct); + + using var serverSession = await rig.ConnectAsync(anonymous: false, ct); + var serverEvents = new RecordedEvents(); + var serverSubscription = await RecordedEvents.SubscribeAsync(serverSession, serverEvents, ct); + var onServer = RecordedEvents.AddEventItem(serverSubscription, ObjectIds.Server); + await serverSubscription.ApplyChangesAsync(ct); + + rig.Project(native: true, Active("seed")); + await RecordedEvents.WaitUntilAsync(() => folderEvents.Has(onA, "seed")); + serverEvents.Has(onServer, "seed").ShouldBeFalse("the key holds no grant yet"); + await folderSubscription.SetPublishingModeAsync(false, ct); + + using (var stall = ItemLockStall.Hold(FindServerItem(rig.Nm, onA.Status.Id))) + { + for (var i = 0; i < FloodSize; i++) rig.Project(native: true, Active($"f{i}")); + rig.Project(native: true, Active("final")); + rig.Nm.DeferredEventReports.ShouldBeGreaterThan(0); + + await serverSession.CallAsync( + ObjectTypeIds.ConditionType, MethodIds.ConditionType_ConditionRefresh, ct, serverSubscription.Id); + await RecordedEvents.WaitUntilAsync(() => serverEvents.HasType(onServer, ObjectTypeIds.RefreshEndEventType)); + + // The grant arrives while the condition's events still wait. + rig.Enforce(true, [ServerEventRig.Key()], anonymousOnA, KeyEquipmentGrant(EqA)); + stall.Release(); + } + + await RecordedEvents.WaitUntilAsync(() => rig.Nm.DeferredEventReports == 0); + await RecordedEvents.WaitUntilAsync(() => serverEvents.Has(onServer, "final")); + var stream = serverEvents.Of(onServer).ToList(); + var end = stream.FindIndex(f => f[RecordedEvents.EventTypeField].GetNodeId() == ObjectTypeIds.RefreshEndEventType); + stream.Take(end).Count(f => f[RecordedEvents.AckedField].TryGetValue(out bool _)) + .ShouldBe(0, "the refresh omitted the condition: no grant"); + var after = stream.Skip(end + 1) + .Where(f => f[RecordedEvents.AckedField].TryGetValue(out bool _)) + .Select(RecordedEvents.Message) + .ToList(); + after.ShouldNotBeEmpty(); + after[^1].ShouldBe("final", "the item ends on the newest state of a condition it was never refreshed"); + after.Count(m => m!.StartsWith('f') && m != "final").ShouldBeGreaterThan(0, "its deferred events were delivered, not withheld"); + } + + /// + /// #854 review: the cut path queues its frozen replay OFF Lock. An event newer than the cut that reaches the + /// item in that window (here a synchronously delivered condition, reported from the probe between the cut and the + /// queueing) must not be followed by its condition's older frozen replay entry. + /// + [Fact] + public async Task An_event_newer_than_the_cut_is_never_followed_by_its_conditions_older_frozen_replay() + { + var ct = TestContext.Current.CancellationToken; + await using var rig = await ServerEventRig.StartAsync("Refresh854Pending", encrypted: false, ct); + rig.KeyRoles = []; + rig.MaterialiseNative(); + rig.Enforce(false, [ServerEventRig.Key()], KeyEquipmentGrant(EqA)); + // A second condition under the device folder only: no folder item, so its events are always synchronous. + const string Free = ServerEventRig.RawDeviceFolder + "/free"; + rig.Sink.MaterialiseAlarmCondition(Free, ServerEventRig.RawDeviceFolder, "free", "OffNormalAlarm", 700, AddressSpaceRealm.Raw, isNative: true); + rig.Sink.WriteAlarmCondition(Free, Active("x1"), DateTime.UtcNow, AddressSpaceRealm.Raw); + + using var folderSession = await rig.ConnectAsync(anonymous: false, ct); + var folderEvents = new RecordedEvents(); + var folderSubscription = await RecordedEvents.SubscribeAsync(folderSession, folderEvents, ct); + var onA = RecordedEvents.AddEventItem(folderSubscription, ServerEventRig.UnsNodeId(folderSession, EqA)); + await folderSubscription.ApplyChangesAsync(ct); + + using var serverSession = await rig.ConnectAsync(anonymous: false, ct); + var serverEvents = new RecordedEvents(); + var serverSubscription = await RecordedEvents.SubscribeAsync(serverSession, serverEvents, ct); + var onServer = RecordedEvents.AddEventItem(serverSubscription, ObjectIds.Server); + await serverSubscription.ApplyChangesAsync(ct); + + rig.Project(native: true, Active("seed")); + await RecordedEvents.WaitUntilAsync(() => folderEvents.Has(onA, "seed") && serverEvents.Has(onServer, "seed")); + await folderSubscription.SetPublishingModeAsync(false, ct); + + var injected = 0; + rig.Nm.ServerObjectRefreshProbe = (cut, _) => + { + if (cut && Interlocked.Exchange(ref injected, 1) == 0) + rig.Sink.WriteAlarmCondition(Free, Active("x2"), DateTime.UtcNow, AddressSpaceRealm.Raw); + }; + + using (var stall = ItemLockStall.Hold(FindServerItem(rig.Nm, onA.Status.Id))) + { + for (var i = 0; i < FloodSize; i++) rig.Project(native: true, Active($"f{i}")); + rig.Nm.DeferredEventReports.ShouldBeGreaterThan(0, "the cut path runs only while something is deferred"); + await serverSession.CallAsync( + ObjectTypeIds.ConditionType, MethodIds.ConditionType_ConditionRefresh, ct, serverSubscription.Id); + await RecordedEvents.WaitUntilAsync(() => serverEvents.HasType(onServer, ObjectTypeIds.RefreshEndEventType)); + stall.Release(); + } + + rig.Nm.ServerObjectRefreshProbe = null; + Volatile.Read(ref injected).ShouldBe(1); + var stream = serverEvents.Of(onServer).ToList(); + var end = stream.FindIndex(f => f[RecordedEvents.EventTypeField].GetNodeId() == ObjectTypeIds.RefreshEndEventType); + var free = stream.Take(end) + .Select(RecordedEvents.Message) + .Where(m => m is "x1" or "x2") + .ToList(); + free.ShouldContain("x2"); + free[^1].ShouldBe("x2", "the older frozen replay of the condition must not follow its newer event"); + } + + /// #854 review: the custom Server-object refresh is refused once the node manager is disposed, like the base's. + [Fact] + public async Task A_refresh_after_disposal_is_refused() + { + var ct = TestContext.Current.CancellationToken; + var rig = await ServerEventRig.StartAsync("Refresh854Dispose", encrypted: false, ct); + var nm = rig.Nm; + await rig.DisposeAsync(); + nm.Dispose(); + Should.Throw(() => nm.ConditionRefresh(null!, Array.Empty())); + } + /// /// The race the item gate closes, made deterministic: a deferred event is between its supersession check and its /// queueing to the Server-object item when the refresh arrives. The refresh must wait for that one delivery (so @@ -294,6 +450,7 @@ public sealed class ConditionRefreshDeferralWireTests } }, stop.Token, TaskCreationOptions.LongRunning, TaskScheduler.Default); + var refusedAsInProgress = 0; for (var r = 0; r < 40; r++) { var call = r % 3 == 2 @@ -301,7 +458,17 @@ public sealed class ConditionRefreshDeferralWireTests serverSubscription.Id, onServer.Status.Id) : serverSession.CallAsync(ObjectTypeIds.ConditionType, MethodIds.ConditionType_ConditionRefresh, ct, serverSubscription.Id); - await call.WaitAsync(TimeSpan.FromSeconds(20), ct); + try + { + await call.WaitAsync(TimeSpan.FromSeconds(20), ct); + } + catch (ServiceResultException ex) when (ex.StatusCode == StatusCodes.BadRefreshInProgress) + { + // The SDK answers the Call before the refresh body runs, so under load the previous refresh of this + // subscription may still be in progress; the SDK refuses the overlap (its rule, not ours). + refusedAsInProgress++; + } + await Task.Delay(TimeSpan.FromMilliseconds(80), ct); } @@ -311,16 +478,17 @@ public sealed class ConditionRefreshDeferralWireTests Volatile.Read(ref deferredSeen).ShouldBeGreaterThan(1, "the stress must actually have run with deferred events"); await RecordedEvents.WaitUntilAsync(() => rig.Nm.DeferredEventReports == 0); - var last = $"s{Volatile.Read(ref counter)}"; + var newest = $"s{Volatile.Read(ref counter)}"; await folderSubscription.SetPublishingModeAsync(true, ct); - await RecordedEvents.WaitUntilAsync(() => serverEvents.Has(onServer, last)); + await RecordedEvents.WaitUntilAsync(() => serverEvents.Has(onServer, newest)); await Task.Delay(TimeSpan.FromMilliseconds(500), ct); - // Walk the Server-object stream: after each RefreshEnd, a condition event must not be older than what that - // refresh replayed (the refresh replays the then-current state, so the floor is its replayed counter). - var floor = -1; + // Walk the Server-object stream (one condition). Outside a refresh its states never go backwards; and each refresh + // ends on the newest state the item has seen — its replay is never older than an event already delivered, and + // nothing older than it follows RefreshEnd. + var max = -1; + var last = -1; var replaying = false; - var replayedInThisRefresh = -1; var violations = new List(); foreach (var f in serverEvents.Of(onServer)) { @@ -328,33 +496,139 @@ public sealed class ConditionRefreshDeferralWireTests if (type == ObjectTypeIds.RefreshStartEventType) { replaying = true; - replayedInThisRefresh = -1; continue; } if (type == ObjectTypeIds.RefreshEndEventType) { replaying = false; - floor = Math.Max(floor, replayedInThisRefresh); + if (last < max) violations.Add($"a refresh ended on s{last} after s{max} had been delivered"); continue; } if (!f[RecordedEvents.AckedField].TryGetValue(out bool _)) continue; var n = int.Parse(RecordedEvents.Message(f)![1..], System.Globalization.CultureInfo.InvariantCulture); - if (replaying) - { - replayedInThisRefresh = n; - continue; - } - - if (n < floor) violations.Add($"s{n} after a refresh that replayed s{floor}"); + if (!replaying && n < max) violations.Add($"s{n} delivered after s{max}"); + max = Math.Max(max, n); + last = n; } violations.ShouldBeEmpty(); + refusedAsInProgress.ShouldBeLessThan(40, "most refreshes must actually have run"); rig.Nm.RefreshSupersededDeliveries.ShouldBeGreaterThan(0, "some refresh must actually have overtaken deferred events"); RecordedEvents.Message(serverEvents.Of(onServer).Last(f => f[RecordedEvents.AckedField].TryGetValue(out bool _))) - .ShouldBe(last, "the stream still converges on the newest state"); - await RecordedEvents.WaitUntilAsync(() => folderEvents.Has(onA, last)); + .ShouldBe(newest, "the stream still converges on the newest state"); + await RecordedEvents.WaitUntilAsync(() => folderEvents.Has(onA, newest)); + } + + /// + /// Review of 1622d07f: a Server-object refresh must not hold the node manager's Lock across a walk of a large + /// address space — ScadaBridge refreshes on every reconnect, and any session may loop ConditionRefresh. Measures + /// the longest any other thread waits for Lock while a refresh of 3000 retained conditions runs, with + /// nothing deferred (the common case) and with a deferral in progress, and pins that the common case does not + /// build or queue its replay under Lock. + /// + /// Whether a deferral episode is in progress when the refresh runs. + [Theory] + [InlineData(false)] + [InlineData(true)] + public async Task A_refresh_of_a_large_address_space_keeps_the_node_manager_lock_hold_short(bool deferring) + { + const int Conditions = 3000; + var ct = TestContext.Current.CancellationToken; + await using var rig = await ServerEventRig.StartAsync("Refresh854Lock", encrypted: false, ct); + rig.KeyRoles = []; + rig.MaterialiseNative(); + rig.Enforce(false, [ServerEventRig.Key()], KeyEquipmentGrant(EqA)); + for (var i = 0; i < Conditions; i++) + { + var id = $"{ServerEventRig.RawDeviceFolder}/c{i}"; + rig.Sink.MaterialiseAlarmCondition(id, ServerEventRig.RawDeviceFolder, $"c{i}", "OffNormalAlarm", 700, AddressSpaceRealm.Raw, isNative: true); + rig.Sink.WriteAlarmCondition(id, Active($"c{i}"), DateTime.UtcNow, AddressSpaceRealm.Raw); + } + + using var folderSession = await rig.ConnectAsync(anonymous: false, ct); + var folderEvents = new RecordedEvents(); + var folderSubscription = await RecordedEvents.SubscribeAsync(folderSession, folderEvents, ct); + var onA = RecordedEvents.AddEventItem(folderSubscription, ServerEventRig.UnsNodeId(folderSession, EqA)); + await folderSubscription.ApplyChangesAsync(ct); + + using var serverSession = await rig.ConnectAsync(anonymous: false, ct); + var serverEvents = new RecordedEvents(); + var serverSubscription = await RecordedEvents.SubscribeAsync(serverSession, serverEvents, ct); + var onServer = RecordedEvents.AddEventItem(serverSubscription, ObjectIds.Server); + onServer.QueueSize = 10_000; + await serverSubscription.ApplyChangesAsync(ct); + + rig.Project(native: true, Active("seed")); + await RecordedEvents.WaitUntilAsync(() => folderEvents.Has(onA, "seed") && serverEvents.Has(onServer, "seed")); + await folderSubscription.SetPublishingModeAsync(false, ct); + + var stall = deferring ? ItemLockStall.Hold(FindServerItem(rig.Nm, onA.Status.Id)) : null; + try + { + if (deferring) + { + for (var i = 0; i < FloodSize; i++) rig.Project(native: true, Active($"f{i}")); + rig.Nm.DeferredEventReports.ShouldBeGreaterThan(0); + } + + var probes = new System.Collections.Concurrent.ConcurrentQueue<(bool Cut, bool LockHeld)>(); + rig.Nm.ServerObjectRefreshProbe = (cut, held) => probes.Enqueue((cut, held)); + var waits = new List(); + for (var round = 0; round < 5; round++) + { + var ends = serverEvents.Of(onServer).Count(f => f[RecordedEvents.EventTypeField].GetNodeId() == ObjectTypeIds.RefreshEndEventType); + using var stop = new CancellationTokenSource(); + var maxWait = 0.0; + var prober = Task.Factory.StartNew(() => + { + var sw = new Stopwatch(); + while (!stop.IsCancellationRequested) + { + sw.Restart(); + lock (rig.Nm.Lock) { } + maxWait = Math.Max(maxWait, sw.Elapsed.TotalMilliseconds); + Thread.Yield(); + } + }, CancellationToken.None, TaskCreationOptions.LongRunning, TaskScheduler.Default); + + await serverSession.CallAsync( + ObjectTypeIds.ConditionType, MethodIds.ConditionType_ConditionRefresh, ct, serverSubscription.Id); + await RecordedEvents.WaitUntilAsync(() => serverEvents.Of(onServer) + .Count(f => f[RecordedEvents.EventTypeField].GetNodeId() == ObjectTypeIds.RefreshEndEventType) > ends); + await stop.CancelAsync(); + await prober; + waits.Add(maxWait); + } + + TestContext.Current.TestOutputHelper?.WriteLine( + $"#854 lock-hold deferring={deferring} conditions={Conditions}: max Lock wait per refresh (ms) = " + + string.Join(", ", waits.Select(w => w.ToString("F1", System.Globalization.CultureInfo.InvariantCulture)))); + // Neither path queues its replay under Lock; only a deferral makes a refresh take the cut path. + probes.ShouldNotBeEmpty(); + probes.ShouldAllBe(p => !p.LockHeld); + if (deferring) probes.ShouldContain(p => p.Cut); + else probes.ShouldAllBe(p => !p.Cut); + } + finally + { + stall?.Dispose(); + } + } + + /// The server-side Server-object (all-events) item behind a client item. + private static IEventMonitoredItem FindAllEventsItem(OtOpcUaNodeManager nm, uint id) + { + var map = (IDictionary)typeof(OtOpcUaNodeManager) + .GetField("_allEventsItems", BindingFlags.Instance | BindingFlags.NonPublic)! + .GetValue(nm)!; + foreach (var key in map.Keys) + { + if (key is IEventMonitoredItem item && item.Id == id) return item; + } + + throw new InvalidOperationException($"no server-side all-events item {id}"); } /// The server-side event item behind a client item, from the node manager's direct-item map. From a0e9a45565fc0897264b79b880985c1b4c5bbbd6 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Wed, 11 Nov 2026 07:39:25 -0500 Subject: [PATCH 3/3] fix(opcua): stale-cut clear keeps newer episodes, dispose barrier + timeout warning, replay marker cleared on throw (#854 re-review) --- .../OtOpcUaNodeManager.cs | 24 +++++++++++++++++-- 1 file changed, 22 insertions(+), 2 deletions(-) diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs index 95ceb77d4..050202a84 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs @@ -624,7 +624,10 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3 // #854 review: refuse new Server-object refreshes and let the ones in flight finish (the base's private // NodeManagerOperation guard does this for its own paths). _disposing = true; - SpinWait.SpinUntil(() => Volatile.Read(ref _refreshesInFlight) == 0, TimeSpan.FromSeconds(10)); + Interlocked.MemoryBarrier(); // the flag write must not be reordered after the in-flight read (#854 re-review) + if (!SpinWait.SpinUntil(() => Volatile.Read(ref _refreshesInFlight) == 0, TimeSpan.FromSeconds(10))) + _logger.LogWarning("Node manager disposed with {Count} Server-object ConditionRefresh call(s) still in flight after 10 s", + Volatile.Read(ref _refreshesInFlight)); Volatile.Read(ref _dataPlaneAuthority).Changed -= OnAuthorityChanged; _liveItems.Clear(); _eventItems.Clear(); @@ -1812,7 +1815,7 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3 if (!_allEventsItemStates.TryGetValue(item, out var state)) continue; lock (state.Gate) { - if (state.Cut is { } cut && cut.Episode != current) state.Cut = null; + if (state.Cut is { } cut && cut.Episode < current) state.Cut = null; // episodes only grow: never clear a newer cut (#854 re-review) } } } @@ -6472,6 +6475,10 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3 var states = items.Select(AllEventsItemStateOf).ToList(); var perItem = new List<(IFilterTarget Event, AlarmConditionState Condition)>[items.Count]; var generation = Interlocked.Increment(ref _refreshGeneration); + // #854 re-review: a throw between installing the "replay pending" markers and finishing the replay must not leave + // a marker set (every sync event to that item would keep taking the gate until its next refresh). + try + { lock (Lock) { var watermark = Interlocked.Read(ref _conditionEventSequence); @@ -6560,6 +6567,19 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3 } } } + } + finally + { + foreach (var state in states) + { + lock (state.Gate) + { + if (state.PendingReplay != generation) continue; + state.PendingReplay = 0; + state.NewerSinceCut.Clear(); + } + } + } } ///