From d9b23e0b958673c67876db8ec5fdc15970c5d990 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Wed, 11 Nov 2026 12:18:11 -0500 Subject: [PATCH 1/5] feat(opcua): per-item delivery for folder event items; folder ConditionRefresh gets the #854 cut (#855) Condition events reaching one of our notifiers no longer go through the SDK MonitoredNode2 channel: the walk serves the attached items itself, per item and per event (audit guard, uncached INodeManager3 verdict, item gate/cut, QueueEvent), as the Server-object path does. No notifier of ours defers any more; the #838 deferral queue and the #854 cut stay as the safety net and folder items share the cut (RefreshDirectItems, flag on and off). An event whose walk suspends is marked deferred before it suspends. Tests produce deferral with a notifier-stall probe instead of holding a folder item's SDK lock (which would now block the reporter). New FolderConditionRefreshWireTests; design note in docs/plans. Claude-Session: https://claude.ai/code/session_01GT67AD99zQQqpLCVm3p7xk --- .claude/rules/data-plane-auth.md | 2 +- ...26-11-11-folder-event-per-item-delivery.md | 102 +++ docs/security.md | 63 +- .../OtOpcUaNodeManager.cs | 520 ++++++++++++--- .../ConditionRefreshDeferralWireTests.cs | 70 +- .../EventQueueOverflowAclWireTests.cs | 31 +- .../FolderConditionRefreshWireTests.cs | 627 ++++++++++++++++++ .../FolderEventFloodWireTests.cs | 50 +- .../FolderEventVerdictInvalidationTests.cs | 8 +- .../NotifierStall.cs | 42 ++ 10 files changed, 1302 insertions(+), 213 deletions(-) create mode 100644 docs/plans/2026-11-11-folder-event-per-item-delivery.md create mode 100644 tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/FolderConditionRefreshWireTests.cs create mode 100644 tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/NotifierStall.cs diff --git a/.claude/rules/data-plane-auth.md b/.claude/rules/data-plane-auth.md index a93cb46bd..ded979d62 100644 --- a/.claude/rules/data-plane-auth.md +++ b/.claude/rules/data-plane-auth.md @@ -217,7 +217,7 @@ Two things to know before relying on any of this: 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 — + `ConditionRefreshDeferralWireTests`. ⚠️ **Since #855 folder items are delivered per item too:** at every notifier we own, `DeliverConditionEventAsync` serves the `MonitoredNode2`'s attached items itself (`DeliverToNotifierItems`: audit guard → `INodeManager3.ValidateEventRolePermissions` per event, UNCACHED → item gate/cut → `QueueEvent`), never through the SDK channel — so no notifier of ours defers, a flood under `Lock` defers nothing, and folder refreshes (`RefreshDirectItems`, flag on AND off) share the #854 cut. The deferral queue + cut remain as the safety net; an event whose walk suspends is marked deferred BEFORE suspending. Tests produce deferral with `NotifierStall` (the `NotifierDeliveryStall` probe) — holding a folder item's SDK lock now blocks the reporter. `FolderConditionRefreshWireTests`; `docs/plans/2026-11-11-folder-event-per-item-delivery.md`. 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`. Overflow WITH the ACL doing work (#838 item 2) — denied condition interleaved, folder + `i=2253` items, non-publishing client, concurrent Read, grant revoked diff --git a/docs/plans/2026-11-11-folder-event-per-item-delivery.md b/docs/plans/2026-11-11-folder-event-per-item-delivery.md new file mode 100644 index 000000000..72ea22512 --- /dev/null +++ b/docs/plans/2026-11-11-folder-event-per-item-delivery.md @@ -0,0 +1,102 @@ +# Folder (notifier) event items: per-item delivery (Gitea #855) + +Status: design for #855 (owner decision 2026-11-11: build per-item delivery). Background: #838 (deferral instead of +blocking under `Lock`), #854 (sequence cut on Server-object refresh), #826 (per-folder event ACL). + +## Problem + +A condition event reaches an event item on an equipment / raw device folder through the SDK's `MonitoredNode2` +for that folder: `OnReportEventAsync` writes it to a bounded channel (4096), and a per-node consumer task decides +each item's verdict (memoised per `(item, event type, SourceNode)`) and calls `QueueEvent`. That gives the node +manager no per-event, per-item decision point, so a `ConditionRefresh` of a folder item cannot withhold an older +event that is still in the channel or in our deferral queue: the refresh is overtaken (the #854 guarantee covers only +Server-object items). The channel's consumer also needs `Lock` on a verdict miss, which is why a burst under `Lock` +fills the channel and defers (#838). + +## Design + +### Where delivery hooks in + +`DeliverConditionEventAsync` already walks the condition's notifier chain hop by hop (our copy of the SDK's +`ReportEventAsync`, #838). At a hop whose `OnReportEventAsync` is the SDK `MonitoredNode2` of **a node this manager +owns** (`handler.Target is MonitoredNode2 m && m.Node == node`), the walk no longer calls the handler. It delivers the +event itself to that node's attached items (`m.EventMonitoredItems`), one by one: the SDK's audit guard, the verdict +(`INodeManager3.ValidateEventRolePermissions` — the SDK's type/source `ReceiveEvents` check plus, under enforcement, +the #826 live-notifier + grant check), then the item's gate and cut, then `QueueEvent`. + +This is exactly how Server-object items are served (`OnConditionReportEvent`, #586/#854): same verdict per event, same +gate, same cut, same `QueueEvent`. + +### Bypass / coexistence with `MonitoredNode2` + +- **Condition events bypass the channel** for our notifiers. The `MonitoredNode2` stays: the SDK still owns item + attachment (subscribe / unsubscribe / transfer / tombstone / revalidation stand-down detach), and its channel still + carries everything else that reaches the node — a condition's SDK-raised method **audit** events (only with auditing + on) and data-change notifications for attribute items on the folder. Those keep the SDK's cached verdict, so + `InvalidateFolderEventVerdicts` stays. +- A stood-down item is not in `EventMonitoredItems`, so it receives nothing, as before. +- A hop whose handler is anything else (none exists today: conditions are never event notifiers and variables cannot + host event items) still goes through the handler — the fallback that keeps the #838 deferral reachable. + +### Consequence for overload + +Every hop the node manager owns now completes synchronously: `QueueEvent` never blocks (the item's own queue discards +oldest at `QueueSize`), and the verdict runs on the reporter's thread, which already holds `Lock`. So a burst under one +`Lock` hold no longer defers at all, and the #838 deadlock shape (a consumer needing `Lock` while the reporter waits on +its channel) is gone for condition events. Folder items stay in report order per item, exactly like Server-object items. + +The #838 deferral queue and the #854 cut machinery remain as the safety net for a hop that does suspend (the fallback +above, or a future SDK change). Folder items are brought under the same cut: + +- An event whose delivery continues off `Lock` — queued in the deferral queue, **or** the in-flight event whose walk + suspended at a hop (new: marked when the walk suspends, so the hops after the suspension take the gate) — takes the + item's gate and is withheld when the item's latest refresh cut supersedes it (`RefreshCut.Supersedes`, unchanged). +- `ConditionRefresh` / `ConditionRefresh2` of folder items has the same two paths as the Server object, flag on and off: + - **Nothing deferred:** the SDK base walk of the item's notifier (off `Lock`, live condition nodes), per-item verdict + as before (#826 filter under enforcement). Kept only if no deferral episode existed at any moment of it; otherwise + the cut path runs too and supersedes it inside the same RefreshStart..RefreshEnd. Its closing check takes `Lock` + once: a report whose walk suspended (marked under `Lock` before suspending) has registered its deferral by then — + without it a folder refresh could pass the check in the instant between the suspension and the count. + - **Deferring — the cut:** verdicts per (item, condition) first, off `Lock`, from a live walk. Then one `Lock` hold: + watermark, episode, the Retain set, and per item the same walk producing frozen snapshots, with the cut + "replay + pending" marker installed under the item's gate. The replay is queued off `Lock`, one event per gate hold. +- State per item (`EventItemState`, previously Server-object only) is keyed by the item object for its lifetime, so it + survives unsubscribe + resubscribe; stale cuts are cleared when the deferral episode ends, for folder and Server items. + +### Locking + +Unchanged order, now for folder items too: `Lock` → item gate → the SDK item's own lock. Synchronous delivery runs +under the reporter's `Lock`; the verdict takes `Lock` re-entrantly and the core node manager's / authorizer's locks +(as the Server-object path already does). Off-`Lock` (deferred) delivery computes the verdict **before** taking the item +gate, so nothing under a gate takes `Lock` or the event-report gate. A reporter waits at most for one in-progress +`QueueEvent` on an item; a refresh waits at most for one deferred delivery already inside the gate, never for a drain. +No reporter ever waits on a channel. + +### Bounds + +No new queue. Per item: the SDK item queue (`QueueSize`, DiscardOldest). Deferral (only on the fallback): unchanged, +1024 per condition / 65 536 total, coalescing. Cuts: one per refreshed item, sets sized by the conditions, cleared at the +end of the deferral episode. + +### ACL (#826) + +Each event is decided per item at delivery against the current generation — no memoised verdict for condition events +any more, so a deploy / revocation / reference removal takes effect on the next event without waiting for an +invalidation. Refresh replays only conditions the item may see (the same check); a condition omitted for want of a grant +is never withheld afterwards (the cut's existing rule). + +### Cost + +The per-event verdict for folder items moves from the SDK's consumer task (cached) to the reporter thread under `Lock` +(uncached), as Server-object items already pay. Folder subscribers are few (ScadaBridge subscribes at `i=2253`). + +## Tests + +- `ConditionRefreshDeferralWireTests` / `FolderEventFloodWireTests` / `EventQueueOverflowAclWireTests`: deferral is no + longer produced by flooding a folder item; tests that need it stall the folder's hop through a test probe + (`NotifierDeliveryStall`, null in production) instead of the folder item's SDK lock (which would now block the reporter, + as it would on the Server object). Flood tests now pin that folder delivery never defers. +- New `FolderConditionRefreshWireTests`: folder-item refresh during deferral (flag on/off, retained / cleared, + ConditionRefresh2, resubscribe), the gate race, newer-than-cut replay, ACL omission not withheld, ACL revocation during + deferral, the in-flight suspended event, and two stress tests (production overload with no stall; stall + refresh + + reports concurrently). diff --git a/docs/security.md b/docs/security.md index ec400d57b..1e09644b1 100644 --- a/docs/security.md +++ b/docs/security.md @@ -901,7 +901,7 @@ Behaviours worth knowing before enabling it: 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 + - The deferral queues, every other Server-object item and every non-refreshing 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. @@ -912,29 +912,48 @@ Behaviours worth knowing before enabling it: - **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 - 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. + - **Folder (notifier) items are delivered per item (Gitea #855).** Before #855 a folder item was fed by the SDK's own + `MonitoredNode2` channel (up to 4096 events accepted before a refresh), with the per-item verdict memoised by (item, + event type, source), so a refresh of a folder item could still be followed by older events. Now, at every notifier + the node manager owns, the walk serves the attached items itself (`DeliverToNotifierItems`), exactly as it serves + Server-object items: the SDK's audit guard, the verdict through `INodeManager3.ValidateEventRolePermissions` (SDK + type/source `ReceiveEvents` + the #826 decision under enforcement) **per event, uncached**, then the item's gate and + cut, then `QueueEvent`. The `MonitoredNode2` stays for item attachment and for what still uses its channel (a + condition's SDK method audit events, data-change items on the folder), so `InvalidateFolderEventVerdicts` stays. + - **No notifier of ours defers any more.** `QueueEvent` never waits for a consumer (the item queue discards oldest at + `QueueSize`) and the verdict runs on the reporter's thread, which already holds `Lock` — so a burst under one + `Lock` hold completes with nothing deferred, and the #838 deadlock shape is gone for condition events. Folder items + therefore see condition events in report order. The cost moves with it: the per-event verdict and filter for + folder items now run under `Lock` (as for Server-object items), where the SDK used to cache the verdict off it. + - **The deferral queue and the cut stay as the safety net**, for a hop that is not ours (none exists today: + conditions are never event notifiers and variables cannot host event items) — and folder items share the cut: + a `ConditionRefresh`/`ConditionRefresh2` of a folder item takes the same two paths as the Server object (plain walk + off `Lock` when nothing is deferred; otherwise verdicts off `Lock`, then one `Lock` hold for the watermark, the + Retain set and, per item, its notifier walk frozen as snapshots, with the cut installed under the item gate and the + replay queued off `Lock`). An event whose walk suspended at a hop is marked deferred **before** it suspends, so + every later hop judges it against the cut; the plain refresh path's closing check takes `Lock` once, so a report + that suspended under it has registered its deferral by then. Lock order is unchanged (`Lock` → item gate → item + lock); off `Lock` the verdict is computed before the gate. 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 (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). + complete **with nothing deferred** (fails if the per-item delivery stops recognising the SDK fan-out), and the item + delivers at most its revised `QueueSize` in increasing order ending with the last event; and a condition **removed** + while its events wait (delivery stalled at the folder) has none of them delivered to either the folder or the + Server-object item. Tests that need a deferral stall delivery at a folder through a test probe + (`NotifierStall` → `OtOpcUaNodeManager.NotifierDeliveryStall`, null in production), which suspends the walk exactly + as a full channel did — holding a folder item's SDK lock, as the pre-#855 tests did, would now block the reporter. + `ConditionRefreshDeferralWireTests` (#854) refreshes the Server-object item while the folder is stalled + (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 (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; 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 a stress test runs reports, + refreshes and an intermittent stall concurrently. `FolderConditionRefreshWireTests` (#855) is the folder-item twin: + the same matrix on a refreshed folder item (a second subscriber on the same folder keeps the whole tail; a + ModifyMonitoredItems re-subscribe keeps the cut), the gate race, a newer-than-cut event on the raw device folder, a + condition omitted by the ACL (reference dropped) not withheld once it returns, a revocation after the refresh, a + production-shape overload (bursts under `Lock` + repeated refreshes, no stall: nothing deferred, the stream never + goes backwards, no replay queued under `Lock`) and a stress test with an intermittent stall. **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 050202a84..c3d96cc82 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs @@ -1476,7 +1476,14 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3 /// synchronous: the walk completes inline while every channel has room. /// /// - /// Deferral. When a channel is full the walk continues asynchronously (holding no thread or lock), + /// Since Gitea #855 no notifier of ours is a channel. At a notifier this manager owns, the walk serves + /// the items itself () instead of writing to the SDK's + /// MonitoredNode2 channel, so it always completes inline and the deferral below is reached only through + /// a hop that is not ours (none exists today) — or the test stall (). It is + /// kept as the safety net, and the refresh cut (#854) covers folder items too. + /// + /// + /// Deferral. When a hop suspends the walk continues asynchronously (holding no thread or lock), /// and the condition gets a queue: its later events wait behind it, so ONE condition's events stay in /// order, while other conditions whose channels have room are not held up (per-condition, not global). /// Events of different conditions may interleave differently than reported — each carries its own @@ -1615,7 +1622,30 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3 node.OnReportEvent?.Invoke(SystemContext, node, e); if (node.OnReportEventAsync is { } handler) - await handler(SystemContext, node, e, CancellationToken.None).ConfigureAwait(false); + { + if (OwnNotifierFanOut(node, handler) is { } monitored) + { + // Gitea #855: one of our notifiers. Its items are served here, per item and per event, exactly as the + // Server-object items are (OnConditionReportEvent) — never through the SDK's channel, whose consumer + // needed Lock (#838) and gave no per-event, per-item point at which a refresh cut could apply. + if (NotifierDeliveryStall?.Invoke(node) is { IsCompleted: false } stall) + { + MarkDeferred(e); + await stall.ConfigureAwait(false); + } + + // Re-checked: a stall may have outlived the condition (the liveness check above ran before it). + if (IsLiveCondition(condition)) DeliverToNotifierItems(monitored, e); + } + else + { + // Not a fan-out of ours (none exists today): the SDK path, which may suspend. Whatever is delivered after + // the suspension runs off Lock, so it must be judged as a deferred event (#855). + var pending = handler(SystemContext, node, e, CancellationToken.None); + if (!pending.IsCompleted) MarkDeferred(e); + await pending.ConfigureAwait(false); + } + } var notifiers = new List(); node.GetNotifiers(SystemContext, notifiers); @@ -1631,6 +1661,81 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3 await DeliverConditionEventAsync(condition, parent, e).ConfigureAwait(false); } + /// Marks an event whose remaining delivery will run off Lock (#855) — before the walk suspends. + private static void MarkDeferred(IFilterTarget e) + { + if (e is ConditionEventSnapshot snapshot) snapshot.Deferred = true; + } + + /// + /// The SDK serving event items on , when that node is one of + /// ours and is the fan-out's own OnReportEventAsync (what + /// MonitoredNode2.Add(IEventMonitoredItem) installs); otherwise null (Gitea #855). + /// + private MonitoredNode2? OwnNotifierFanOut(NodeState node, NodeStateReportEventAsyncHandler handler) => + handler.Target is MonitoredNode2 monitored && ReferenceEquals(monitored.Node, node) && OwnsNamespace(node.NodeId) + ? monitored + : null; + + /// + /// Test probe (#855): consulted before a condition event is delivered to the items of one of our notifiers; an + /// incomplete task is awaited first, so the walk suspends there exactly as it would behind a slow notifier — the + /// way tests produce a deferral now that no notifier of ours can block. Null in production. + /// + internal Func? NotifierDeliveryStall { get; set; } + + /// + /// Delivers one condition event to the event items attached to one of this manager's notifiers (an equipment or + /// raw device folder), item by item (Gitea #855) — the folder twin of . + /// + /// The notifier's SDK fan-out; only its attached-item list is used. + /// The condition's snapshot. + /// + /// + /// Mirrors the SDK's MonitoredNode2.ProcessEventForItemAsync — its audit guard, then the verdict through + /// (the SDK's type + SourceNode ReceiveEvents + /// check and, under enforcement, #826's live-notifier + grant check), then QueueEvent — but per event: + /// nothing is memoised, so a deploy, revocation or reference removal decides the very next event. The item's + /// gate and refresh cut apply exactly as on the Server object (). + /// + /// + /// Synchronous. On a report it runs on the reporter's thread under Lock (as the Server-object delivery + /// does), and QueueEvent never waits for a consumer — the item's own queue discards oldest at its + /// QueueSize — so no notifier of ours defers any more. Off Lock (a deferred event) the verdict, + /// which takes Lock, is computed before the item gate is taken: nothing under a gate takes Lock. + /// A stood-down item (#562) is not in the attached list and receives nothing, as before. + /// + /// + private void DeliverToNotifierItems(MonitoredNode2 monitored, IFilterTarget e) + { + var deferred = e is ConditionEventSnapshot { Deferred: true } d ? d : null; + foreach (var item in monitored.EventMonitoredItems.Values) + { + if (item is null) continue; + try + { + if (IsAuditEvent(e) && !IsAuditDeliverableTo(item)) continue; + if (ServiceResult.IsBad(((INodeManager3)this).ValidateEventRolePermissions(item, e))) continue; + QueueConditionEventToItem(item, e, deferred, DeferredFolderDeliveryProbe); + } + catch (Exception ex) + { + // As the SDK's per-item delivery: one item's failure must not cost the others (or the rest of the walk). + _logger.LogWarning(ex, "OtOpcUaNodeManager: failed to deliver a condition event to event item {ItemId}", item.Id); + } + } + } + + /// The SDK's audit test from MonitoredNode2.ProcessEventForItemAsync: the event, or the snapshot's handle. + private static bool IsAuditEvent(IFilterTarget e) => + e is AuditEventState || e is InstanceStateSnapshot { Handle: AuditEventState }; + + /// + /// Test probe (#855): the folder twin of — runs between a deferred + /// event's supersession check and its queueing to a folder item, inside the item's gate. Null in production. + /// + internal Action? DeferredFolderDeliveryProbe { get; set; } + /// Whether is the instance the address space currently holds for its NodeId. private bool IsLiveCondition(NodeState condition) => _alarmConditions.TryGetValue(MapKey(condition.NodeId), out var live) && ReferenceEquals(live, condition); @@ -1711,18 +1816,27 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3 public bool Retain { get; } = retain; + private volatile bool _deferred; + /// - /// 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. + /// Set when the event's delivery continues off Lock: under the event-report gate, before the drain can + /// dequeue it, when it waits in a deferral queue; or by the walk itself, before it suspends at a hop (#855), for + /// the in-flight event whose remaining hops then run on another thread. Only such an event reaches an item off + /// Lock, which is the only way a refresh can be overtaken. /// - public bool Deferred { get; set; } + public bool Deferred + { + get => _deferred; + set => _deferred = value; + } } /// - /// 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. + /// Per event item state — Server-object (#854) and, since #855, folder items: 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 + private sealed class EventItemState { /// 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). @@ -1749,21 +1863,21 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3 private int _refreshGeneration; /// - /// One per Server-object item for the item's whole lifetime (#854 review): keyed by + /// One per event 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(); + private readonly ConditionalWeakTable _eventItemStates = new(); - /// Server-object items currently holding a refresh cut (test observation, #854 review). + /// Server-object and folder event items currently holding a refresh cut (test observation, #854 review). internal int ActiveRefreshCuts { get { var n = 0; - foreach (var item in _allEventsItems.Keys) + foreach (var item in TrackedEventItems()) { - if (!_allEventsItemStates.TryGetValue(item, out var state)) continue; + if (!_eventItemStates.TryGetValue(item, out var state)) continue; lock (state.Gate) n += state.Cut is null ? 0 : 1; } @@ -1771,8 +1885,15 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3 } } - private AllEventsItemState AllEventsItemStateOf(IEventMonitoredItem item) => - _allEventsItemStates.GetValue(item, static _ => new AllEventsItemState()); + private EventItemState EventItemStateOf(IEventMonitoredItem item) => + _eventItemStates.GetValue(item, static _ => new EventItemState()); + + /// + /// Every live event item this manager tracks: the Server-object items and the direct (folder) items (#855). An + /// item that left both maps keeps its state in until collected, but any cut it holds + /// is from an ended episode by the time it could deliver again, and so inert. + /// + private IEnumerable TrackedEventItems() => _allEventsItems.Keys.Concat(_eventItems.Keys); /// /// Advanced (under the event-report gate) each time the deferral count returns to zero: the end of a deferral @@ -1810,9 +1931,9 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3 private void ClearStaleRefreshCuts() { var current = Interlocked.Read(ref _deferralEpisode); - foreach (var item in _allEventsItems.Keys) + foreach (var item in TrackedEventItems()) { - if (!_allEventsItemStates.TryGetValue(item, out var state)) continue; + if (!_eventItemStates.TryGetValue(item, out var state)) continue; lock (state.Gate) { if (state.Cut is { } cut && cut.Episode < current) state.Cut = null; // episodes only grow: never clear a newer cut (#854 re-review) @@ -5671,38 +5792,51 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3 continue; } - // #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) + QueueConditionEventToItem(item, e, deferred, DeferredServerObjectDeliveryProbe); + } + } + + /// + /// Queues a condition event to one event item — Server-object or folder (#855) — under the item's refresh gate and + /// cut (#854). The caller has already decided the item may receive it, and holds no item gate. + /// + /// The item. + /// The event. + /// The event, when its delivery runs off Lock (). + /// Test probe run inside the gate, before queueing a deferred event. + private void QueueConditionEventToItem(IEventMonitoredItem item, IFilterTarget e, ConditionEventSnapshot? deferred, Action? probe) + { + // #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. + EventItemState? state; + if (deferred is not null) state = EventItemStateOf(item); + else if (!_eventItemStates.TryGetValue(item, out state) || state.PendingReplay == 0 || e is not ConditionEventSnapshot) + { + item.QueueEvent(e); + return; + } + + // 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) + { + // 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)) { - item.QueueEvent(e); - continue; + Interlocked.Increment(ref _refreshSupersededDeliveries); + OtOpcUaTelemetry.ConditionEventDeliveriesSuperseded.Add(1); + return; } - // 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) - { - // 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; - } + if (state.PendingReplay != 0 && e is ConditionEventSnapshot { } ce && ce.Sequence > state.PendingWatermark) + state.NewerSinceCut.Add(ce.ConditionId); - 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); - } + if (deferred is not null) probe?.Invoke(); + item.QueueEvent(e); } } @@ -6080,8 +6214,11 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3 /// would be. /// /// - /// Runs on the SDK's per-node delivery task, where the inherited role-permission check already takes - /// Lock (via GetNodeMetadata); the scope resolution here takes it the same way. + /// Runs per event, uncached, from the per-item folder delivery (#855) — on the reporter's thread under + /// Lock, or off it for a deferred event (before the item gate is taken) — and, for the events that still + /// travel the SDK channel (a condition's method audits), on the SDK's per-node delivery task, where the + /// inherited role-permission check already takes Lock (via GetNodeMetadata); the scope + /// resolution here takes it the same way. /// /// private bool IsNotifierEventVisibleTo(IEventMonitoredItem item, IFilterTarget e) @@ -6146,13 +6283,15 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3 /// Good to deliver; otherwise the veto. /// /// - /// The SDK's MonitoredNode2 asks this, per event and per item, before it queues an event to an item - /// attached to one of this manager's notifiers. The Server-object path (#586) and ConditionRefresh decide - /// the same question themselves, so a folder's subscribers stop receiving a moved alarm (or a native alarm - /// whose reference was dropped) the moment the generation says so, not when the rebuild lands. + /// Condition events to an item attached to one of this manager's notifiers are decided here per event and per + /// item by (#855, uncached); the SDK's MonitoredNode2 still asks it + /// for what travels its channel (a condition's method audit events). The Server-object path (#586) and + /// ConditionRefresh decide the same question themselves, so a folder's subscribers stop receiving a moved + /// alarm (or a native alarm whose reference was dropped) the moment the generation says so, not when the + /// rebuild lands. /// /// - /// ⚠️ The SDK MEMOIZES the verdict per (item, event type, SourceNode) until the monitored node's + /// ⚠️ On the SDK channel path the SDK MEMOIZES the verdict per (item, event type, SourceNode) until the monitored node's /// permission generation moves, so must run whenever the inputs /// change: a new authority generation or authorizer, the revalidation tick, and the end of every apply. A /// condition's SourceNode is the condition itself, so a verdict is never shared between two conditions. @@ -6278,9 +6417,9 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3 // 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. + // Items on one of this manager's own notifiers go through RefreshDirectItems in both modes since #855 — the + // base's loop (a stood-down item is off its notifier's fan-out, so not refreshed) plus the #826 decision with + // enforcement on, and the same cut against deferred events as the Server object. /// public override ServiceResult ConditionRefresh(OperationContext context, IList monitoredItems) { @@ -6301,11 +6440,7 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3 var enforced = AclEnforced; var systemContext = SystemContext.Copy(context); - if (direct.Count > 0) - { - if (enforced) RefreshDirectItems(systemContext, direct); - else base.ConditionRefresh(context, direct); - } + if (direct.Count > 0) RefreshDirectItems(systemContext, direct, enforced); if (serverItems.Count > 0) RefreshServerObjectItems(systemContext, serverItems, enforced); return ServiceResult.Good; @@ -6339,7 +6474,9 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3 /// 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. + /// refresh had just dropped re-added, until the newest drained. (Since #855 folder items are no longer a + /// channel, so nothing of ours defers today; the cut stays as the safety net, and folder items share it — + /// .) /// /// /// Common case — nothing deferred: unchanged, and off Lock. The replay is the notifier walk the SDK @@ -6353,10 +6490,10 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3 /// 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 / + /// 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 + /// 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 () @@ -6406,14 +6543,7 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3 { // 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(); - } - } + StopPendingReplay(item); foreach (var (e, _, scopes) in replay) { @@ -6472,7 +6602,7 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3 } } - var states = items.Select(AllEventsItemStateOf).ToList(); + var states = items.Select(EventItemStateOf).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 @@ -6540,20 +6670,34 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3 } } - // 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)); + QueueFrozenReplay(items, states, perItem.Select(entries => entries.Select(x => (x.Event, x.Condition.NodeId)).ToList()).ToArray(), generation); + } + finally + { + ClearPendingReplays(states, generation); + } + } + + /// + /// Queues a cut-path refresh's frozen replay to its items OFF Lock, one replayed event per hold of the item + /// gate (#854; folder items too since #855): 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. + /// + private static void QueueFrozenReplay( + List items, List states, List<(IFilterTarget Event, NodeId ConditionId)>[] perItem, int generation) + { for (var i = 0; i < items.Count; i++) { var item = items[i]; var state = states[i]; - foreach (var (e, condition) in perItem[i]) + foreach (var (e, conditionId) in perItem[i]) { lock (state.Gate) { if (state.PendingReplay != generation) break; - if (state.NewerSinceCut.Contains(condition.NodeId)) continue; + if (state.NewerSinceCut.Contains(conditionId)) continue; item.QueueEvent(e); } } @@ -6567,51 +6711,227 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3 } } } - } - finally + } + + /// + /// #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). + /// + private static void ClearPendingReplays(List states, int generation) + { + foreach (var state in states) { - foreach (var state in states) + lock (state.Gate) { - lock (state.Gate) - { - if (state.PendingReplay != generation) continue; - state.PendingReplay = 0; - state.NewerSinceCut.Clear(); - } + if (state.PendingReplay != generation) continue; + state.PendingReplay = 0; + state.NewerSinceCut.Clear(); } } } /// - /// ConditionRefresh for event items on this manager's own notifiers, under enforcement (Gitea #826): the SDK - /// base's loop exactly — an item no longer attached (stood down by revalidation, or tombstoned) is not - /// refreshed — plus the per-event decision live delivery makes (), so a - /// refresh cannot replay to a folder's subscriber an alarm that folder's live events withhold. + /// Stops a cut-path replay of still being queued by a concurrent refresh: a live replay that + /// is about to be queued is newer, and the frozen one must not follow it. + /// + private void StopPendingReplay(IEventMonitoredItem item) + { + if (!_eventItemStates.TryGetValue(item, out var pending)) return; + lock (pending.Gate) + { + pending.PendingReplay = 0; + pending.NewerSinceCut.Clear(); + } + } + + /// + /// ConditionRefresh (and ConditionRefresh2) for event items on this manager's own notifiers — equipment and raw device + /// folders — flag on and off: never overtaken by a deferred event older than what it replays (Gitea #855), the folder + /// twin of . /// /// The refresh's system context. /// The direct (non-Server-object) items being refreshed. - private void RefreshDirectItems(ServerSystemContext systemContext, List items) + /// Whether the data-plane ACL is enforced. + /// + /// + /// Nothing deferred (the only case today — no notifier of ours defers since #855): unchanged, off + /// Lock. The SDK base's loop — an item no longer attached (stood down by revalidation, or tombstoned) + /// is not refreshed; the item's notifier is walked and each live condition queued through the SDK's role check — + /// plus, under enforcement, the per-event decision live delivery makes (, + /// #826), so a refresh cannot replay to a folder's subscriber an alarm that folder's live events withhold. It + /// stands only if no deferral episode existed at any moment of it; the closing check takes Lock once, so a + /// report whose walk suspended (it marks the event before suspending, under Lock) has been counted by then. + /// Otherwise the cut path runs as well, and its replay — later inside the same RefreshStart..RefreshEnd — + /// supersedes anything that slipped in between. + /// + /// + /// Deferring — the cut, exactly as on the Server object: verdicts per (item, condition) first, off + /// Lock, from a live walk; then one Lock hold takes the watermark, the episode, the Retain set and, + /// per item, the same walk as frozen snapshots, and installs the cut and the "replay pending" marker under the + /// item's gate; the replay is queued off Lock (). Deferred events reaching + /// the item afterwards are judged against the cut (). + /// + /// + private void RefreshDirectItems(ServerSystemContext systemContext, List items, bool enforced) { - foreach (var item in items) + long episodeBefore; + bool deferringBefore; + lock (_eventReportGate) { - NodeState? notifier; - lock (Lock) + episodeBefore = _deferralEpisode; + deferringBefore = _deferredEventReports > 0; + } + + if (!deferringBefore) + { + foreach (var item in items) { - if (!MonitoredItems.ContainsKey(item.Id)) continue; - notifier = (item.ManagerHandle as NodeHandle)?.Node; + var notifier = AttachedNotifierOf(item); + if (notifier is null) continue; + + StopPendingReplay(item); + var events = new List(); + notifier.ConditionRefresh(systemContext, events, true); + DirectRefreshProbe?.Invoke(false, Monitor.IsEntered(Lock)); + foreach (var e in events) + { + if (!IsDirectRefreshEventAllowed(item, e, enforced)) continue; + item.QueueEvent(e); + } } - if (notifier is null) continue; + // Lock first: a report that suspended under it has registered its deferral by the time Lock is free. + lock (Lock) + { + lock (_eventReportGate) + { + if (_deferralEpisode == episodeBefore && _deferredEventReports == 0) return; + } + } + } + RefreshDirectItemsWithCut(systemContext, items, enforced); + } + + /// The notifier an attached direct item subscribes to, or null for one that is not attached (not refreshed). + private NodeState? AttachedNotifierOf(IEventMonitoredItem item) + { + lock (Lock) + { + return MonitoredItems.ContainsKey(item.Id) ? (item.ManagerHandle as NodeHandle)?.Node : null; + } + } + + /// A direct item's refresh verdict for one replayed event: the #826 decision under enforcement, then the SDK's. + private bool IsDirectRefreshEventAllowed(IEventMonitoredItem item, IFilterTarget e, bool enforced) => + (!enforced || IsNotifierEventVisibleTo(item, e)) && !ServiceResult.IsBad(ValidateEventRolePermissions(item, e)); + + /// + /// Test probe (#855): called by a direct-item 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? DirectRefreshProbe { get; set; } + + /// The cut path of — see there. + private void RefreshDirectItemsWithCut(ServerSystemContext systemContext, List items, bool enforced) + { + // Verdicts per (item, condition) before Lock, from a live walk: they depend on the event type, source, notifier + // wiring and grants, not on the condition's state. + var allowed = new HashSet<(IEventMonitoredItem, NodeState)>(); + var known = new HashSet(System.Collections.Generic.ReferenceEqualityComparer.Instance); + foreach (var item in items) + { + if (AttachedNotifierOf(item) is not { } notifier) continue; var events = new List(); notifier.ConditionRefresh(systemContext, events, true); foreach (var e in events) { - if (!IsNotifierEventVisibleTo(item, e)) continue; - if (ServiceResult.IsBad(ValidateEventRolePermissions(item, e))) continue; - item.QueueEvent(e); + if (ConditionOf(e) is not { } node) continue; + known.Add(node); + if (IsDirectRefreshEventAllowed(item, e, enforced)) allowed.Add((item, node)); } } + + var states = items.Select(EventItemStateOf).ToList(); + var perItem = new List<(IFilterTarget Event, NodeId ConditionId)>[items.Count]; + var generation = Interlocked.Increment(ref _refreshGeneration); + try + { + lock (Lock) + { + var watermark = Interlocked.Read(ref _conditionEventSequence); + long episode; + bool deferring; + lock (_eventReportGate) + { + episode = _deferralEpisode; + deferring = _deferredEventReports > 0; + } + + var retained = new HashSet(); + foreach (var condition in _alarmConditions.Values) + { + if (condition.Retain?.Value == true) retained.Add(condition.NodeId); + } + + for (var i = 0; i < items.Count; i++) + { + var item = items[i]; + var entries = new List<(IFilterTarget Event, NodeId ConditionId)>(); + perItem[i] = entries; + var state = states[i]; + + // Lock is held, so this is the attachment the walk below sees. An unattached item is not refreshed and + // gets no cut (it receives nothing until re-attached, and then every event is decided as it arrives). + var notifier = MonitoredItems.ContainsKey(item.Id) ? (item.ManagerHandle as NodeHandle)?.Node : null; + if (notifier is null) + { + lock (state.Gate) + { + state.PendingReplay = 0; + state.NewerSinceCut.Clear(); + } + + continue; + } + + // Frozen at the watermark: the replay is exactly the state every event at or below it describes. + var live = new List(); + notifier.ConditionRefresh(SystemContext, live, true); + var refreshed = new HashSet(); + foreach (var target in live) + { + if (target is not BaseInstanceState instance) continue; + // A condition materialised since the verdicts were taken is decided here (rare). + if (!allowed.Contains((item, instance))) + { + if (known.Contains(instance)) continue; + if (!IsDirectRefreshEventAllowed(item, target, enforced)) continue; + } + + var snapshot = new InstanceStateSnapshot(); + snapshot.Initialize(SystemContext, instance); + entries.Add((snapshot, instance.NodeId)); + refreshed.Add(instance.NodeId); + } + + lock (state.Gate) + { + state.Cut = deferring ? new RefreshCut(watermark, episode, refreshed, retained) : null; + state.PendingWatermark = watermark; + state.NewerSinceCut.Clear(); + state.PendingReplay = generation; + } + } + } + + DirectRefreshProbe?.Invoke(true, Monitor.IsEntered(Lock)); + QueueFrozenReplay(items, states, perItem, generation); + } + finally + { + ClearPendingReplays(states, generation); + } } /// The condition a refresh event describes, or null when it names none (withheld: fail closed). 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 18cc6d003..bba4a23eb 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/ConditionRefreshDeferralWireTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/ConditionRefreshDeferralWireTests.cs @@ -24,16 +24,18 @@ namespace ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests; /// 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. +/// The overload is produced deterministically: delivery is stalled at the EqA folder (), so the +/// condition's walk suspends there and its later events are deferred — without holding the node manager's Lock, +/// which the refresh needs. (Before Gitea #855 this held the folder item's SDK lock so the folder's 4096-slot channel +/// filled; folder items are now queued per item by the reporter, and no notifier of ours defers on its own.) The +/// folder-item twin of these tests is . /// 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; + // Every event after the stalled one is deferred; < 1024 pending, so nothing coalesces. + private const int FloodSize = 500; /// Per-folder / Server-object ACL on or off (off is the production default). /// @@ -84,7 +86,7 @@ public sealed class ConditionRefreshDeferralWireTests 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))) + using (var stall = NotifierStall.Hold(rig.Nm, EqA)) { 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"); @@ -200,7 +202,7 @@ public sealed class ConditionRefreshDeferralWireTests 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))) + using (var stall = NotifierStall.Hold(rig.Nm, EqA)) { for (var i = 0; i < FloodSize; i++) rig.Project(native: true, Active($"f{i}")); rig.Project(native: true, Active("final")); @@ -271,7 +273,7 @@ public sealed class ConditionRefreshDeferralWireTests rig.Sink.WriteAlarmCondition(Free, Active("x2"), DateTime.UtcNow, AddressSpaceRealm.Raw); }; - using (var stall = ItemLockStall.Hold(FindServerItem(rig.Nm, onA.Status.Id))) + using (var stall = NotifierStall.Hold(rig.Nm, EqA)) { 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"); @@ -345,7 +347,7 @@ public sealed class ConditionRefreshDeferralWireTests proceed.Wait(TimeSpan.FromSeconds(30)); }; - using (ItemLockStall.Hold(FindServerItem(rig.Nm, onA.Status.Id))) + using (NotifierStall.Hold(rig.Nm, EqA)) { for (var i = 0; i < FloodSize; i++) rig.Project(native: true, Active($"f{i}")); rig.Project(native: true, Active("final")); @@ -416,7 +418,6 @@ public sealed class ConditionRefreshDeferralWireTests 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; @@ -440,7 +441,7 @@ public sealed class ConditionRefreshDeferralWireTests { while (!stop.IsCancellationRequested) { - using (ItemLockStall.Hold(item)) + using (NotifierStall.Hold(rig.Nm, EqA)) { Thread.Sleep(400); if (rig.Nm.DeferredEventReports > 0) Interlocked.Increment(ref deferredSeen); @@ -568,7 +569,7 @@ public sealed class ConditionRefreshDeferralWireTests 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; + var stall = deferring ? NotifierStall.Hold(rig.Nm, EqA) : null; try { if (deferring) @@ -622,7 +623,7 @@ public sealed class ConditionRefreshDeferralWireTests } /// The server-side Server-object (all-events) item behind a client item. - private static IEventMonitoredItem FindAllEventsItem(OtOpcUaNodeManager nm, uint id) + internal static IEventMonitoredItem FindAllEventsItem(OtOpcUaNodeManager nm, uint id) { var map = (IDictionary)typeof(OtOpcUaNodeManager) .GetField("_allEventsItems", BindingFlags.Instance | BindingFlags.NonPublic)! @@ -636,7 +637,7 @@ public sealed class ConditionRefreshDeferralWireTests } /// 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) + internal static IEventMonitoredItem FindServerItem(OtOpcUaNodeManager nm, uint id) { var map = (IDictionary)typeof(OtOpcUaNodeManager) .GetField("_eventItems", BindingFlags.Instance | BindingFlags.NonPublic)! @@ -664,45 +665,4 @@ public sealed class ConditionRefreshDeferralWireTests 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(); - } - } } diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/EventQueueOverflowAclWireTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/EventQueueOverflowAclWireTests.cs index 1ca3b158a..81039208d 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/EventQueueOverflowAclWireTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/EventQueueOverflowAclWireTests.cs @@ -27,13 +27,16 @@ namespace ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests; /// no ACL-denied event reaches an item during the overflow, including deferred events whose grant was /// revoked while they waited. /// +/// Since Gitea #855 the folder items are served per item by the reporter (as the Server-object items always were), +/// so the flood no longer defers at all — the first test pins that — and the revocation test stalls the folder's +/// delivery () to get a deferred backlog. /// public sealed class EventQueueOverflowAclWireTests { private const string EqA = ServerEventRig.EqA; private const string EqB = ServerEventRig.EqB; - // Native events per flood: past the SDK's 4096-slot per-notifier channel, so the native condition defers. + // Native events per flood: past the SDK's 4096-slot per-notifier channel (which deferred them before #855). private const int PerCondition = 6000; /// @@ -150,13 +153,12 @@ public sealed class EventQueueOverflowAclWireTests readLatencies.Max().ShouldBeLessThan(TimeSpan.FromSeconds(5), "a Read must not queue behind the whole flood"); (await TimedReadAsync(anonymous, eqA, ct)).ShouldBeLessThan(TimeSpan.FromSeconds(2), "the node manager is free afterwards"); - // Memory: the node manager's own overflow store stayed within its per-condition budget (queued + in flight)… - Volatile.Read(ref peakDeferred).ShouldBeLessThanOrEqualTo( - 2 * (OtOpcUaNodeManager.MaxDeferredEventsPerCondition + 1), "deferral is bounded per condition"); - if (singleHold) rig.Nm.DiscardedEventReports.ShouldBeGreaterThan(0, "the native tail overflowed and coalesced"); + // Memory: the node manager held nothing of its own for the overflow (#855: no notifier of ours defers, so the + // item queues are the only store, bounded below)… + Volatile.Read(ref peakDeferred).ShouldBe(0, "folder items are served per item by the reporter: nothing defers"); + rig.Nm.DiscardedEventReports.ShouldBe(0); - // …let the SDK channel finish draining into the (still unpublished) item queues, then publish. - await Task.Delay(TimeSpan.FromSeconds(2), ct); + // …then publish the (still unpublished) item queues. await keySub.SetPublishingModeAsync(true, ct); var lastNative = $"n{PerCondition - 1}"; await RecordedEvents.WaitUntilAsync(() => keyEvents.Has(onB, lastNative) && keyEvents.Has(keyServer, lastNative)); @@ -165,10 +167,9 @@ public sealed class EventQueueOverflowAclWireTests foreach (var item in new[] { onB, keyServer }) { var received = keyEvents.Of(item); - // ACL: not one event of the denied (EqA-owned) scripted alarm, overflow or not. (Under a single hold the - // deferred native tail drains last and evicts every synchronously-queued event from the 1000-slot queue, so - // there this check cannot see a leak; the "after-s" leg below is the falsifier for that mode — verified by - // running with enforcement off: chunked fails here, single-hold fails at "after-s".) + // ACL: not one event of the denied (EqA-owned) scripted alarm, overflow or not. (The 1000-slot queue keeps + // only the last events of the flood, which interleave n/s, so a leak shows here in both modes; the + // "after-s" leg below is a second falsifier.) received.Where(f => RecordedEvents.Message(f)?.StartsWith('s') == true) .ShouldBeEmpty($"{item.DisplayName}: an ACL-denied event leaked during the overflow"); @@ -211,7 +212,7 @@ public sealed class EventQueueOverflowAclWireTests } /// - /// A grant revoked while a deferred backlog waits behind a full notifier channel: the deferred events are decided + /// A grant revoked while a deferred backlog waits behind a stalled notifier: the deferred events are decided /// at delivery against the CURRENT generation, so none of them — nor any event reported after the revocation — /// reaches the revoked session's folder or Server-object item, while a still-granted session sharing the same /// folder channel receives the whole tail (positive control). @@ -249,6 +250,9 @@ public sealed class EventQueueOverflowAclWireTests await anonSub.ApplyChangesAsync(ct); anonA.Status.Error.ShouldBeNull(); + // The EqA folder's delivery stalls (#855: nothing of ours defers on its own any more), so the native condition's + // walk suspends there and the rest of the flood is deferred behind it. + using var stall = NotifierStall.Hold(rig.Nm, EqA); var flood = Task.Factory.StartNew(() => { lock (rig.Nm.Lock) @@ -276,11 +280,12 @@ public sealed class EventQueueOverflowAclWireTests throw new ShouldAssertException("revoking a grant under Lock while events were deferred stalled (deadlock)"); } + stall.Release(); await RecordedEvents.WaitUntilAsync(() => rig.Nm.DeferredEventReports == 0); await Task.Delay(TimeSpan.FromSeconds(2), ct); await keySub.SetPublishingModeAsync(true, ct); - // Positive control: the still-granted session on the SAME folder channel gets the whole deferred tail. + // Positive control: the still-granted session on the SAME folder gets the whole deferred tail. var lastNative = $"n{PerCondition - 1}"; await RecordedEvents.WaitUntilAsync(() => anonEvents.Has(anonA, lastNative) && anonEvents.Has(anonA, "after-revoke") diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/FolderConditionRefreshWireTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/FolderConditionRefreshWireTests.cs new file mode 100644 index 000000000..b01a93427 --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/FolderConditionRefreshWireTests.cs @@ -0,0 +1,627 @@ +using System.Diagnostics; +using System.Globalization; +using Opc.Ua; +using Opc.Ua.Client; +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.Abstractions; +using ZB.MOM.WW.OtOpcUa.Core.Authorization; +using MonitoredItem = Opc.Ua.Client.MonitoredItem; + +namespace ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests; + +/// +/// Gitea #855: a ConditionRefresh of an event item on an equipment / raw device FOLDER gets the guarantee the +/// Server object got in #854 — no condition event older than what the refresh replayed reaches the item after +/// RefreshEnd, per-condition order is kept, the per-folder ACL (#826) decides every event and every replayed condition, +/// and neither a reporter nor the refresh ever waits on a stalled delivery. +/// +/// +/// +/// Folder items are now served per item by the node manager (no SDK channel), so in production no notifier defers +/// at all; pins that under a +/// flood with concurrent refreshes. Every other test produces the deferral the cut exists for by stalling delivery +/// at a notifier (), the way a full SDK channel used to. +/// +/// +/// The rig's native condition is wired into EqA and EqB and hangs under the raw device folder, so its walk is +/// condition → EqA → … → device folder; a stall at EqA suspends it before EqA's items and every later hop. +/// +/// +public sealed class FolderConditionRefreshWireTests +{ + private const string EqA = ServerEventRig.EqA; + + // Every event after the stalled one is deferred; < 1024 pending, so nothing coalesces. + private const int FloodSize = 500; + + /// Per-folder ACL on or off (off is the production default). + /// Whether the condition is still Retain at refresh time (re-sent) or cleared (omitted). + /// ConditionRefresh2 on the folder item instead of ConditionRefresh on its subscription. + /// + /// The folder item is modified (ModifyMonitoredItems re-subscribes it to its notifier) between the refresh and the + /// drain: its cut must survive, as the Server object's does across an unsubscribe + resubscribe. + /// + [Theory] + [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_folder_item_after_RefreshEnd( + bool enforced, bool finalRetained, bool refresh2, bool modify) + { + var ct = TestContext.Current.CancellationToken; + await using var rig = await ServerEventRig.StartAsync("Refresh855", encrypted: false, ct); + rig.KeyRoles = []; + rig.MaterialiseNative(); + rig.Enforce(enforced, [ServerEventRig.Key()], KeyEquipmentGrant(EqA)); + + // A second, non-refreshing subscriber on the SAME folder: it must keep the whole deferred tail. + using var otherSession = await rig.ConnectAsync(anonymous: false, ct); + var otherEvents = new RecordedEvents(); + var otherSubscription = await RecordedEvents.SubscribeAsync(otherSession, otherEvents, ct); + var otherOnA = RecordedEvents.AddEventItem(otherSubscription, ServerEventRig.UnsNodeId(otherSession, EqA)); + await otherSubscription.ApplyChangesAsync(ct); + otherOnA.Status.Error.ShouldBeNull(); + + using var session = await rig.ConnectAsync(anonymous: false, ct); + var events = new RecordedEvents(); + var subscription = await RecordedEvents.SubscribeAsync(session, events, ct); + var onA = RecordedEvents.AddEventItem(subscription, ServerEventRig.UnsNodeId(session, EqA)); + await subscription.ApplyChangesAsync(ct); + onA.Status.Error.ShouldBeNull(); + + rig.Project(native: true, Active("seed")); + await RecordedEvents.WaitUntilAsync(() => events.Has(onA, "seed") && otherEvents.Has(otherOnA, "seed")); + + using (var stall = NotifierStall.Hold(rig.Nm, EqA)) + { + for (var i = 0; i < FloodSize; i++) rig.Project(native: true, Active($"f{i}")); + rig.Nm.DeferredEventReports.ShouldBeGreaterThan(0, "the stall must actually have deferred events"); + rig.Project(native: true, finalRetained ? Active("final") : Cleared("final")); + + var refresh = Stopwatch.StartNew(); + await RefreshAsync(session, subscription, onA, refresh2, ct); + refresh.Elapsed.ShouldBeLessThan(TimeSpan.FromSeconds(10), "ConditionRefresh must not wait for the stalled delivery"); + await RecordedEvents.WaitUntilAsync(() => events.HasType(onA, ObjectTypeIds.RefreshEndEventType)); + rig.Nm.DeferredEventReports.ShouldBeGreaterThan(0, "the deferred events were still waiting at RefreshEnd"); + rig.Nm.ActiveRefreshCuts.ShouldBe(1, "only the refreshed folder item holds a cut"); + + if (modify) + { + onA.QueueSize = 1001; + await subscription.ApplyChangesAsync(ct); + onA.Status.Error.ShouldBeNull(); + } + + stall.Release(); + } + + await RecordedEvents.WaitUntilAsync(() => rig.Nm.DeferredEventReports == 0); + await RecordedEvents.WaitUntilAsync(() => rig.Nm.ActiveRefreshCuts == 0); + rig.Project(native: true, Active("after")); + await RecordedEvents.WaitUntilAsync(() => events.Has(onA, "after") && otherEvents.Has(otherOnA, "after")); + await Task.Delay(TimeSpan.FromMilliseconds(300), ct); + + var (replayed, afterEnd) = Window(events.Of(onA)); + 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"); + afterEnd.ShouldBe(["final", "after"], "stale Retain=true events would re-add the alarm; the Retain=false one agrees"); + events.Of(onA).Last(f => RecordedEvents.Message(f) == "final")[RecordedEvents.RetainField].GetBoolean().ShouldBeFalse(); + } + + rig.Nm.RefreshSupersededDeliveries.ShouldBeGreaterThan(0); + + // The other subscriber on the same folder did not refresh: it keeps every deferred event, in order. + var flood = Indices(otherEvents.Of(otherOnA), 'f'); + flood.ShouldBe(Enumerable.Range(0, FloodSize).ToList(), "the non-refreshing folder subscriber keeps the whole tail, in order"); + otherEvents.Has(otherOnA, "final").ShouldBeTrue(); + } + + /// + /// The race the item gate closes, for a folder item: a deferred event is between its supersession check and its + /// queueing to the folder item when the refresh arrives. The refresh must wait for that one delivery, so it lands + /// before the replay rather than after RefreshEnd. + /// + [Fact] + public async Task A_folder_refresh_waits_for_a_deferred_delivery_already_past_its_check() + { + var ct = TestContext.Current.CancellationToken; + await using var rig = await ServerEventRig.StartAsync("Refresh855Gate", encrypted: false, ct); + rig.KeyRoles = []; + rig.MaterialiseNative(); + rig.Enforce(false, [ServerEventRig.Key()], KeyEquipmentGrant(EqA)); + + using var session = await rig.ConnectAsync(anonymous: false, ct); + var events = new RecordedEvents(); + var subscription = await RecordedEvents.SubscribeAsync(session, events, ct); + var onA = RecordedEvents.AddEventItem(subscription, ServerEventRig.UnsNodeId(session, EqA)); + await subscription.ApplyChangesAsync(ct); + + rig.Project(native: true, Active("seed")); + await RecordedEvents.WaitUntilAsync(() => events.Has(onA, "seed")); + + using var entered = new ManualResetEventSlim(); + using var proceed = new ManualResetEventSlim(); + var probed = 0; + rig.Nm.DeferredFolderDeliveryProbe = () => + { + if (Interlocked.Increment(ref probed) != 1) return; + entered.Set(); + proceed.Wait(TimeSpan.FromSeconds(30)); + }; + + using (NotifierStall.Hold(rig.Nm, EqA)) + { + for (var i = 0; i < FloodSize; i++) rig.Project(native: true, Active($"f{i}")); + rig.Project(native: true, Active("final")); + } + + // The released walk parks on its first deferred delivery to the folder item, past the check, inside the gate. + entered.Wait(TimeSpan.FromSeconds(10), ct).ShouldBeTrue("the drain must reach the folder item"); + var refresh = session.CallAsync( + ObjectTypeIds.ConditionType, MethodIds.ConditionType_ConditionRefresh, ct, subscription.Id); + await Task.Delay(TimeSpan.FromSeconds(1), ct); + events.HasType(onA, 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(() => events.HasType(onA, ObjectTypeIds.RefreshEndEventType)); + rig.Nm.DeferredFolderDeliveryProbe = null; + + await RecordedEvents.WaitUntilAsync(() => rig.Nm.DeferredEventReports == 0); + rig.Project(native: true, Active("after")); + await RecordedEvents.WaitUntilAsync(() => events.Has(onA, "after")); + await Task.Delay(TimeSpan.FromMilliseconds(300), ct); + + var (window, afterEnd) = Window(events.Of(onA)); + window.Count.ShouldBe(2, "the parked delivery, then the replay"); + window[0]!.ShouldStartWith("f"); + window[1].ShouldBe("final"); + afterEnd.ShouldBe(["after"]); + } + + /// + /// The cut path queues its frozen replay OFF Lock. An event newer than the cut that reaches the folder item + /// in that window (a condition delivered synchronously, 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_folder_cut_is_never_followed_by_its_older_frozen_replay() + { + var ct = TestContext.Current.CancellationToken; + await using var rig = await ServerEventRig.StartAsync("Refresh855Pending", encrypted: false, ct); + rig.KeyRoles = []; + rig.MaterialiseNative(); + rig.Enforce(false, [ServerEventRig.Key()], KeyEquipmentGrant(EqA)); + // A second condition under the device folder only: its walk never reaches the stalled EqA, so it stays 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); + + // EqA needs an event item of its own to be a notifier fan-out the stall can apply to. + using var eqASession = await rig.ConnectAsync(anonymous: false, ct); + var eqAEvents = new RecordedEvents(); + var eqASubscription = await RecordedEvents.SubscribeAsync(eqASession, eqAEvents, ct); + RecordedEvents.AddEventItem(eqASubscription, ServerEventRig.UnsNodeId(eqASession, EqA)); + await eqASubscription.ApplyChangesAsync(ct); + + // The refreshed item is on the raw device folder, which both conditions report through. + using var session = await rig.ConnectAsync(anonymous: false, ct); + var events = new RecordedEvents(); + var subscription = await RecordedEvents.SubscribeAsync(session, events, ct); + var onDevice = RecordedEvents.AddEventItem(subscription, RawNodeId(session, ServerEventRig.RawDeviceFolder)); + await subscription.ApplyChangesAsync(ct); + onDevice.Status.Error.ShouldBeNull(); + + rig.Project(native: true, Active("seed")); + await RecordedEvents.WaitUntilAsync(() => events.Has(onDevice, "seed")); + + var injected = 0; + rig.Nm.DirectRefreshProbe = (cut, _) => + { + if (cut && Interlocked.Exchange(ref injected, 1) == 0) + rig.Sink.WriteAlarmCondition(Free, Active("x2"), DateTime.UtcNow, AddressSpaceRealm.Raw); + }; + + using (var stall = NotifierStall.Hold(rig.Nm, EqA)) + { + 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 session.CallAsync(ObjectTypeIds.ConditionType, MethodIds.ConditionType_ConditionRefresh, ct, subscription.Id); + await RecordedEvents.WaitUntilAsync(() => events.HasType(onDevice, ObjectTypeIds.RefreshEndEventType)); + stall.Release(); + } + + rig.Nm.DirectRefreshProbe = null; + Volatile.Read(ref injected).ShouldBe(1, "the folder refresh took the cut path"); + var stream = events.Of(onDevice).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"); + } + + /// + /// ACL on: a condition the folder refresh omitted because the item may not see it there (its reference into EqA was + /// dropped by the generation) is not withheld afterwards — the reference returns before the drain, and the item must + /// end on the condition's newest state. And while omitted, the replay itself carried nothing of it. + /// + [Fact] + public async Task Deferred_events_of_a_condition_the_folder_refresh_omitted_by_the_acl_are_not_withheld() + { + var ct = TestContext.Current.CancellationToken; + await using var rig = await ServerEventRig.StartAsync("Refresh855Acl", encrypted: false, ct); + rig.KeyRoles = []; + rig.MaterialiseNative(); + NodeAcl[] grants = [KeyEquipmentGrant(EqA)]; + var state = rig.Enforce(true, [ServerEventRig.Key()], grants); + Deploy(state, "rev-2", grants, ServerEventRig.EqB); // the native alarm is no longer projected into EqA + + using var session = await rig.ConnectAsync(anonymous: false, ct); + var events = new RecordedEvents(); + var subscription = await RecordedEvents.SubscribeAsync(session, events, ct); + var onA = RecordedEvents.AddEventItem(subscription, ServerEventRig.UnsNodeId(session, EqA)); + await subscription.ApplyChangesAsync(ct); + onA.Status.Error.ShouldBeNull(); + + rig.Project(native: true, Active("seed")); + await Task.Delay(TimeSpan.FromMilliseconds(500), ct); + events.Has(onA, "seed").ShouldBeFalse("the generation does not project the alarm into EqA"); + + using (var stall = NotifierStall.Hold(rig.Nm, EqA)) + { + 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 session.CallAsync(ObjectTypeIds.ConditionType, MethodIds.ConditionType_ConditionRefresh, ct, subscription.Id); + await RecordedEvents.WaitUntilAsync(() => events.HasType(onA, ObjectTypeIds.RefreshEndEventType)); + + // The reference returns while the condition's events still wait. + Deploy(state, "rev-3", grants, EqA, ServerEventRig.EqB); + stall.Release(); + } + + await RecordedEvents.WaitUntilAsync(() => rig.Nm.DeferredEventReports == 0); + await RecordedEvents.WaitUntilAsync(() => events.Has(onA, "final")); + var (replayed, after) = Window(events.Of(onA)); + replayed.ShouldBeEmpty("the refresh omitted the condition: not visible through EqA at the time"); + 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')).ShouldBeGreaterThan(0, "its deferred events were delivered, not withheld"); + } + + /// + /// ACL on: a grant revoked while a refreshed folder item's deferred events wait. Neither the replay (taken while + /// granted) nor anything after it leaks once revoked; a still-granted subscriber on the same folder gets the tail. + /// + [Fact] + public async Task A_grant_revoked_after_a_folder_refresh_withholds_every_deferred_event() + { + var ct = TestContext.Current.CancellationToken; + await using var rig = await ServerEventRig.StartAsync("Refresh855Revoke", encrypted: false, ct); + rig.KeyRoles = []; + rig.MaterialiseNative(); + var anonGrant = ServerEventRig.EquipmentGrant(EqA, NodePermissions.ReadOnly); + var state = rig.Enforce(true, [ServerEventRig.Key()], KeyEquipmentGrant(EqA), anonGrant); + + using var keyed = await rig.ConnectAsync(anonymous: false, ct); + var keyEvents = new RecordedEvents(); + var keySub = await RecordedEvents.SubscribeAsync(keyed, keyEvents, ct); + var keyA = RecordedEvents.AddEventItem(keySub, ServerEventRig.UnsNodeId(keyed, EqA)); + await keySub.ApplyChangesAsync(ct); + + using var anonymous = await rig.ConnectAsync(anonymous: true, ct); + var anonEvents = new RecordedEvents(); + var anonSub = await RecordedEvents.SubscribeAsync(anonymous, anonEvents, ct); + var anonA = RecordedEvents.AddEventItem(anonSub, ServerEventRig.UnsNodeId(anonymous, EqA)); + await anonSub.ApplyChangesAsync(ct); + + rig.Project(native: true, Active("seed")); + await RecordedEvents.WaitUntilAsync(() => keyEvents.Has(keyA, "seed") && anonEvents.Has(anonA, "seed")); + + using (var stall = NotifierStall.Hold(rig.Nm, EqA)) + { + for (var i = 0; i < FloodSize; i++) rig.Project(native: true, Active($"f{i}")); + await keyed.CallAsync(ObjectTypeIds.ConditionType, MethodIds.ConditionType_ConditionRefresh, ct, keySub.Id); + await RecordedEvents.WaitUntilAsync(() => keyEvents.HasType(keyA, ObjectTypeIds.RefreshEndEventType)); + + state.Apply( + ServerEventRig.Cluster, "rev-2", [ServerEventRig.Key()], [anonGrant], ServerEventRig.ScopePaths(), + ServerEventRig.EquipmentOwnerPaths(), ServerEventRig.ScriptedAlarmOwners(), ServerEventRig.FolderScopes(), + ServerEventRig.TagReferences()); + rig.Project(native: true, Active("after-revoke")); + stall.Release(); + } + + await RecordedEvents.WaitUntilAsync(() => rig.Nm.DeferredEventReports == 0); + await RecordedEvents.WaitUntilAsync(() => + anonEvents.Has(anonA, $"f{FloodSize - 1}") && anonEvents.Has(anonA, "after-revoke")); + await Task.Delay(TimeSpan.FromSeconds(1), ct); + + var (_, after) = Window(keyEvents.Of(keyA)); + after.ShouldBeEmpty("after RefreshEnd the revoked session gets nothing: older events are superseded, later ones denied"); + keyEvents.Has(keyA, "after-revoke").ShouldBeFalse(); + } + + /// + /// Production shape, no stall: bursts reported under one Lock hold (an apply pass) and single reports, against + /// folder items that refresh repeatedly while their client is slow to publish. Folder items are served per item, so + /// nothing is ever deferred; each refresh's replay is never older than an event the item already received, nothing + /// after RefreshEnd goes backwards, the stream converges on the newest state, and no replay is queued under Lock. + /// + /// Per-folder ACL on or off. + [Theory] + [InlineData(false)] + [InlineData(true)] + public async Task Production_overload_never_defers_and_never_overtakes_a_folder_refresh(bool enforced) + { + var ct = TestContext.Current.CancellationToken; + await using var rig = await ServerEventRig.StartAsync("Refresh855Load", encrypted: false, ct); + rig.KeyRoles = []; + rig.MaterialiseNative(); + rig.Enforce(enforced, [ServerEventRig.Key()], KeyEquipmentGrant(EqA)); + + using var session = await rig.ConnectAsync(anonymous: false, ct); + var events = new RecordedEvents(); + var subscription = await RecordedEvents.SubscribeAsync(session, events, ct); + var onA = RecordedEvents.AddEventItem(subscription, ServerEventRig.UnsNodeId(session, EqA)); + onA.QueueSize = 100_000; // keep every event so the ordering check sees the whole stream + await subscription.ApplyChangesAsync(ct); + + rig.Project(native: true, Active("s0")); + await RecordedEvents.WaitUntilAsync(() => events.Has(onA, "s0")); + + var probes = new System.Collections.Concurrent.ConcurrentQueue<(bool Cut, bool LockHeld)>(); + rig.Nm.DirectRefreshProbe = (cut, held) => probes.Enqueue((cut, held)); + + using var stop = CancellationTokenSource.CreateLinkedTokenSource(ct); + var counter = 0; + var peakDeferred = 0; + var reporter = Task.Factory.StartNew(() => + { + while (!stop.IsCancellationRequested) + { + lock (rig.Nm.Lock) + { + for (var i = 0; i < 1000; i++) rig.Project(native: true, Active($"s{Interlocked.Increment(ref counter)}")); + } + + peakDeferred = Math.Max(peakDeferred, rig.Nm.DeferredEventReports); + rig.Project(native: true, Active($"s{Interlocked.Increment(ref counter)}")); + Thread.Sleep(2); + } + }, stop.Token, TaskCreationOptions.LongRunning, TaskScheduler.Default); + + var refusedAsInProgress = 0; + for (var r = 0; r < 30; r++) + { + try + { + await RefreshAsync(session, subscription, onA, refresh2: r % 3 == 2, ct).WaitAsync(TimeSpan.FromSeconds(20), ct); + } + catch (ServiceResultException ex) when (ex.StatusCode == StatusCodes.BadRefreshInProgress) + { + refusedAsInProgress++; + } + + await Task.Delay(TimeSpan.FromMilliseconds(80), ct); + } + + await stop.CancelAsync(); + await reporter.WaitAsync(TimeSpan.FromSeconds(30), CancellationToken.None); + rig.Nm.DirectRefreshProbe = null; + + peakDeferred.ShouldBe(0, "no notifier of ours defers: folder items are served per item by the reporter"); + rig.Nm.DiscardedEventReports.ShouldBe(0); + var newest = $"s{Volatile.Read(ref counter)}"; + await RecordedEvents.WaitUntilAsync(() => events.Has(onA, newest)); + await Task.Delay(TimeSpan.FromMilliseconds(500), ct); + + CheckStream(events.Of(onA)).ShouldBeEmpty(); + refusedAsInProgress.ShouldBeLessThan(30, "most refreshes must actually have run"); + probes.ShouldNotBeEmpty(); + probes.ShouldAllBe(p => !p.Cut && !p.LockHeld); + RecordedEvents.Message(events.Of(onA).Last(f => f[RecordedEvents.AckedField].TryGetValue(out bool _))) + .ShouldBe(newest, "the stream converges on the newest state"); + } + + /// + /// Race / stress with deferral: reports (bursts under Lock and single), repeated refreshes of the folder item + /// and an intermittent stall of its folder's delivery all run at once. Outside a refresh the folder item's stream + /// never goes backwards, each refresh ends on the newest state the item has seen, and it converges. Nothing + /// deadlocks, and neither refresh path queues its replay under Lock. + /// + /// Per-folder ACL on or off. + [Theory] + [InlineData(false)] + [InlineData(true)] + public async Task Concurrent_reports_folder_refreshes_and_stalls_never_deliver_a_pre_refresh_state_after_RefreshEnd(bool enforced) + { + var ct = TestContext.Current.CancellationToken; + await using var rig = await ServerEventRig.StartAsync("Refresh855Stress", encrypted: false, ct); + rig.KeyRoles = []; + rig.MaterialiseNative(); + rig.Enforce(enforced, [ServerEventRig.Key()], KeyEquipmentGrant(EqA)); + + using var session = await rig.ConnectAsync(anonymous: false, ct); + var events = new RecordedEvents(); + var subscription = await RecordedEvents.SubscribeAsync(session, events, ct); + var onA = RecordedEvents.AddEventItem(subscription, ServerEventRig.UnsNodeId(session, EqA)); + onA.QueueSize = 100_000; + await subscription.ApplyChangesAsync(ct); + + rig.Project(native: true, Active("s0")); + await RecordedEvents.WaitUntilAsync(() => events.Has(onA, "s0")); + + var probes = new System.Collections.Concurrent.ConcurrentQueue<(bool Cut, bool LockHeld)>(); + rig.Nm.DirectRefreshProbe = (cut, held) => probes.Enqueue((cut, held)); + + using var stop = CancellationTokenSource.CreateLinkedTokenSource(ct); + var counter = 0; + var reporter = Task.Factory.StartNew(() => + { + while (!stop.IsCancellationRequested) + { + lock (rig.Nm.Lock) + { + for (var i = 0; i < 300; 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 (NotifierStall.Hold(rig.Nm, EqA)) + { + Thread.Sleep(300); + if (rig.Nm.DeferredEventReports > 0) Interlocked.Increment(ref deferredSeen); + } + + Thread.Sleep(100); + } + }, stop.Token, TaskCreationOptions.LongRunning, TaskScheduler.Default); + + var refusedAsInProgress = 0; + var stressDeadline = DateTime.UtcNow + TimeSpan.FromSeconds(90); + for (var r = 0; r < 40 || (Volatile.Read(ref deferredSeen) <= 1 && DateTime.UtcNow < stressDeadline); r++) + { + try + { + await RefreshAsync(session, subscription, onA, refresh2: r % 3 == 2, ct).WaitAsync(TimeSpan.FromSeconds(20), ct); + } + catch (ServiceResultException ex) when (ex.StatusCode == StatusCodes.BadRefreshInProgress) + { + refusedAsInProgress++; + } + + 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); + rig.Nm.DirectRefreshProbe = null; + Volatile.Read(ref deferredSeen).ShouldBeGreaterThan(1, "the stress must actually have run with deferred events"); + + await RecordedEvents.WaitUntilAsync(() => rig.Nm.DeferredEventReports == 0); + var newest = $"s{Volatile.Read(ref counter)}"; + await RecordedEvents.WaitUntilAsync(() => events.Has(onA, newest)); + await Task.Delay(TimeSpan.FromMilliseconds(500), ct); + + CheckStream(events.Of(onA)).ShouldBeEmpty(); + refusedAsInProgress.ShouldBeLessThan(40, "most refreshes must actually have run"); + rig.Nm.RefreshSupersededDeliveries.ShouldBeGreaterThan(0, "some folder refresh must actually have overtaken deferred events"); + probes.ShouldContain(p => p.Cut, "some folder refresh must have taken the cut path"); + probes.ShouldAllBe(p => !p.LockHeld); + RecordedEvents.Message(events.Of(onA).Last(f => f[RecordedEvents.AckedField].TryGetValue(out bool _))) + .ShouldBe(newest, "the stream converges on the newest state"); + } + + /// + /// Walks one condition's stream: 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 follows). + /// + private static List CheckStream(IReadOnlyList> stream) + { + var max = -1; + var last = -1; + var replaying = false; + var violations = new List(); + foreach (var f in stream) + { + var type = f[RecordedEvents.EventTypeField].GetNodeId(); + if (type == ObjectTypeIds.RefreshStartEventType) + { + replaying = true; + continue; + } + + if (type == ObjectTypeIds.RefreshEndEventType) + { + replaying = false; + 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..], CultureInfo.InvariantCulture); + if (!replaying && n < max) violations.Add($"s{n} delivered after s{max}"); + max = Math.Max(max, n); + last = n; + } + + return violations; + } + + private static Task RefreshAsync(ISession session, Subscription subscription, MonitoredItem item, bool refresh2, CancellationToken ct) => + refresh2 + ? session.CallAsync(ObjectTypeIds.ConditionType, MethodIds.ConditionType_ConditionRefresh2, ct, subscription.Id, item.Status.Id) + : session.CallAsync(ObjectTypeIds.ConditionType, MethodIds.ConditionType_ConditionRefresh, ct, subscription.Id); + + /// The condition-event messages inside the item's LAST RefreshStart..RefreshEnd, and those after it. + private static (List Replayed, List AfterEnd) Window(IReadOnlyList> stream) + { + var list = stream.ToList(); + var start = list.FindLastIndex(f => f[RecordedEvents.EventTypeField].GetNodeId() == ObjectTypeIds.RefreshStartEventType); + var end = list.FindLastIndex(f => f[RecordedEvents.EventTypeField].GetNodeId() == ObjectTypeIds.RefreshEndEventType); + start.ShouldBeGreaterThanOrEqualTo(0); + end.ShouldBeGreaterThan(start); + static bool IsCondition(ArrayOf f) => f[RecordedEvents.AckedField].TryGetValue(out bool _); + return ( + list.Skip(start + 1).Take(end - start - 1).Where(IsCondition).Select(RecordedEvents.Message).ToList(), + list.Skip(end + 1).Where(IsCondition).Select(RecordedEvents.Message).ToList()); + } + + private static List Indices(IReadOnlyList> received, char prefix) => + received + .Select(RecordedEvents.Message) + .Where(m => m is { Length: > 1 } && m[0] == prefix && char.IsAsciiDigit(m[1])) + .Select(m => int.Parse(m![1..], CultureInfo.InvariantCulture)) + .ToList(); + + private static NodeId RawNodeId(ISession session, string id) => + new(id, (ushort)session.NamespaceUris.GetIndex(V3NodeIds.RawNamespaceUri)); + + private static void Deploy(DataPlaneAuthorityState state, string revision, NodeAcl[] grants, params string[] referencing) => + state.Apply( + ServerEventRig.Cluster, revision, [ServerEventRig.Key()], grants, ServerEventRig.ScopePaths(), + ServerEventRig.EquipmentOwnerPaths(), ServerEventRig.ScriptedAlarmOwners(), ServerEventRig.FolderScopes(), + new Dictionary> { [ServerEventRig.RawAlarmPath] = referencing.ToHashSet() }); + + 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, + }; +} diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/FolderEventFloodWireTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/FolderEventFloodWireTests.cs index d99ac83e6..97f2271de 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/FolderEventFloodWireTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/FolderEventFloodWireTests.cs @@ -19,6 +19,13 @@ namespace ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests; /// consumer — a deadlock that also freezes every other operation needing Lock. Publishing is switched off /// on the client so nothing downstream drains either (the slow / absent consumer case). /// +/// +/// Since Gitea #855 condition events no longer go through that channel at all: the node manager queues them to a +/// folder's items itself, per item, on the reporter's thread (as it always did for Server-object items), so the same +/// flood now completes without deferring anything. The deadlock test therefore also pins that per-item delivery is in +/// force (it would defer again if the bypass stopped recognising the SDK's fan-out); the ghost test produces its +/// deferral with . +/// public sealed class FolderEventFloodWireTests { private const string EqA = ServerEventRig.EqA; @@ -56,8 +63,12 @@ public sealed class FolderEventFloodWireTests // No publishing: the server-side item queue (QueueSize 1000) is the only sink, as with a stalled client. await subscription.SetPublishingModeAsync(false, ct); + // #855: the folder's SDK fan-out is the one the node manager recognises and serves per item itself. + rig.Nm.TryGetFolder(EqA)!.OnReportEventAsync?.Target.ShouldBeOfType(); + // The whole burst under ONE hold of the node manager's Lock — the shape of any multi-event apply/projection // pass — starting from an uncached verdict (the item is new), so the delivery task needs Lock for event 1. + var peakDeferred = 0; var flood = Task.Factory.StartNew(() => { lock (rig.Nm.Lock) @@ -65,6 +76,7 @@ public sealed class FolderEventFloodWireTests for (var i = 0; i < FloodSize; i++) { rig.Project(native: true, State($"f{i}")); + peakDeferred = Math.Max(peakDeferred, rig.Nm.DeferredEventReports); Volatile.Write(ref reported, i + 1); } } @@ -83,11 +95,9 @@ public sealed class FolderEventFloodWireTests + "on Lock (deadlock)"); } - // The overflow was deferred rather than blocking, and drains once Lock is free. - await RecordedEvents.WaitUntilAsync(() => rig.Nm.DeferredEventReports == 0); - // …and the SDK's channel (up to 4096 accepted events) finishes draining into the item queue, still with - // publishing off, so the count below measures the item queue and not events in transit. - await Task.Delay(TimeSpan.FromSeconds(2), ct); + // #855: nothing was deferred at all — every event was queued to the folder item by the reporter itself. + peakDeferred.ShouldBe(0, "no notifier of ours defers: folder items are served per item, not through the SDK channel"); + rig.Nm.DeferredEventReports.ShouldBe(0); // Delivery is bounded by the item's queue (QueueSize 1000, oldest discarded), not by the flood, and the // events that survive arrive in report order, ending with the last one reported. @@ -104,8 +114,7 @@ public sealed class FolderEventFloodWireTests flooded.Count.ShouldBeLessThanOrEqualTo((int)onA.Status.QueueSize, "delivery is bounded by the item's queue, not the flood"); flooded.ShouldBe(flooded.Order().ToList(), "events leave in the order they were reported"); flooded[^1].ShouldBe(FloodSize - 1); - rig.Nm.DiscardedEventReports.ShouldBeGreaterThan(0, - "past the per-condition budget the oldest waiting events give way to the newest (coalesced)"); + rig.Nm.DiscardedEventReports.ShouldBe(0, "nothing was deferred, so nothing was coalesced (the item queue is the bound)"); // And the node manager is live: a later event is decided by the per-folder ACL and delivered. rig.Project(native: true, State("after")); @@ -127,7 +136,8 @@ public sealed class FolderEventFloodWireTests /// #838 review — events deferred behind a full channel must not outlive their condition. A deploy that removes /// the condition while its events wait used to let them reach Server-object (i=2253) subscribers after the /// NodeDeleted — a ghost Retain alarm for ScadaBridge — and the folder subscribers too. Every hop of a deferred - /// report now re-checks that the condition is still the live instance. + /// report now re-checks that the condition is still the live instance (#855: also after a stall at a hop, before + /// the folder's items are served). /// /// Per-folder / Server-object ACL on or off (off is the production default). [Theory] @@ -151,16 +161,20 @@ public sealed class FolderEventFloodWireTests onServer.Status.Error.ShouldBeNull(); await subscription.SetPublishingModeAsync(false, ct); - // One Lock hold: flood past the channel (the tail is deferred), then the deploy removes the condition. - await Task.Factory.StartNew(() => + // One Lock hold, delivery stalled at the folder: flood (the tail is deferred), then the deploy removes the condition. + using (var stall = NotifierStall.Hold(rig.Nm, EqA)) { - lock (rig.Nm.Lock) + await Task.Factory.StartNew(() => { - for (var i = 0; i < FloodSize; i++) rig.Project(native: true, State($"f{i}")); - rig.Nm.DeferredEventReports.ShouldBeGreaterThan(0, "the flood must actually have deferred events"); - rig.Nm.RemoveAlarmConditionNode(ServerEventRig.RawAlarmPath, AddressSpaceRealm.Raw).ShouldBeTrue(); - } - }, ct, TaskCreationOptions.LongRunning, TaskScheduler.Default).WaitAsync(TimeSpan.FromSeconds(60), ct); + lock (rig.Nm.Lock) + { + for (var i = 0; i < FloodSize; i++) rig.Project(native: true, State($"f{i}")); + rig.Nm.DeferredEventReports.ShouldBeGreaterThan(0, "the flood must actually have deferred events"); + rig.Nm.RemoveAlarmConditionNode(ServerEventRig.RawAlarmPath, AddressSpaceRealm.Raw).ShouldBeTrue(); + } + }, ct, TaskCreationOptions.LongRunning, TaskScheduler.Default).WaitAsync(TimeSpan.FromSeconds(60), ct); + stall.Release(); + } await RecordedEvents.WaitUntilAsync(() => rig.Nm.DeferredEventReports == 0); rig.Nm.DiscardedEventReports.ShouldBeGreaterThan(0, "the waiting events were dropped, not delivered"); @@ -174,8 +188,8 @@ public sealed class FolderEventFloodWireTests .Select(m => int.Parse(m![1..], System.Globalization.CultureInfo.InvariantCulture)) .ToArray(); - // Positive control: the events reported before the channel filled did arrive (Server object: synchronously at - // the condition's own hop). + // Positive control: the event reported before the stall did arrive at the Server object (synchronously, at the + // condition's own hop, before the walk reached the stalled folder). Flooded(onServer).ShouldNotBeEmpty(); // None of the deferred tail — which was still waiting when the condition was removed — reached anyone. Flooded(onServer).Max().ShouldBeLessThan(FloodSize - 1, "no ghost reached the Server-object subscriber"); diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/FolderEventVerdictInvalidationTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/FolderEventVerdictInvalidationTests.cs index 37fe86526..f20a6ea7a 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/FolderEventVerdictInvalidationTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/FolderEventVerdictInvalidationTests.cs @@ -97,10 +97,10 @@ public sealed class FolderEventVerdictInvalidationTests rig.Project(native: false, State("s1")); await RecordedEvents.WaitUntilAsync(() => events.Has(onA, "s1")); - // Holding the node-manager lock parks EqA's SDK delivery task: the native event "n1" is queued first and its - // verdict (a cache miss after the first invalidation) needs the lock in the inherited role check — its node is - // never removed, so that check always takes it. The OLD scripted instance's "s2", queued behind it on the same - // single-reader queue, is therefore decided only after the replace and the apply's invalidation. + // Before Gitea #855, holding the node-manager lock parked EqA's SDK delivery task, so the OLD scripted instance's + // "s2" was decided (and its verdict cached) only after the replace and the apply's invalidation. Condition events + // are now decided per item at report time, uncached — "s2" is delivered while its instance is live — and this + // pins that the replace still leaves the new instance's events flowing either way. lock (rig.Nm.Lock) { rig.Nm.CompleteAddressSpaceApply(); // forget the cached allows, so both verdicts must be computed diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/NotifierStall.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/NotifierStall.cs new file mode 100644 index 000000000..755f26a09 --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/NotifierStall.cs @@ -0,0 +1,42 @@ +using Opc.Ua; + +namespace ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests; + +/// +/// Stalls condition-event delivery at one of the node manager's notifiers (Gitea #855): every condition event whose walk +/// reaches that notifier suspends there until , without blocking the reporter or holding the node +/// manager's Lock — so the condition's later events are deferred, as they were behind a full SDK channel before +/// folder items were delivered per item. +/// +/// +/// The pre-#855 tests produced deferral by holding a folder item's server-side MonitoredItem lock, which parked +/// the SDK's per-node delivery task. Folder items are now queued synchronously by the reporter (as Server-object items +/// always were), so holding an item lock would block the reporter instead; this goes through the node manager's test +/// probe (). The probe is consulted only at a notifier that has +/// event items (the hop the per-item delivery serves), so the stalled notifier must carry at least one. +/// +internal sealed class NotifierStall : IDisposable +{ + private readonly OtOpcUaNodeManager _nm; + private readonly TaskCompletionSource _released = new(TaskCreationOptions.RunContinuationsAsynchronously); + private readonly Func _probe; + + private NotifierStall(OtOpcUaNodeManager nm, string notifierId) + { + _nm = nm; + _probe = node => !_released.Task.IsCompleted && node.NodeId.IdentifierAsString == notifierId ? _released.Task : null; + nm.NotifierDeliveryStall = _probe; + } + + /// Starts stalling delivery at the notifier whose NodeId identifier is . + public static NotifierStall Hold(OtOpcUaNodeManager nm, string notifierId) => new(nm, notifierId); + + /// Lets every stalled walk continue (on pool threads, off Lock). + public void Release() => _released.TrySetResult(); + + public void Dispose() + { + Release(); + if (ReferenceEquals(_nm.NotifierDeliveryStall, _probe)) _nm.NotifierDeliveryStall = null; + } +} From d2b1545659bd1420764ce16d1fcee76eece0bd61 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Wed, 11 Nov 2026 12:31:02 -0500 Subject: [PATCH 2/5] =?UTF-8?q?fix(opcua):=20#855=20review=201=20=E2=80=94?= =?UTF-8?q?=20re-decide=20frozen=20replay=20entries;=20a=20failed=20delive?= =?UTF-8?q?ry=20is=20not=20a=20deferral?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - QueueFrozenReplay re-asks each item's verdict right before each entry's gate hold (Server-object and folder cut paths): a grant revoked between the pre-Lock verdicts and the replay no longer lets the replay through. A denied entry is also removed from the cut's re-sent set. - ReportEventWithoutBlocking logs and returns on a delivery that faulted inline instead of booking it as a deferral episode (which ran the drain and ClearStaleRefreshCuts under the event-report gate while Lock was held). - OnConditionReportEvent isolates failures per Server-object item. - Tests: revocation between cut and replay (folder + Server object), throwing hop not booked as a deferral; production-overload sampling inside Lock. Claude-Session: https://claude.ai/code/session_01GT67AD99zQQqpLCVm3p7xk --- docs/security.md | 6 +- .../OtOpcUaNodeManager.cs | 79 ++++++++++-- .../FolderConditionRefreshWireTests.cs | 112 +++++++++++++++++- 3 files changed, 182 insertions(+), 15 deletions(-) diff --git a/docs/security.md b/docs/security.md index 1e09644b1..1d02687b5 100644 --- a/docs/security.md +++ b/docs/security.md @@ -924,7 +924,11 @@ Behaviours worth knowing before enabling it: `QueueSize`) and the verdict runs on the reporter's thread, which already holds `Lock` — so a burst under one `Lock` hold completes with nothing deferred, and the #838 deadlock shape is gone for condition events. Folder items therefore see condition events in report order. The cost moves with it: the per-event verdict and filter for - folder items now run under `Lock` (as for Server-object items), where the SDK used to cache the verdict off it. + folder items now run under `Lock` (as for Server-object items), where the SDK used to cache the verdict off it — + including the SDK item queue's own per-event costs (its duplicate scan, and at a full queue a list shift), so many + folder items on one notifier whose clients stop publishing make each report proportionally slower under `Lock`. + With auditing on (off in this repo), a condition's SDK method **audit** event still takes the folder's channel, so + at a folder item it may now arrive after the condition event that method produced (they were one FIFO before). - **The deferral queue and the cut stay as the safety net**, for a hop that is not ours (none exists today: conditions are never event notifiers and variables cannot host event items) — and folder items share the cut: a `ConditionRefresh`/`ConditionRefresh2` of a folder item takes the same two paths as the Server object (plain walk diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs index c3d96cc82..56e116e51 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs @@ -1529,6 +1529,16 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3 if (delivery.IsCompletedSuccessfully) return; + // #855 review: a delivery that FAILED inline is over, not suspended. Booking it as a deferral used to open and close + // a deferral episode synchronously (moving the episode a concurrent plain refresh checks, and clearing cuts under + // the event-report gate while Lock was held). Only an incomplete delivery is deferred. + if (delivery.IsCompleted) + { + _logger.LogError(delivery.AsTask().Exception?.GetBaseException(), + "OtOpcUaNodeManager: failed to report Part 9 condition event for {NodeId}", condition.NodeId); + return; + } + lock (_eventReportGate) { var queue = new DeferredConditionReports(); @@ -1901,6 +1911,9 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3 /// private long _deferralEpisode; + /// Deferral episodes ended since start (test observation, #855 review). + internal long DeferralEpisodes => Interlocked.Read(ref _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 deferral episode the cut was taken in; once it ends the cut is stale and inert. @@ -5782,17 +5795,26 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3 foreach (var item in _allEventsItems.Keys) { - if (e is AuditEventState && !IsAuditDeliverableTo(item)) continue; - if (scopes is not null && !IsConditionVisibleTo(item, scopes)) continue; - if (ServiceResult.IsBad(ValidateEventRolePermissions(item, e))) continue; - if (reportingSession is { IsNull: false } reportingId - && item.Session?.Id is { } own - && !own.Equals(reportingId)) + try { - continue; - } + if (e is AuditEventState && !IsAuditDeliverableTo(item)) continue; + if (scopes is not null && !IsConditionVisibleTo(item, scopes)) continue; + if (ServiceResult.IsBad(ValidateEventRolePermissions(item, e))) continue; + if (reportingSession is { IsNull: false } reportingId + && item.Session?.Id is { } own + && !own.Equals(reportingId)) + { + continue; + } - QueueConditionEventToItem(item, e, deferred, DeferredServerObjectDeliveryProbe); + QueueConditionEventToItem(item, e, deferred, DeferredServerObjectDeliveryProbe); + } + catch (Exception ex) + { + // #855 review: one item's failure (an item disposed under us, the SDK's operation guard at shutdown) must + // not cost the other Server-object items, nor abort the walk before the folder hops. + _logger.LogWarning(ex, "OtOpcUaNodeManager: failed to deliver a condition event to Server-object item {ItemId}", item.Id); + } } } @@ -6671,7 +6693,24 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3 } ServerObjectRefreshProbe?.Invoke(true, Monitor.IsEntered(Lock)); - QueueFrozenReplay(items, states, perItem.Select(entries => entries.Select(x => (x.Event, x.Condition.NodeId)).ToList()).ToArray(), generation); + // #855 review: the verdicts above were taken before Lock; a deploy may have revoked a grant since. Each replay entry + // is re-decided right before it is queued (scopes resolved once per condition). + var scopeCache = new Dictionary>(); + QueueFrozenReplay( + items, states, perItem.Select(entries => entries.Select(x => (x.Event, x.Condition.NodeId)).ToList()).ToArray(), + generation, + (item, e) => + { + if (enforced) + { + if (ConditionOf(e) is not { } condition) return false; + if (!scopeCache.TryGetValue(condition.NodeId, out var scopes)) + scopeCache[condition.NodeId] = scopes = ConditionVisibilityScopes(condition); + if (!IsConditionVisibleTo(item, scopes)) return false; + } + + return !ServiceResult.IsBad(ValidateEventRolePermissions(item, e)); + }); } finally { @@ -6685,8 +6724,15 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3 /// 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. /// + /// + /// The item's verdict for one entry, asked right before the entry's gate hold (never under it: it takes Lock), + /// against the CURRENT generation — the verdicts the cut was built on were taken earlier (#855 review). An entry + /// denied here is dropped AND removed from the cut's re-sent set, so the condition counts as "omitted for the + /// grants" and its later events are left to the ACL rather than withheld as superseded. + /// private static void QueueFrozenReplay( - List items, List states, List<(IFilterTarget Event, NodeId ConditionId)>[] perItem, int generation) + List items, List states, List<(IFilterTarget Event, NodeId ConditionId)>[] perItem, + int generation, Func allowed) { for (var i = 0; i < items.Count; i++) { @@ -6694,9 +6740,17 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3 var state = states[i]; foreach (var (e, conditionId) in perItem[i]) { + if (state.PendingReplay != generation) break; + var permitted = allowed(item, e); lock (state.Gate) { if (state.PendingReplay != generation) break; + if (!permitted) + { + state.Cut?.Refreshed.Remove(conditionId); + continue; + } + if (state.NewerSinceCut.Contains(conditionId)) continue; item.QueueEvent(e); } @@ -6926,7 +6980,8 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3 } DirectRefreshProbe?.Invoke(true, Monitor.IsEntered(Lock)); - QueueFrozenReplay(items, states, perItem, generation); + // #855 review: re-decided per entry right before queueing — a deploy may have revoked since the verdicts. + QueueFrozenReplay(items, states, perItem, generation, (item, e) => IsDirectRefreshEventAllowed(item, e, enforced)); } finally { diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/FolderConditionRefreshWireTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/FolderConditionRefreshWireTests.cs index b01a93427..0a9bc98b9 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/FolderConditionRefreshWireTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/FolderConditionRefreshWireTests.cs @@ -358,6 +358,110 @@ public sealed class FolderConditionRefreshWireTests keyEvents.Has(keyA, "after-revoke").ShouldBeFalse(); } + /// + /// #855 review: the cut path decides its replay before taking Lock; a grant revoked between that decision + /// and the queueing of the frozen replay must still withhold the replay (it is re-decided right before each entry). + /// The revocation lands from the refresh probe, which runs after the cut is installed and before the replay is queued. + /// + /// Refresh a Server-object item instead of a folder item. + [Theory] + [InlineData(false)] + [InlineData(true)] + public async Task A_grant_revoked_between_the_cut_and_its_replay_withholds_the_replay(bool serverObject) + { + var ct = TestContext.Current.CancellationToken; + await using var rig = await ServerEventRig.StartAsync("Refresh855RevokeCut", encrypted: false, ct); + rig.KeyRoles = []; + rig.MaterialiseNative(); + var anonGrant = ServerEventRig.EquipmentGrant(EqA, NodePermissions.ReadOnly); + var state = rig.Enforce(true, [ServerEventRig.Key()], KeyEquipmentGrant(EqA), anonGrant); + + // An anonymous subscriber keeps EqA a notifier fan-out the stall applies to (and stays granted throughout). + using var anonymous = await rig.ConnectAsync(anonymous: true, ct); + var anonEvents = new RecordedEvents(); + var anonSub = await RecordedEvents.SubscribeAsync(anonymous, anonEvents, ct); + RecordedEvents.AddEventItem(anonSub, ServerEventRig.UnsNodeId(anonymous, EqA)); + await anonSub.ApplyChangesAsync(ct); + + using var keyed = await rig.ConnectAsync(anonymous: false, ct); + var keyEvents = new RecordedEvents(); + var keySub = await RecordedEvents.SubscribeAsync(keyed, keyEvents, ct); + var item = RecordedEvents.AddEventItem(keySub, serverObject ? ObjectIds.Server : ServerEventRig.UnsNodeId(keyed, EqA)); + await keySub.ApplyChangesAsync(ct); + item.Status.Error.ShouldBeNull(); + + rig.Project(native: true, Active("seed")); + await RecordedEvents.WaitUntilAsync(() => keyEvents.Has(item, "seed")); + + var revoked = 0; + void Revoke(bool cut, bool _) + { + if (!cut || Interlocked.Exchange(ref revoked, 1) != 0) return; + state.Apply( + ServerEventRig.Cluster, "rev-2", [ServerEventRig.Key()], [anonGrant], ServerEventRig.ScopePaths(), + ServerEventRig.EquipmentOwnerPaths(), ServerEventRig.ScriptedAlarmOwners(), ServerEventRig.FolderScopes(), + ServerEventRig.TagReferences()); + } + + if (serverObject) rig.Nm.ServerObjectRefreshProbe = Revoke; + else rig.Nm.DirectRefreshProbe = Revoke; + + using (var stall = NotifierStall.Hold(rig.Nm, EqA)) + { + for (var i = 0; i < FloodSize; i++) rig.Project(native: true, Active($"f{i}")); + rig.Nm.DeferredEventReports.ShouldBeGreaterThan(0); + await keyed.CallAsync(ObjectTypeIds.ConditionType, MethodIds.ConditionType_ConditionRefresh, ct, keySub.Id); + await RecordedEvents.WaitUntilAsync(() => keyEvents.HasType(item, ObjectTypeIds.RefreshEndEventType)); + stall.Release(); + } + + rig.Nm.ServerObjectRefreshProbe = null; + rig.Nm.DirectRefreshProbe = null; + Volatile.Read(ref revoked).ShouldBe(1, "the refresh took the cut path"); + await RecordedEvents.WaitUntilAsync(() => rig.Nm.DeferredEventReports == 0); + await Task.Delay(TimeSpan.FromMilliseconds(500), ct); + + var (replayed, after) = Window(keyEvents.Of(item)); + replayed.ShouldBeEmpty("the grant was revoked before the replay was queued"); + after.ShouldBeEmpty("and nothing reaches the item after RefreshEnd either"); + } + + /// + /// #855 review: a delivery that fails inline (here: a hop that throws) is over, not suspended — it must not be booked + /// as a deferral episode (which moved the episode a concurrent plain refresh checks, and cleared cuts under the + /// event-report gate while Lock was held). The Server-object item, served before the failing hop, still gets + /// the event, and the next event is delivered normally. + /// + [Fact] + public async Task A_delivery_that_throws_is_not_booked_as_a_deferral() + { + var ct = TestContext.Current.CancellationToken; + await using var rig = await ServerEventRig.StartAsync("Refresh855Fault", encrypted: false, ct); + rig.KeyRoles = []; + rig.MaterialiseNative(); + rig.Enforce(false, [ServerEventRig.Key()], KeyEquipmentGrant(EqA)); + + using var session = await rig.ConnectAsync(anonymous: false, ct); + var events = new RecordedEvents(); + var subscription = await RecordedEvents.SubscribeAsync(session, events, ct); + var onA = RecordedEvents.AddEventItem(subscription, ServerEventRig.UnsNodeId(session, EqA)); + var onServer = RecordedEvents.AddEventItem(subscription, ObjectIds.Server); + await subscription.ApplyChangesAsync(ct); + + var episodes = rig.Nm.DeferralEpisodes; + rig.Nm.NotifierDeliveryStall = node => + node.NodeId.IdentifierAsString == EqA ? throw new InvalidOperationException("hop fault (test)") : null; + rig.Project(native: true, Active("faulted")); + rig.Nm.NotifierDeliveryStall = null; + + rig.Nm.DeferredEventReports.ShouldBe(0); + rig.Nm.DeferralEpisodes.ShouldBe(episodes, "a failed delivery is not a deferral episode"); + rig.Project(native: true, Active("next")); + await RecordedEvents.WaitUntilAsync(() => + events.Has(onServer, "faulted") && events.Has(onServer, "next") && events.Has(onA, "next")); + events.Has(onA, "faulted").ShouldBeFalse("the faulted hop delivered nothing"); + } + /// /// Production shape, no stall: bursts reported under one Lock hold (an apply pass) and single reports, against /// folder items that refresh repeatedly while their client is slow to publish. Folder items are served per item, so @@ -398,10 +502,14 @@ public sealed class FolderConditionRefreshWireTests { lock (rig.Nm.Lock) { - for (var i = 0; i < 1000; i++) rig.Project(native: true, Active($"s{Interlocked.Increment(ref counter)}")); + for (var i = 0; i < 1000; i++) + { + rig.Project(native: true, Active($"s{Interlocked.Increment(ref counter)}")); + // Sampled inside Lock, after each report: a deferral cannot open and drain unseen (#855 review). + peakDeferred = Math.Max(peakDeferred, rig.Nm.DeferredEventReports); + } } - peakDeferred = Math.Max(peakDeferred, rig.Nm.DeferredEventReports); rig.Project(native: true, Active($"s{Interlocked.Increment(ref counter)}")); Thread.Sleep(2); } From f775625805b8a859aafefae5ad6996b1add45f41 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Wed, 11 Nov 2026 12:40:06 -0500 Subject: [PATCH 3/5] =?UTF-8?q?fix(opcua):=20#855=20review=202=20=E2=80=94?= =?UTF-8?q?=20frozen=20replay=20skips=20non-live=20conditions;=20per-hop?= =?UTF-8?q?=20ACL=20resolution;=20drain=20never=20inline?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - QueueFrozenReplay's verdict refuses a condition a deploy removed or replaced since the cut (Server-object and folder), so the cut path cannot replay a ghost or judge a replaced scripted alarm on its old folder. - DeliverToNotifierItems resolves the #826 decision's item-independent part (live condition, notifier check, scopes) once per hop; only the grant and SDK role checks stay per item. - A deferred delivery's drain starts on the pool, never inline under Lock. - Test: a condition removed between the cut and its replay (folder + Server object). Claude-Session: https://claude.ai/code/session_01GT67AD99zQQqpLCVm3p7xk --- docs/security.md | 3 +- .../OtOpcUaNodeManager.cs | 40 ++++++++++-- .../FolderConditionRefreshWireTests.cs | 63 +++++++++++++++++++ 3 files changed, 101 insertions(+), 5 deletions(-) diff --git a/docs/security.md b/docs/security.md index 1d02687b5..a284f90c8 100644 --- a/docs/security.md +++ b/docs/security.md @@ -917,7 +917,8 @@ Behaviours worth knowing before enabling it: event type, source), so a refresh of a folder item could still be followed by older events. Now, at every notifier the node manager owns, the walk serves the attached items itself (`DeliverToNotifierItems`), exactly as it serves Server-object items: the SDK's audit guard, the verdict through `INodeManager3.ValidateEventRolePermissions` (SDK - type/source `ReceiveEvents` + the #826 decision under enforcement) **per event, uncached**, then the item's gate and + type/source `ReceiveEvents` + the #826 decision under enforcement — its item-independent part, the live condition, + the notifier check and the visibility scopes, resolved once per hop) **per event, uncached**, then the item's gate and cut, then `QueueEvent`. The `MonitoredNode2` stays for item attachment and for what still uses its channel (a condition's SDK method audit events, data-change items on the folder), so `InvalidateFolderEventVerdicts` stays. - **No notifier of ours defers any more.** `QueueEvent` never waits for a consumer (the item queue discards oldest at diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs index 56e116e51..74104d3d6 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs @@ -1539,13 +1539,19 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3 return; } + DeferredConditionReports queue; lock (_eventReportGate) { - var queue = new DeferredConditionReports(); + queue = new DeferredConditionReports(); _deferredReports[condition] = queue; CountDeferred(+1, condition); - _ = DrainDeferredAsync(condition, queue, delivery.AsTask()); } + + // #855 review 2: started on the pool, never inline. The suspended delivery may complete between the check above and + // here; an inline drain would then run (and end the episode, and clear cuts over every item gate) on this thread, + // under Lock. The queue is registered first, so this condition's later reports wait behind it either way. + var inFlight = delivery.AsTask(); + _ = Task.Run(() => DrainDeferredAsync(condition, queue, inFlight)); } /// Queues a later event behind a condition's deferred one, coalescing past the budget. Under the gate. @@ -1719,13 +1725,33 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3 private void DeliverToNotifierItems(MonitoredNode2 monitored, IFilterTarget e) { var deferred = e is ConditionEventSnapshot { Deferred: true } d ? d : null; + + // #855 review 2: the #826 decision's item-independent part — the live condition, whether this notifier is still a + // legitimate notifier of it, and its visibility scopes — is resolved ONCE per hop, not per item (it is what + // IsNotifierEventVisibleTo resolves for an item attached here). Only the grant check and the SDK's role check + // remain per item. + var enforced = AclEnforced; + List? scopes = null; + if (enforced) + { + if (EventConditionOf(e) is not { } named + || !_alarmConditions.TryGetValue(MapKey(named.NodeId), out var live) + || !IsLiveNotifierOf(live, monitored.Node)) + { + return; // fail closed for every item on this notifier + } + + scopes = ConditionVisibilityScopes(live); + } + foreach (var item in monitored.EventMonitoredItems.Values) { if (item is null) continue; try { if (IsAuditEvent(e) && !IsAuditDeliverableTo(item)) continue; - if (ServiceResult.IsBad(((INodeManager3)this).ValidateEventRolePermissions(item, e))) continue; + if (ServiceResult.IsBad(ValidateEventRolePermissions(item, e))) continue; + if (scopes is not null && !IsConditionVisibleTo(item, scopes)) continue; QueueConditionEventToItem(item, e, deferred, DeferredFolderDeliveryProbe); } catch (Exception ex) @@ -6701,6 +6727,9 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3 generation, (item, e) => { + // #855 review 2: never replay a condition a deploy has removed or replaced since the cut — its frozen + // state would be a ghost (and, with the ACL on, a replaced scripted alarm would be judged on its old folder). + if (ConditionOf(e) is { } frozen && !IsLiveCondition(frozen)) return false; if (enforced) { if (ConditionOf(e) is not { } condition) return false; @@ -6981,7 +7010,10 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3 DirectRefreshProbe?.Invoke(true, Monitor.IsEntered(Lock)); // #855 review: re-decided per entry right before queueing — a deploy may have revoked since the verdicts. - QueueFrozenReplay(items, states, perItem, generation, (item, e) => IsDirectRefreshEventAllowed(item, e, enforced)); + QueueFrozenReplay( + items, states, perItem, generation, + (item, e) => (ConditionOf(e) is not AlarmConditionState frozen || IsLiveCondition(frozen)) + && IsDirectRefreshEventAllowed(item, e, enforced)); } finally { diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/FolderConditionRefreshWireTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/FolderConditionRefreshWireTests.cs index 0a9bc98b9..59287d809 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/FolderConditionRefreshWireTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/FolderConditionRefreshWireTests.cs @@ -426,6 +426,69 @@ public sealed class FolderConditionRefreshWireTests after.ShouldBeEmpty("and nothing reaches the item after RefreshEnd either"); } + /// + /// #855 review 2: a condition removed by a deploy between the cut and the queueing of its frozen replay must not be + /// replayed — its frozen Retain=true state would be a ghost no Retain=false ever follows. ACL off, so nothing but the + /// liveness check stands in the way. + /// + /// Refresh a Server-object item instead of a folder item. + [Theory] + [InlineData(false)] + [InlineData(true)] + public async Task A_condition_removed_between_the_cut_and_its_replay_is_not_replayed(bool serverObject) + { + var ct = TestContext.Current.CancellationToken; + await using var rig = await ServerEventRig.StartAsync("Refresh855RemoveCut", encrypted: false, ct); + rig.KeyRoles = []; + rig.MaterialiseNative(); + rig.Enforce(false, [ServerEventRig.Key()], KeyEquipmentGrant(EqA)); + + // A second subscriber keeps EqA a notifier fan-out the stall applies to. + using var other = await rig.ConnectAsync(anonymous: true, ct); + var otherEvents = new RecordedEvents(); + var otherSub = await RecordedEvents.SubscribeAsync(other, otherEvents, ct); + RecordedEvents.AddEventItem(otherSub, ServerEventRig.UnsNodeId(other, EqA)); + await otherSub.ApplyChangesAsync(ct); + + using var session = await rig.ConnectAsync(anonymous: false, ct); + var events = new RecordedEvents(); + var subscription = await RecordedEvents.SubscribeAsync(session, events, ct); + var item = RecordedEvents.AddEventItem(subscription, serverObject ? ObjectIds.Server : ServerEventRig.UnsNodeId(session, EqA)); + await subscription.ApplyChangesAsync(ct); + + rig.Project(native: true, Active("seed")); + await RecordedEvents.WaitUntilAsync(() => events.Has(item, "seed")); + + var removed = 0; + void Remove(bool cut, bool _) + { + if (cut && Interlocked.Exchange(ref removed, 1) == 0) + rig.Nm.RemoveAlarmConditionNode(ServerEventRig.RawAlarmPath, AddressSpaceRealm.Raw).ShouldBeTrue(); + } + + if (serverObject) rig.Nm.ServerObjectRefreshProbe = Remove; + else rig.Nm.DirectRefreshProbe = Remove; + + using (var stall = NotifierStall.Hold(rig.Nm, EqA)) + { + for (var i = 0; i < FloodSize; i++) rig.Project(native: true, Active($"f{i}")); + rig.Nm.DeferredEventReports.ShouldBeGreaterThan(0); + await session.CallAsync(ObjectTypeIds.ConditionType, MethodIds.ConditionType_ConditionRefresh, ct, subscription.Id); + await RecordedEvents.WaitUntilAsync(() => events.HasType(item, ObjectTypeIds.RefreshEndEventType)); + stall.Release(); + } + + rig.Nm.ServerObjectRefreshProbe = null; + rig.Nm.DirectRefreshProbe = null; + Volatile.Read(ref removed).ShouldBe(1, "the refresh took the cut path"); + await RecordedEvents.WaitUntilAsync(() => rig.Nm.DeferredEventReports == 0); + await Task.Delay(TimeSpan.FromMilliseconds(500), ct); + + var (replayed, after) = Window(events.Of(item)); + replayed.ShouldBeEmpty("the condition was removed before its frozen replay was queued"); + after.ShouldBeEmpty("and none of its deferred events follows"); + } + /// /// #855 review: a delivery that fails inline (here: a hop that throws) is over, not suspended — it must not be booked /// as a deferral episode (which moved the episode a concurrent plain refresh checks, and cleared cuts under the From 7783fadae1b8bb54a58c73a9933ea36eabc71c22 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Wed, 11 Nov 2026 12:51:03 -0500 Subject: [PATCH 4/5] =?UTF-8?q?fix(opcua):=20#855=20review=203=20=E2=80=94?= =?UTF-8?q?=20plain=20refresh=20skips=20removed=20conditions;=20per-hop=20?= =?UTF-8?q?failure=20isolation?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Both plain ConditionRefresh paths (Server object and folder) skip a condition a deploy removed or replaced between the off-Lock walk and the queueing (the SDK base did not either; only the cut path checked). - A failure delivering at one notifier hop is logged and the walk continues to the remaining hops; a failure deciding the Server-object items withholds only them. - Test: a condition removed during a plain refresh (folder + Server object). Claude-Session: https://claude.ai/code/session_01GT67AD99zQQqpLCVm3p7xk --- .../OtOpcUaNodeManager.cs | 31 ++++++++++++- .../FolderConditionRefreshWireTests.cs | 43 +++++++++++++++++++ 2 files changed, 72 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 74104d3d6..f6af07d1a 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs @@ -1651,7 +1651,19 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3 } // Re-checked: a stall may have outlived the condition (the liveness check above ran before it). - if (IsLiveCondition(condition)) DeliverToNotifierItems(monitored, e); + if (IsLiveCondition(condition)) + { + // #855 review 3: one hop's failure must not cost the hops after it (before #855 this hop was a channel + // write, which cannot throw; the SDK isolated failures per item). + try + { + DeliverToNotifierItems(monitored, e); + } + catch (Exception ex) + { + _logger.LogWarning(ex, "OtOpcUaNodeManager: failed to deliver a condition event at notifier {NodeId}", node.NodeId); + } + } } else { @@ -5815,7 +5827,18 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3 // through its OLD equipment folder (#808). if (!IsLiveCondition(node)) return; - var scopes = AclEnforced ? ConditionVisibilityScopes(node) : null; + List? scopes; + try + { + scopes = AclEnforced ? ConditionVisibilityScopes(node) : null; + } + catch (Exception ex) + { + // #855 review 3: withheld from the Server-object items (fail closed), but the walk goes on to the folders. + _logger.LogWarning(ex, "OtOpcUaNodeManager: failed to decide a condition event for Server-object items ({NodeId})", node.NodeId); + return; + } + var reportingSession = (context as ISessionSystemContext)?.SessionId; var deferred = e is ConditionEventSnapshot { Deferred: true } d ? d : null; @@ -6595,6 +6618,8 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3 foreach (var (e, _, scopes) in replay) { + // #855 review 3: never replay a condition a deploy removed or replaced since the walk collected it. + if (ConditionOf(e) is AlarmConditionState walked && !IsLiveCondition(walked)) continue; if (scopes is not null && !IsConditionVisibleTo(item, scopes)) continue; if (ServiceResult.IsBad(ValidateEventRolePermissions(item, e))) continue; item.QueueEvent(e); @@ -6878,6 +6903,8 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3 DirectRefreshProbe?.Invoke(false, Monitor.IsEntered(Lock)); foreach (var e in events) { + // #855 review 3: never replay a condition a deploy removed or replaced since the walk collected it. + if (ConditionOf(e) is AlarmConditionState walked && !IsLiveCondition(walked)) continue; if (!IsDirectRefreshEventAllowed(item, e, enforced)) continue; item.QueueEvent(e); } diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/FolderConditionRefreshWireTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/FolderConditionRefreshWireTests.cs index 59287d809..b6deba5e2 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/FolderConditionRefreshWireTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/FolderConditionRefreshWireTests.cs @@ -489,6 +489,49 @@ public sealed class FolderConditionRefreshWireTests after.ShouldBeEmpty("and none of its deferred events follows"); } + /// + /// #855 review 3: the plain refresh path (nothing deferred — the only path production takes) collects live + /// conditions off Lock and queues them afterwards; a condition removed in between must not be replayed. + /// + /// Refresh a Server-object item instead of a folder item. + [Theory] + [InlineData(false)] + [InlineData(true)] + public async Task A_condition_removed_during_a_plain_refresh_is_not_replayed(bool serverObject) + { + var ct = TestContext.Current.CancellationToken; + await using var rig = await ServerEventRig.StartAsync("Refresh855RemovePlain", encrypted: false, ct); + rig.KeyRoles = []; + rig.MaterialiseNative(); + rig.Enforce(false, [ServerEventRig.Key()], KeyEquipmentGrant(EqA)); + + using var session = await rig.ConnectAsync(anonymous: false, ct); + var events = new RecordedEvents(); + var subscription = await RecordedEvents.SubscribeAsync(session, events, ct); + var item = RecordedEvents.AddEventItem(subscription, serverObject ? ObjectIds.Server : ServerEventRig.UnsNodeId(session, EqA)); + await subscription.ApplyChangesAsync(ct); + + rig.Project(native: true, Active("seed")); + await RecordedEvents.WaitUntilAsync(() => events.Has(item, "seed")); + + var removed = 0; + void Remove(bool cut, bool _) + { + if (!cut && Interlocked.Exchange(ref removed, 1) == 0) + rig.Nm.RemoveAlarmConditionNode(ServerEventRig.RawAlarmPath, AddressSpaceRealm.Raw).ShouldBeTrue(); + } + + if (serverObject) rig.Nm.ServerObjectRefreshProbe = Remove; + else rig.Nm.DirectRefreshProbe = Remove; + await session.CallAsync(ObjectTypeIds.ConditionType, MethodIds.ConditionType_ConditionRefresh, ct, subscription.Id); + await RecordedEvents.WaitUntilAsync(() => events.HasType(item, ObjectTypeIds.RefreshEndEventType)); + rig.Nm.ServerObjectRefreshProbe = null; + rig.Nm.DirectRefreshProbe = null; + + Volatile.Read(ref removed).ShouldBe(1, "the plain path ran"); + Window(events.Of(item)).Replayed.ShouldBeEmpty("the condition was removed before the replay was queued"); + } + /// /// #855 review: a delivery that fails inline (here: a hop that throws) is over, not suspended — it must not be booked /// as a deferral episode (which moved the episode a concurrent plain refresh checks, and cleared cuts under the From ec17a8797b58189b142c9edcf4cc0ae21095bdf4 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Wed, 11 Nov 2026 13:05:06 -0500 Subject: [PATCH 5/5] =?UTF-8?q?fix(opcua):=20#855=20review=204=20=E2=80=94?= =?UTF-8?q?=20per-item=20"known"=20in=20the=20folder=20cut;=20pin=20the=20?= =?UTF-8?q?post-stall=20liveness=20check?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - RefreshDirectItemsWithCut keys its "decided off Lock" set per (item, condition), so a condition one item's walk saw is still decided for another. - The deferral warning no longer blames slow folder subscribers (no folder of ours defers since #855). - FolderEventFloodWireTests' removal test now asserts the folder item got nothing and the Server object exactly the pre-stall event (it passed with the post-stall liveness re-check deleted). Claude-Session: https://claude.ai/code/session_01GT67AD99zQQqpLCVm3p7xk --- .../OtOpcUaNodeManager.cs | 13 +++++++------ .../FolderEventFloodWireTests.cs | 5 ++++- 2 files changed, 11 insertions(+), 7 deletions(-) diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs index f6af07d1a..0bed75dad 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs @@ -1806,9 +1806,9 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3 { _lastDeferralWarningUtc = now; _logger.LogWarning( - "OtOpcUaNodeManager: condition events are being deferred — an event notifier's delivery channel is " - + "full (first: {NodeId}). Folder event subscribers are not keeping up, or Lock is held across a large " - + "burst; events stay ordered per condition, and past {PerCondition} waiting per condition the oldest " + "OtOpcUaNodeManager: condition events are being deferred — a notifier hop the node manager does not " + + "serve itself suspended (first: {NodeId}); since #855 no folder of ours does, so this is unexpected. " + + "Events stay ordered per condition, and past {PerCondition} waiting per condition the oldest " + "give way to the newest (Gitea #838)", condition.NodeId, MaxDeferredEventsPerCondition); } @@ -6948,7 +6948,8 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3 // Verdicts per (item, condition) before Lock, from a live walk: they depend on the event type, source, notifier // wiring and grants, not on the condition's state. var allowed = new HashSet<(IEventMonitoredItem, NodeState)>(); - var known = new HashSet(System.Collections.Generic.ReferenceEqualityComparer.Instance); + // Per (item, condition): a condition one item's walk saw is still undecided for another item (#855 review 4). + var known = new HashSet<(IEventMonitoredItem, NodeState)>(); foreach (var item in items) { if (AttachedNotifierOf(item) is not { } notifier) continue; @@ -6957,7 +6958,7 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3 foreach (var e in events) { if (ConditionOf(e) is not { } node) continue; - known.Add(node); + known.Add((item, node)); if (IsDirectRefreshEventAllowed(item, e, enforced)) allowed.Add((item, node)); } } @@ -7015,7 +7016,7 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3 // A condition materialised since the verdicts were taken is decided here (rare). if (!allowed.Contains((item, instance))) { - if (known.Contains(instance)) continue; + if (known.Contains((item, instance))) continue; if (!IsDirectRefreshEventAllowed(item, target, enforced)) continue; } diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/FolderEventFloodWireTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/FolderEventFloodWireTests.cs index 97f2271de..266637e48 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/FolderEventFloodWireTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/FolderEventFloodWireTests.cs @@ -193,7 +193,10 @@ public sealed class FolderEventFloodWireTests Flooded(onServer).ShouldNotBeEmpty(); // None of the deferred tail — which was still waiting when the condition was removed — reached anyone. Flooded(onServer).Max().ShouldBeLessThan(FloodSize - 1, "no ghost reached the Server-object subscriber"); - Flooded(onA).DefaultIfEmpty(-1).Max().ShouldBeLessThan(FloodSize - 1, "no ghost reached the folder subscriber"); + // Only the in-flight f0 ever reached the stalled EqA hop, and the condition was gone when the stall released: the + // post-stall liveness re-check must have withheld it (#855 review 4 — "< FloodSize - 1" passed without that check). + Flooded(onA).ShouldBeEmpty("no ghost reached the folder subscriber"); + Flooded(onServer).ShouldBe([0], "only the event reported before the stall reached the Server object"); } private static AlarmConditionSnapshot State(string message) =>