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..a284f90c8 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,53 @@ 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 — 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 + `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 — + 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 + 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..0bed75dad 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 @@ -1522,13 +1529,29 @@ 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; + } + + 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. @@ -1615,7 +1638,42 @@ 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)) + { + // #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 + { + // 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 +1689,101 @@ 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; + + // #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(ValidateEventRolePermissions(item, e))) continue; + if (scopes is not null && !IsConditionVisibleTo(item, scopes)) 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); @@ -1653,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); } @@ -1711,18 +1864,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 +1911,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 +1933,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 @@ -1780,6 +1949,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. @@ -1810,9 +1982,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) @@ -5655,54 +5827,87 @@ 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; 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; - } - - // #854: a synchronous event (under Lock) needs the gate only while a refresh's replay is pending, and the - // pending marker is installed under Lock, so an item no refresh has touched has nothing to order. A deferred - // event always takes the gate — its state is created here if need be, so a refresh racing it shares it. - AllEventsItemState? state; - if (deferred is not null) state = AllEventsItemStateOf(item); - else if (!_allEventsItemStates.TryGetValue(item, out state) || state.PendingReplay == 0 || e is not ConditionEventSnapshot) - { - item.QueueEvent(e); - continue; - } - - // 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)) + 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)) { - 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 (deferred is not null) DeferredServerObjectDeliveryProbe?.Invoke(); - item.QueueEvent(e); + 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); + } + } + } + + /// + /// 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)) + { + Interlocked.Increment(ref _refreshSupersededDeliveries); + OtOpcUaTelemetry.ConditionEventDeliveriesSuperseded.Add(1); + return; + } + + if (state.PendingReplay != 0 && e is ConditionEventSnapshot { } ce && ce.Sequence > state.PendingWatermark) + state.NewerSinceCut.Add(ce.ConditionId); + + if (deferred is not null) probe?.Invoke(); + item.QueueEvent(e); } } @@ -6080,8 +6285,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 +6354,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 +6488,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 +6511,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 +6545,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 +6561,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,17 +6614,12 @@ 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) { + // #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); @@ -6472,7 +6675,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 +6743,69 @@ 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)); + // #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) => + { + // #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; + 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 + { + 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. + /// + /// + /// 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, Func allowed) + { 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]) { + if (state.PendingReplay != generation) break; + var permitted = allowed(item, e); lock (state.Gate) { if (state.PendingReplay != generation) break; - if (state.NewerSinceCut.Contains(condition.NodeId)) continue; + if (!permitted) + { + state.Cut?.Refreshed.Remove(conditionId); + continue; + } + + if (state.NewerSinceCut.Contains(conditionId)) continue; item.QueueEvent(e); } } @@ -6567,51 +6819,234 @@ 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) + { + // #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); + } } - 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)>(); + // 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; 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((item, 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((item, 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)); + // #855 review: re-decided per entry right before queueing — a deploy may have revoked since the verdicts. + QueueFrozenReplay( + items, states, perItem, generation, + (item, e) => (ConditionOf(e) is not AlarmConditionState frozen || IsLiveCondition(frozen)) + && IsDirectRefreshEventAllowed(item, e, enforced)); + } + 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..b6deba5e2 --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/FolderConditionRefreshWireTests.cs @@ -0,0 +1,841 @@ +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(); + } + + /// + /// #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 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 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 + /// 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 + /// 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)}")); + // Sampled inside Lock, after each report: a deferral cannot open and drain unseen (#855 review). + 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..266637e48 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,12 +188,15 @@ 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"); - 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) => 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; + } +}