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(); + } + } +}