diff --git a/.claude/rules/data-plane-auth.md b/.claude/rules/data-plane-auth.md index 62a185a44..3268b7f18 100644 --- a/.claude/rules/data-plane-auth.md +++ b/.claude/rules/data-plane-auth.md @@ -212,9 +212,12 @@ Two things to know before relying on any of this: `OnConditionReportEvent`) — a deferred event of a removed/replaced condition is dropped, never a ghost on `i=2253`; per-CONDITION ordered queues on overflow; coalesce past 1024/condition or 65 536 total (oldest not-started gives way to newest); Warning/Information + `otopcua.opcua.condition_events.{deferred,discarded}`. A Server-object ConditionRefresh - (and ConditionRefresh2) takes a sequence CUT in one `Lock` hold and withholds older deferred events from THAT item - only (re-sent condition, or a Retain=true event of an omitted one) under a per-item gate held across check+QueueEvent - (#854, `ConditionRefreshDeferralWireTests`); folder items keep the residual (SDK channel + memoised verdict). Folder events follow Part 3's shape — + (and ConditionRefresh2, #854) with nothing deferred is the plain walk OFF `Lock`; while deferring it takes a sequence + CUT (one short `Lock` hold: watermark + frozen snapshots of Retain conditions from `_alarmConditions`), queues the + replay off `Lock` one event per hold of a per-item gate, and withholds older deferred events from THAT item only + (re-sent condition, or Retain=true of one omitted because NOT retained — never one omitted by ACL). Item state is a + `ConditionalWeakTable` keyed by the item; cuts clear when the deferral count hits 0. Tests: + `ConditionRefreshDeferralWireTests`; folder items keep the residual (SDK channel + memoised verdict). Folder events follow Part 3's shape — `ReceiveEvents` on EventType + SourceNode (SDK check, kept first) AND our decision; non-condition events on a folder item fail closed under enforcement. Tests: `FolderEventFloodWireTests`. - **ns=0 standard nodes stay ungated by design, even with `Security:Acl:Enabled`** (#617) — `ServerStatus`, diff --git a/docs/security.md b/docs/security.md index 5c6d21880..3e15eeb37 100644 --- a/docs/security.md +++ b/docs/security.md @@ -885,16 +885,33 @@ Behaviours worth knowing before enabling it: - **`ConditionRefresh` is not overtaken by deferred events on Server-object items (Gitea #854).** A refresh of an `i=2253` item replays each condition's current state between RefreshStart and RefreshEnd; the condition's deferred events — all older — used to follow RefreshEnd, briefly showing a stale state or re-adding an alarm the refresh had - just dropped (ScadaBridge's snapshot swap drops absent conditions, then a stale active event re-added it). Now every - condition event carries its report sequence; the refresh takes its **cut** (the sequence, whether anything is - deferred, the replay, and its queueing) in **one hold of `Lock`**, so the replay is exactly the state at the cut, and - installs it per item under that item's gate, which a deferred delivery also holds across its check and - `QueueEvent`. A deferred event at or below the cut is withheld **from that item only** when the refresh re-sent its - condition, or when it is `Retain=true` for a condition the refresh omitted; an older `Retain=false` event of an - omitted condition still arrives (it agrees with the omission, and a client that does not reconcile on RefreshEnd - needs it). The deferral queues, every other Server-object item and every folder item are untouched; the refresh never - waits for a drain. Meter: `otopcua.opcua.condition_events.superseded` (per item and event). Both ConditionRefresh and - ConditionRefresh2, flag on and off. + just dropped (ScadaBridge's snapshot swap drops absent conditions, then a stale active event re-added it). Every + condition event now carries its report sequence. Two paths, chosen per refresh: + - **Nothing deferred (the common case): unchanged and off `Lock`.** The SDK-shaped notifier walk, queued as live + condition nodes; `Lock` is taken only to copy the root-notifier list. It stands only if no deferral episode existed + at any moment of it (deferral count zero before and after, episode number unchanged); otherwise the cut path runs + too, and its replay — later in the same RefreshStart..RefreshEnd — supersedes anything that slipped in between. + - **Deferring: the cut.** ACL/role verdicts first, off `Lock`. Then ONE short `Lock` hold takes the sequence + watermark and a frozen snapshot of each Retain condition straight from the condition map (O(conditions), no tree + walk), and installs per item, under that item's gate, the **cut** plus a "replay pending" marker. The replay is + queued **off `Lock`**, one event per hold of the item gate. While the marker is set, every condition event to the + item takes the gate: one newer than the cut marks its condition so the older frozen entry is dropped instead of + following it; a deferred one at or below the cut is withheld **from that item only** when the refresh re-sent its + condition, or when it is `Retain=true` for a condition the refresh omitted **because it was not retained**. An + older `Retain=false` event of such a condition still arrives (it agrees with the omission), and a condition that was + retained but omitted for the item's **grants** is never withheld (a grant added before the drain lets its events + through; the ACL decides them at delivery as ever). + - The deferral queues, every other Server-object item and every folder item are untouched. A refresh waits for the + drain at most for one in-progress `QueueEvent` on that item (the gate), never for the drain itself; a reporter + holding `Lock` likewise waits at most for one. Each item keeps one gate/cut state for its lifetime (keyed by the + item, so an unsubscribe + resubscribe keeps it), and cuts are cleared when the deferral count returns to zero. + The custom refresh is refused (`ObjectDisposedException`) once the node manager is disposing, and disposal waits + for refreshes in flight, as the base's private operation guard does. Meter: + `otopcua.opcua.condition_events.superseded` (per item and event). ConditionRefresh and ConditionRefresh2, flag + on and off. + - **Measured** (3000 retained conditions, longest `Lock` wait another thread saw during one refresh): the first + #854 cut (`1622d07f`, everything under `Lock`) 95–106 ms idle / 102–127 ms deferring; now 0–3 ms idle / + 23–32 ms deferring (the frozen snapshots). - ⚠️ **Residual: folder (notifier) items.** A refresh of an item on an equipment/device folder can still be followed by older events of a condition. Two sources, neither reachable per item: the SDK's own `MonitoredNode2` channel (up to 4096 events already accepted before the refresh — this exists in the stock SDK too, and in normal operation @@ -912,8 +929,12 @@ Behaviours worth knowing before enabling it: `ConditionRefreshDeferralWireTests` (#854) stalls a folder item's SDK lock so its channel fills without `Lock` held, refreshes the Server-object item (ConditionRefresh and ConditionRefresh2, flag on/off, condition retained or cleared) and pins that nothing older than the replay follows RefreshEnd while the non-refreshing folder item still receives the - whole tail; a probe-driven test pins that a refresh waits for a delivery already past its check (fails without the - item gate); and a stress test runs reports, refreshes and an intermittent stall concurrently. + whole tail (also across an unsubscribe + resubscribe of the item, and the cut is cleared after the drain); a + probe-driven test pins that a refresh waits for a delivery already past its check (fails without the item gate); an + ACL-on test pins that a condition omitted for want of a grant is not withheld once the grant lands; a 3000-condition + test measures the `Lock` hold and pins that neither path queues its replay under `Lock`; and a stress test runs + reports, refreshes and an intermittent stall concurrently (its stream never goes backwards outside a refresh, and + every refresh ends on the newest state delivered). **Model-change and audit events on the Server object are gated too** (Gitea #588). The node manager's own server-wide events — `GeneralModelChangeEvent` (NodeAdded / NodeDeleted / DataTypeChanged) and the diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs index 4458bb50e..95ceb77d4 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs @@ -1,5 +1,6 @@ using System.Collections.Concurrent; using System.Globalization; +using System.Runtime.CompilerServices; using Microsoft.Extensions.Logging; using Microsoft.Extensions.Logging.Abstractions; using Opc.Ua; @@ -620,6 +621,10 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3 { if (disposing) { + // #854 review: refuse new Server-object refreshes and let the ones in flight finish (the base's private + // NodeManagerOperation guard does this for its own paths). + _disposing = true; + SpinWait.SpinUntil(() => Volatile.Read(ref _refreshesInFlight) == 0, TimeSpan.FromSeconds(10)); Volatile.Read(ref _dataPlaneAuthority).Changed -= OnAuthorityChanged; _liveItems.Clear(); _eventItems.Clear(); @@ -1554,19 +1559,30 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3 _logger.LogError(ex, "OtOpcUaNodeManager: failed to report Part 9 condition event for {NodeId}", condition.NodeId); } - IFilterTarget next; + IFilterTarget? next = null; + var episodeEnded = false; lock (_eventReportGate) { CountDeferred(-1, condition); if (queue.Pending.Count == 0) { _deferredReports.Remove(condition); - return; + episodeEnded = _deferredEventReports == 0; + } + else + { + next = queue.Pending.Dequeue(); } - - next = queue.Pending.Dequeue(); } + if (episodeEnded) + { + ClearStaleRefreshCuts(); + return; + } + + if (next is null) return; + try { current = DeliverConditionEventAsync(condition, condition, next).AsTask(); @@ -1641,7 +1657,14 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3 condition.NodeId, MaxDeferredEventsPerCondition); } } - else if (_deferredEventReports == 0 && before > 0 && _episodeWarned) + if (_deferredEventReports == 0 && before > 0) + { + // #854: every deferred event has been delivered, so no refresh cut taken during this episode can match + // anything any more; a new number makes them all stale at once (see RefreshCut.Episode). + _deferralEpisode++; + } + + if (_deferredEventReports == 0 && before > 0 && _episodeWarned) { _logger.LogInformation( "OtOpcUaNodeManager: deferred condition events drained ({Deferred} deferred, {Discarded} discarded this episode)", @@ -1704,21 +1727,94 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3 /// Under . public RefreshCut? Cut; + + /// + /// The refresh whose frozen replay is being queued to the item (0 = none). Set under Lock AND the gate + /// at the cut; cleared under the gate once queued. While set, every condition event delivered to the item + /// takes the gate, and one newer than the cut marks its condition in so the replay + /// does not follow it with an older state. + /// + public volatile int PendingReplay; + + /// Under : the watermark of . + public long PendingWatermark; + + /// Under : conditions an event newer than the pending replay already reached. + public readonly HashSet NewerSinceCut = []; } + private int _refreshGeneration; + + /// + /// One per Server-object item for the item's whole lifetime (#854 review): keyed by + /// the item object itself, so an unsubscribe + resubscribe of the same item between a refresh and the drain keeps its + /// cut and its gate, and a drain holding the state always holds the one a refresh installs into. + /// + private readonly ConditionalWeakTable _allEventsItemStates = new(); + + /// Server-object items currently holding a refresh cut (test observation, #854 review). + internal int ActiveRefreshCuts + { + get + { + var n = 0; + foreach (var item in _allEventsItems.Keys) + { + if (!_allEventsItemStates.TryGetValue(item, out var state)) continue; + lock (state.Gate) n += state.Cut is null ? 0 : 1; + } + + return n; + } + } + + private AllEventsItemState AllEventsItemStateOf(IEventMonitoredItem item) => + _allEventsItemStates.GetValue(item, static _ => new AllEventsItemState()); + + /// + /// Advanced (under the event-report gate) each time the deferral count returns to zero: the end of a deferral + /// episode, after which no event a cut could withhold is still undelivered (#854 review). + /// + private long _deferralEpisode; + /// What a Server-object item's latest ConditionRefresh conveyed (#854). /// Every condition event with a sequence at or below this was reported before the replay was read. - /// The conditions the replay re-sent to the item; the others were Retain=false (or not visible). - private sealed record RefreshCut(long Watermark, HashSet Refreshed) + /// The deferral episode the cut was taken in; once it ends the cut is stale and inert. + /// The conditions the replay re-sent to THIS item. + /// Every condition that was Retain at the watermark, whether or not this item was sent it + /// (one set shared by every item of the refresh). + private sealed record RefreshCut(long Watermark, long Episode, HashSet Refreshed, HashSet RetainedAtCut) { /// - /// Whether the refresh already told the item something at least as new as : a re-sent - /// condition's current state supersedes every older event of it; for a condition the replay OMITTED (it was not - /// retained), an older Retain=true event would wrongly re-add it, while an older Retain=false event agrees with - /// the omission and is still delivered (a client that did not reconcile on RefreshEnd needs it). + /// Whether the refresh already told the item something at least as new as . A re-sent + /// condition: its replayed state supersedes every older event of it. A condition the replay omitted because it + /// was NOT retained: an older Retain=true event would wrongly re-add it, so it is withheld; an older Retain=false + /// event agrees with the omission and is delivered (a client that did not reconcile on RefreshEnd needs it). A + /// condition that WAS retained but was omitted for this item (ACL / role at refresh time): never withheld — the + /// refresh told the item nothing about it, so a grant added before the drain still lets its events through + /// (decided by the ACL at delivery, as ever). /// public bool Supersedes(ConditionEventSnapshot e) => - e.Sequence <= Watermark && (e.Retain || Refreshed.Contains(e.ConditionId)); + e.Sequence <= Watermark + && (Refreshed.Contains(e.ConditionId) || (e.Retain && !RetainedAtCut.Contains(e.ConditionId))); + } + + /// + /// Drops every refresh cut taken in an episode that has ended (#854 review), so a cut does not outlive the events it + /// guards. Called by the drain after the deferral count reached zero, holding no lock; a cut installed concurrently + /// for the new episode is kept. + /// + private void ClearStaleRefreshCuts() + { + var current = Interlocked.Read(ref _deferralEpisode); + foreach (var item in _allEventsItems.Keys) + { + if (!_allEventsItemStates.TryGetValue(item, out var state)) continue; + lock (state.Gate) + { + if (state.Cut is { } cut && cut.Episode != current) state.Cut = null; + } + } } @@ -5362,7 +5458,7 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3 /// unsubscribe arms, which the SDK's MasterNodeManager drives for every create, modify, /// transfer, restore and delete of an all-events item, subscription teardown included. /// - private readonly ConcurrentDictionary _allEventsItems = new(); + private readonly ConcurrentDictionary _allEventsItems = new(); // The SDK's base implementation attaches a Server-object item to the event // fan-out of every one of this manager's root notifiers, so every alarm in the server reached it @@ -5384,7 +5480,7 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3 lock (Lock) { if (unsubscribe) _allEventsItems.TryRemove(monitoredItem, out _); - else _allEventsItems.GetOrAdd(monitoredItem, static _ => new AllEventsItemState()); + else _allEventsItems[monitoredItem] = 0; ReconcileAllEventsMonitoring(); } @@ -5560,7 +5656,7 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3 var reportingSession = (context as ISessionSystemContext)?.SessionId; var deferred = e is ConditionEventSnapshot { Deferred: true } d ? d : null; - foreach (var (item, state) in _allEventsItems) + foreach (var item in _allEventsItems.Keys) { if (e is AuditEventState && !IsAuditDeliverableTo(item)) continue; if (scopes is not null && !IsConditionVisibleTo(item, scopes)) continue; @@ -5572,25 +5668,36 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3 continue; } - if (deferred is null) + // #854: a synchronous event (under Lock) needs the gate only while a refresh's replay is pending, and the + // pending marker is installed under Lock, so an item no refresh has touched has nothing to order. A deferred + // event always takes the gate — its state is created here if need be, so a refresh racing it shares it. + AllEventsItemState? state; + if (deferred is not null) state = AllEventsItemStateOf(item); + else if (!_allEventsItemStates.TryGetValue(item, out state) || state.PendingReplay == 0 || e is not ConditionEventSnapshot) { item.QueueEvent(e); continue; } - // #854: a deferred event reaches this item off Lock, so a ConditionRefresh of the item may have replayed a - // newer state since it was reported. Decided and queued under the item's gate, which the refresh holds while - // it queues its replay and installs its cut — so it lands either before the replay or not at all. + // A deferred event reaches this item off Lock, so a ConditionRefresh of the item may have replayed a newer + // state since it was reported; and while a refresh's frozen replay is being queued, an event newer than it must + // not be followed by it. Both are decided and queued under the item's gate, which the refresh also holds for + // each replayed event it queues — so a deferred event lands before the replay or is judged against the cut. lock (state.Gate) { - if (state.Cut?.Supersedes(deferred) == true) + // A cut from an episode that has ended is stale (only waiting for ClearStaleRefreshCuts). + if (deferred is not null + && state.Cut is { } cut && cut.Episode == Interlocked.Read(ref _deferralEpisode) && cut.Supersedes(deferred)) { Interlocked.Increment(ref _refreshSupersededDeliveries); OtOpcUaTelemetry.ConditionEventDeliveriesSuperseded.Add(1); continue; } - DeferredServerObjectDeliveryProbe?.Invoke(); + if (state.PendingReplay != 0 && e is ConditionEventSnapshot { } ce && ce.Sequence > state.PendingWatermark) + state.NewerSinceCut.Add(ce.ConditionId); + + if (deferred is not null) DeferredServerObjectDeliveryProbe?.Invoke(); item.QueueEvent(e); } } @@ -6174,26 +6281,47 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3 /// public override ServiceResult ConditionRefresh(OperationContext context, IList monitoredItems) { - var direct = new List(); - var serverItems = new List(); - foreach (var item in monitoredItems) + // The base's NodeManagerOperation guard is private; this is its equivalent for the custom paths (#854 review): + // refused once disposal starts, and Dispose waits for the ones in flight. + Interlocked.Increment(ref _refreshesInFlight); + try { - if (item is null) continue; - (item.NodeId == ObjectIds.Server ? serverItems : direct).Add(item); - } + if (_disposing) throw new ObjectDisposedException(GetType().Name); - var enforced = AclEnforced; - var systemContext = SystemContext.Copy(context); - if (direct.Count > 0) + var direct = new List(); + var serverItems = new List(); + foreach (var item in monitoredItems) + { + if (item is null) continue; + (item.NodeId == ObjectIds.Server ? serverItems : direct).Add(item); + } + + var enforced = AclEnforced; + var systemContext = SystemContext.Copy(context); + if (direct.Count > 0) + { + if (enforced) RefreshDirectItems(systemContext, direct); + else base.ConditionRefresh(context, direct); + } + + if (serverItems.Count > 0) RefreshServerObjectItems(systemContext, serverItems, enforced); + return ServiceResult.Good; + } + finally { - if (enforced) RefreshDirectItems(systemContext, direct); - else base.ConditionRefresh(context, direct); + Interlocked.Decrement(ref _refreshesInFlight); } - - if (serverItems.Count > 0) RefreshServerObjectItems(systemContext, serverItems, enforced); - return ServiceResult.Good; } + private int _refreshesInFlight; + private volatile bool _disposing; + + /// + /// Test probe (#854 review): called once per Server-object refresh with whether it took the cut path and whether + /// the calling thread held Lock while it queued the replay. Null in production. + /// + internal Action? ServerObjectRefreshProbe { get; set; } + /// /// ConditionRefresh (and ConditionRefresh2) for Server-object (i=2253) items: each condition under a root /// notifier once, through the item's ACL decision when (#586) — and never overtaken by @@ -6211,84 +6339,224 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3 /// refresh had just dropped re-added, until the newest drained. /// /// - /// The cut. Everything — the watermark (), whether any event is - /// deferred, the replay itself and its queueing — happens in ONE hold of Lock, under which every condition - /// event is reported. So the replay is exactly the state at the watermark: an event at or below it is not newer - /// than the replay, one above it was reported after it. Each item's replay is queued and its - /// installed under the item's , which a deferred - /// delivery to the item also holds (): a deferred event therefore lands - /// before the replay, or is judged against the cut — withheld if superseded (). - /// Only events marked are judged: any other was delivered to - /// Server-object items synchronously under Lock, so it cannot follow a replay it predates. + /// Common case — nothing deferred: unchanged, and off Lock. The replay is the notifier walk the SDK + /// base does, queued as live condition nodes, with Lock taken only to copy the root-notifier list. It is + /// kept only if no deferral episode existed at any moment of it: the deferral count is zero before AND after and + /// the episode number () did not move. Otherwise a deferred event may have reached + /// the item after the replay, so the cut path below runs as well — its second replay, queued later inside the + /// same RefreshStart/RefreshEnd, supersedes anything that slipped in between. + /// + /// + /// Deferring — the cut. In ONE hold of Lock (under which every condition event is reported): the + /// watermark (), the episode, and a frozen snapshot of each Retain + /// condition read from (O(conditions), no tree walk), then queued to each item + /// with its installed under the item's . The ACL / + /// role verdicts are computed BEFORE taking Lock, against the live conditions (they depend on the event + /// type and source, not state). The cut and a "replay pending" marker are installed per item under its + /// inside that Lock hold; the replay is then queued OFF + /// Lock, one event per hold of the gate. While the marker is set every condition event to the item takes + /// the gate: one newer than the cut marks its condition so its (older) frozen replay entry is dropped instead of + /// following it; a deferred one at or below the cut is judged against the cut () + /// — it lands before the replay entry or not at all. /// /// /// Per item, never global. Only the refreshed item stops receiving the superseded events; the deferral - /// queue, every other Server-object item and every folder item still get them, in order. The refresh never - /// waits for a drain. No cut is kept while nothing is deferred (the common case). + /// queue, every other Server-object item and every folder item still get them, in order. A refresh waits for a + /// drain at most for one in-progress QueueEvent on that item (the gate), never for the drain itself. + /// Cuts die with their episode (). /// /// /// Lock order is Lock → item gate → the SDK item's own lock, the same as a synchronous report - /// (Lock → ); the drain takes the item gate without Lock, and - /// nothing under an item gate takes Lock or the event-report gate. + /// (Lock → ); the drain and the replay queueing take the item gate + /// without Lock, and nothing under an item gate takes Lock or the event-report gate. A gate is held + /// for one QueueEvent at a time, so a reporter holding Lock waits at most for one. /// /// /// With enforcement off this replaces the SDK base's Server-object loop with the same result: the base queued /// one copy per root notifier a condition is wired to, which the item's queue collapsed to one (they share an /// EventId); here it is one per condition. A replayed event that names no condition is withheld only under - /// enforcement (fail closed), as before. + /// enforcement (fail closed), as before. The cut path replays from , which holds + /// every condition this manager materialises (all of them sit under a root notifier); branches are replayed as + /// the SDK's ConditionState.ConditionRefresh lists them. /// /// private void RefreshServerObjectItems(ServerSystemContext systemContext, List items, bool enforced) { - lock (Lock) + long episodeBefore; + bool deferringBefore; + lock (_eventReportGate) { - var watermark = Interlocked.Read(ref _conditionEventSequence); - bool deferring; - lock (_eventReportGate) deferring = _deferredEventReports > 0; + episodeBefore = _deferralEpisode; + deferringBefore = _deferredEventReports > 0; + } + if (!deferringBefore) + { + List notifiers; + lock (Lock) notifiers = RootNotifiers?.ToList() ?? []; var events = new List(); - foreach (var notifier in RootNotifiers?.ToList() ?? []) notifier.ConditionRefresh(systemContext, events, true); - - // One event per condition, with its scopes resolved once. - var replay = new List<(IFilterTarget Event, NodeState? Condition, List? Scopes)>(); - var seen = new HashSet(System.Collections.Generic.ReferenceEqualityComparer.Instance); - foreach (var e in events) - { - if (ConditionOf(e) is not { } condition) - { - if (!enforced) replay.Add((e, null, null)); - continue; - } - - if (!seen.Add(condition)) continue; - replay.Add((e, condition, enforced ? ConditionVisibilityScopes(condition) : null)); - } + foreach (var notifier in notifiers) notifier.ConditionRefresh(systemContext, events, true); + var replay = DedupeReplay(events, enforced); + ServerObjectRefreshProbe?.Invoke(false, Monitor.IsEntered(Lock)); foreach (var item in items) { - var queued = new List(); - var refreshed = new HashSet(); - foreach (var (e, condition, scopes) in replay) + // A cut-path replay of this item still being queued by a concurrent refresh is older than this live one: + // stop it, so it cannot follow this replay with frozen states. + if (_allEventsItemStates.TryGetValue(item, out var pending)) + { + lock (pending.Gate) + { + pending.PendingReplay = 0; + pending.NewerSinceCut.Clear(); + } + } + + foreach (var (e, _, scopes) in replay) { if (scopes is not null && !IsConditionVisibleTo(item, scopes)) continue; if (ServiceResult.IsBad(ValidateEventRolePermissions(item, e))) continue; - queued.Add(e); - if (condition is not null) refreshed.Add(condition.NodeId); + item.QueueEvent(e); } + } - if (!_allEventsItems.TryGetValue(item, out var state)) + lock (_eventReportGate) + { + if (_deferralEpisode == episodeBefore && _deferredEventReports == 0) return; + } + } + + RefreshServerObjectItemsWithCut(systemContext, items, enforced); + } + + /// One replay entry per condition (the first the walk produced), with its ACL scopes resolved once. + private List<(IFilterTarget Event, NodeState? Condition, List? Scopes)> DedupeReplay( + List events, bool enforced) + { + var replay = new List<(IFilterTarget Event, NodeState? Condition, List? Scopes)>(); + var seen = new HashSet(System.Collections.Generic.ReferenceEqualityComparer.Instance); + foreach (var e in events) + { + if (ConditionOf(e) is not { } condition) + { + if (!enforced) replay.Add((e, null, null)); + continue; + } + + if (!seen.Add(condition)) continue; + replay.Add((e, condition, enforced ? ConditionVisibilityScopes(condition) : null)); + } + + return replay; + } + + /// The cut path of — see there. + private void RefreshServerObjectItemsWithCut(ServerSystemContext systemContext, List items, bool enforced) + { + // ACL / role verdicts per (item, condition) before Lock: they depend on the event type, source and grants only. + List conditions; + lock (Lock) conditions = [.. _alarmConditions.Values]; + var known = new HashSet(conditions, System.Collections.Generic.ReferenceEqualityComparer.Instance); + var allowed = new HashSet<(IEventMonitoredItem, AlarmConditionState)>(); + foreach (var condition in conditions) + { + var scopes = enforced ? ConditionVisibilityScopes(condition) : null; + foreach (var item in items) + { + if (scopes is not null && !IsConditionVisibleTo(item, scopes)) continue; + if (ServiceResult.IsBad(ValidateEventRolePermissions(item, condition))) continue; + allowed.Add((item, condition)); + } + } + + var states = items.Select(AllEventsItemStateOf).ToList(); + var perItem = new List<(IFilterTarget Event, AlarmConditionState Condition)>[items.Count]; + var generation = Interlocked.Increment(ref _refreshGeneration); + lock (Lock) + { + var watermark = Interlocked.Read(ref _conditionEventSequence); + long episode; + bool deferring; + lock (_eventReportGate) + { + episode = _deferralEpisode; + deferring = _deferredEventReports > 0; + } + + // Frozen at the watermark: the replay is exactly the state every event at or below it describes. + var retained = new HashSet(); + var replay = new List<(IFilterTarget Event, AlarmConditionState Condition)>(); + var branches = new List(); + foreach (var condition in _alarmConditions.Values) + { + if (condition.Retain?.Value != true) continue; + retained.Add(condition.NodeId); + branches.Clear(); + condition.ConditionRefresh(SystemContext, branches, true); + foreach (var target in branches) { - // Not (or no longer) subscribed: no deferred delivery can reach it, so there is nothing to cut. - foreach (var e in queued) item.QueueEvent(e); - continue; + if (target is not BaseInstanceState instance) continue; + var snapshot = new InstanceStateSnapshot(); + snapshot.Initialize(SystemContext, instance); + replay.Add((snapshot, condition)); + } + } + + for (var i = 0; i < items.Count; i++) + { + var item = items[i]; + var entries = new List<(IFilterTarget Event, AlarmConditionState Condition)>(); + var refreshed = new HashSet(); + foreach (var (e, condition) in replay) + { + // A condition materialised since the verdicts were taken is decided here (rare). + if (!allowed.Contains((item, condition))) + { + if (known.Contains(condition)) continue; + if (enforced && !IsConditionVisibleTo(item, ConditionVisibilityScopes(condition))) continue; + if (ServiceResult.IsBad(ValidateEventRolePermissions(item, e))) continue; + } + + entries.Add((e, condition)); + refreshed.Add(condition.NodeId); } + perItem[i] = entries; + var state = states[i]; lock (state.Gate) { - foreach (var e in queued) item.QueueEvent(e); - // A cut is only needed while an event at or below the watermark may still be delivered. Refreshes are - // serialised by Lock, so a later one always carries the later watermark and simply replaces it. - state.Cut = deferring ? new RefreshCut(watermark, refreshed) : null; + state.Cut = deferring ? new RefreshCut(watermark, episode, refreshed, retained) : null; + state.PendingWatermark = watermark; + state.NewerSinceCut.Clear(); + state.PendingReplay = generation; + } + } + } + + // Queued off Lock, one replayed event per hold of the item gate: an event that reached the item since the cut is + // newer than the frozen replay of its condition, so that replay entry is dropped rather than queued after it (the + // newer event is already inside RefreshStart..RefreshEnd). A later refresh of the item supersedes this one. + ServerObjectRefreshProbe?.Invoke(true, Monitor.IsEntered(Lock)); + for (var i = 0; i < items.Count; i++) + { + var item = items[i]; + var state = states[i]; + foreach (var (e, condition) in perItem[i]) + { + lock (state.Gate) + { + if (state.PendingReplay != generation) break; + if (state.NewerSinceCut.Contains(condition.NodeId)) continue; + item.QueueEvent(e); + } + } + + lock (state.Gate) + { + if (state.PendingReplay == generation) + { + state.PendingReplay = 0; + state.NewerSinceCut.Clear(); } } } diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/ConditionRefreshDeferralWireTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/ConditionRefreshDeferralWireTests.cs index b1d456e98..29886a6bd 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/ConditionRefreshDeferralWireTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/ConditionRefreshDeferralWireTests.cs @@ -9,6 +9,7 @@ using Xunit; using ZB.MOM.WW.OtOpcUa.Commons.OpcUa; using ZB.MOM.WW.OtOpcUa.Configuration.Entities; using ZB.MOM.WW.OtOpcUa.Configuration.Enums; +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; using ZB.MOM.WW.OtOpcUa.Core.Authorization; using MonitoredItem = Opc.Ua.Client.MonitoredItem; @@ -40,15 +41,21 @@ public sealed class ConditionRefreshDeferralWireTests /// (so the refresh omits it and its final Retain=false event must still arrive). /// /// ConditionRefresh2 on the single item instead of ConditionRefresh on the subscription. + /// + /// The Server-object item is unsubscribed and resubscribed (the SDK drives this on a monitoring-mode change) + /// between the refresh and the drain: its cut must survive (#854 review — it lived in the subscription map entry). + /// [Theory] - [InlineData(false, true, false)] - [InlineData(true, true, false)] - [InlineData(false, false, false)] - [InlineData(true, false, false)] - [InlineData(false, true, true)] - [InlineData(true, false, true)] + [InlineData(false, true, false, false)] + [InlineData(true, true, false, false)] + [InlineData(false, false, false, false)] + [InlineData(true, false, false, false)] + [InlineData(false, true, true, false)] + [InlineData(true, false, true, false)] + [InlineData(false, true, false, true)] + [InlineData(true, false, false, true)] public async Task Deferred_events_older_than_a_refresh_never_reach_the_refreshed_server_object_item_after_RefreshEnd( - bool enforced, bool finalRetained, bool refresh2) + bool enforced, bool finalRetained, bool refresh2, bool resubscribe) { var ct = TestContext.Current.CancellationToken; await using var rig = await ServerEventRig.StartAsync("Refresh854", encrypted: false, ct); @@ -100,10 +107,21 @@ public sealed class ConditionRefreshDeferralWireTests refresh.Elapsed.ShouldBeLessThan(TimeSpan.FromSeconds(10), "ConditionRefresh must not wait for the stalled drain"); await RecordedEvents.WaitUntilAsync(() => serverEvents.HasType(onServer, ObjectTypeIds.RefreshEndEventType)); rig.Nm.DeferredEventReports.ShouldBeGreaterThan(0, "the deferred events were still waiting at RefreshEnd"); + rig.Nm.ActiveRefreshCuts.ShouldBe(1); + if (resubscribe) + { + var serverItem = FindAllEventsItem(rig.Nm, onServer.Status.Id); + rig.Nm.SubscribeToAllEvents(null!, serverSubscription.Id, serverItem, unsubscribe: true); + rig.Nm.SubscribeToAllEvents(null!, serverSubscription.Id, serverItem, unsubscribe: false); + } + stall.Release(); } await RecordedEvents.WaitUntilAsync(() => rig.Nm.DeferredEventReports == 0); + // #854 review: a cut does not outlive its deferral episode. + await RecordedEvents.WaitUntilAsync(() => rig.Nm.ActiveRefreshCuts == 0); + rig.Nm.ActiveRefreshCuts.ShouldBe(0, "the cut is cleared when the deferral count returns to zero"); // Positive control: a later transition is delivered (and everything deferred before it has been decided). rig.Project(native: true, Active("after")); await RecordedEvents.WaitUntilAsync(() => serverEvents.Has(onServer, "after")); @@ -149,6 +167,144 @@ public sealed class ConditionRefreshDeferralWireTests folderEvents.Has(onA, "final").ShouldBeTrue(); } + /// + /// #854 review (ACL on): a condition the refresh omitted for the item because of its grants — not because it was + /// cleared — must not have its deferred events withheld: a grant added before the drain lets them through, and the + /// item must end on the condition's newest state (it would otherwise miss an active alarm entirely). + /// + [Fact] + public async Task Deferred_events_of_a_condition_the_refresh_omitted_for_want_of_a_grant_are_not_withheld() + { + var ct = TestContext.Current.CancellationToken; + await using var rig = await ServerEventRig.StartAsync("Refresh854Acl", encrypted: false, ct); + rig.KeyRoles = []; + rig.MaterialiseNative(); + var anonymousOnA = ServerEventRig.EquipmentGrant(EqA, NodePermissions.ReadOnly); + rig.Enforce(true, [ServerEventRig.Key()], anonymousOnA); + + // The stalled folder subscriber is anonymous (granted EqA); the refreshing key session holds no grant yet. + using var folderSession = await rig.ConnectAsync(anonymous: true, ct); + var folderEvents = new RecordedEvents(); + var folderSubscription = await RecordedEvents.SubscribeAsync(folderSession, folderEvents, ct); + var onA = RecordedEvents.AddEventItem(folderSubscription, ServerEventRig.UnsNodeId(folderSession, EqA)); + await folderSubscription.ApplyChangesAsync(ct); + + using var serverSession = await rig.ConnectAsync(anonymous: false, ct); + var serverEvents = new RecordedEvents(); + var serverSubscription = await RecordedEvents.SubscribeAsync(serverSession, serverEvents, ct); + var onServer = RecordedEvents.AddEventItem(serverSubscription, ObjectIds.Server); + await serverSubscription.ApplyChangesAsync(ct); + + rig.Project(native: true, Active("seed")); + await RecordedEvents.WaitUntilAsync(() => folderEvents.Has(onA, "seed")); + serverEvents.Has(onServer, "seed").ShouldBeFalse("the key holds no grant yet"); + await folderSubscription.SetPublishingModeAsync(false, ct); + + using (var stall = ItemLockStall.Hold(FindServerItem(rig.Nm, onA.Status.Id))) + { + for (var i = 0; i < FloodSize; i++) rig.Project(native: true, Active($"f{i}")); + rig.Project(native: true, Active("final")); + rig.Nm.DeferredEventReports.ShouldBeGreaterThan(0); + + await serverSession.CallAsync( + ObjectTypeIds.ConditionType, MethodIds.ConditionType_ConditionRefresh, ct, serverSubscription.Id); + await RecordedEvents.WaitUntilAsync(() => serverEvents.HasType(onServer, ObjectTypeIds.RefreshEndEventType)); + + // The grant arrives while the condition's events still wait. + rig.Enforce(true, [ServerEventRig.Key()], anonymousOnA, KeyEquipmentGrant(EqA)); + stall.Release(); + } + + await RecordedEvents.WaitUntilAsync(() => rig.Nm.DeferredEventReports == 0); + await RecordedEvents.WaitUntilAsync(() => serverEvents.Has(onServer, "final")); + var stream = serverEvents.Of(onServer).ToList(); + var end = stream.FindIndex(f => f[RecordedEvents.EventTypeField].GetNodeId() == ObjectTypeIds.RefreshEndEventType); + stream.Take(end).Count(f => f[RecordedEvents.AckedField].TryGetValue(out bool _)) + .ShouldBe(0, "the refresh omitted the condition: no grant"); + var after = stream.Skip(end + 1) + .Where(f => f[RecordedEvents.AckedField].TryGetValue(out bool _)) + .Select(RecordedEvents.Message) + .ToList(); + after.ShouldNotBeEmpty(); + after[^1].ShouldBe("final", "the item ends on the newest state of a condition it was never refreshed"); + after.Count(m => m!.StartsWith('f') && m != "final").ShouldBeGreaterThan(0, "its deferred events were delivered, not withheld"); + } + + /// + /// #854 review: the cut path queues its frozen replay OFF Lock. An event newer than the cut that reaches the + /// item in that window (here a synchronously delivered condition, reported from the probe between the cut and the + /// queueing) must not be followed by its condition's older frozen replay entry. + /// + [Fact] + public async Task An_event_newer_than_the_cut_is_never_followed_by_its_conditions_older_frozen_replay() + { + var ct = TestContext.Current.CancellationToken; + await using var rig = await ServerEventRig.StartAsync("Refresh854Pending", encrypted: false, ct); + rig.KeyRoles = []; + rig.MaterialiseNative(); + rig.Enforce(false, [ServerEventRig.Key()], KeyEquipmentGrant(EqA)); + // A second condition under the device folder only: no folder item, so its events are always synchronous. + const string Free = ServerEventRig.RawDeviceFolder + "/free"; + rig.Sink.MaterialiseAlarmCondition(Free, ServerEventRig.RawDeviceFolder, "free", "OffNormalAlarm", 700, AddressSpaceRealm.Raw, isNative: true); + rig.Sink.WriteAlarmCondition(Free, Active("x1"), DateTime.UtcNow, AddressSpaceRealm.Raw); + + using var folderSession = await rig.ConnectAsync(anonymous: false, ct); + var folderEvents = new RecordedEvents(); + var folderSubscription = await RecordedEvents.SubscribeAsync(folderSession, folderEvents, ct); + var onA = RecordedEvents.AddEventItem(folderSubscription, ServerEventRig.UnsNodeId(folderSession, EqA)); + await folderSubscription.ApplyChangesAsync(ct); + + using var serverSession = await rig.ConnectAsync(anonymous: false, ct); + var serverEvents = new RecordedEvents(); + var serverSubscription = await RecordedEvents.SubscribeAsync(serverSession, serverEvents, ct); + var onServer = RecordedEvents.AddEventItem(serverSubscription, ObjectIds.Server); + await serverSubscription.ApplyChangesAsync(ct); + + rig.Project(native: true, Active("seed")); + await RecordedEvents.WaitUntilAsync(() => folderEvents.Has(onA, "seed") && serverEvents.Has(onServer, "seed")); + await folderSubscription.SetPublishingModeAsync(false, ct); + + var injected = 0; + rig.Nm.ServerObjectRefreshProbe = (cut, _) => + { + if (cut && Interlocked.Exchange(ref injected, 1) == 0) + rig.Sink.WriteAlarmCondition(Free, Active("x2"), DateTime.UtcNow, AddressSpaceRealm.Raw); + }; + + using (var stall = ItemLockStall.Hold(FindServerItem(rig.Nm, onA.Status.Id))) + { + for (var i = 0; i < FloodSize; i++) rig.Project(native: true, Active($"f{i}")); + rig.Nm.DeferredEventReports.ShouldBeGreaterThan(0, "the cut path runs only while something is deferred"); + await serverSession.CallAsync( + ObjectTypeIds.ConditionType, MethodIds.ConditionType_ConditionRefresh, ct, serverSubscription.Id); + await RecordedEvents.WaitUntilAsync(() => serverEvents.HasType(onServer, ObjectTypeIds.RefreshEndEventType)); + stall.Release(); + } + + rig.Nm.ServerObjectRefreshProbe = null; + Volatile.Read(ref injected).ShouldBe(1); + var stream = serverEvents.Of(onServer).ToList(); + var end = stream.FindIndex(f => f[RecordedEvents.EventTypeField].GetNodeId() == ObjectTypeIds.RefreshEndEventType); + var free = stream.Take(end) + .Select(RecordedEvents.Message) + .Where(m => m is "x1" or "x2") + .ToList(); + free.ShouldContain("x2"); + free[^1].ShouldBe("x2", "the older frozen replay of the condition must not follow its newer event"); + } + + /// #854 review: the custom Server-object refresh is refused once the node manager is disposed, like the base's. + [Fact] + public async Task A_refresh_after_disposal_is_refused() + { + var ct = TestContext.Current.CancellationToken; + var rig = await ServerEventRig.StartAsync("Refresh854Dispose", encrypted: false, ct); + var nm = rig.Nm; + await rig.DisposeAsync(); + nm.Dispose(); + Should.Throw(() => nm.ConditionRefresh(null!, Array.Empty())); + } + /// /// The race the item gate closes, made deterministic: a deferred event is between its supersession check and its /// queueing to the Server-object item when the refresh arrives. The refresh must wait for that one delivery (so @@ -294,6 +450,7 @@ public sealed class ConditionRefreshDeferralWireTests } }, stop.Token, TaskCreationOptions.LongRunning, TaskScheduler.Default); + var refusedAsInProgress = 0; for (var r = 0; r < 40; r++) { var call = r % 3 == 2 @@ -301,7 +458,17 @@ public sealed class ConditionRefreshDeferralWireTests serverSubscription.Id, onServer.Status.Id) : serverSession.CallAsync(ObjectTypeIds.ConditionType, MethodIds.ConditionType_ConditionRefresh, ct, serverSubscription.Id); - await call.WaitAsync(TimeSpan.FromSeconds(20), ct); + try + { + await call.WaitAsync(TimeSpan.FromSeconds(20), ct); + } + catch (ServiceResultException ex) when (ex.StatusCode == StatusCodes.BadRefreshInProgress) + { + // The SDK answers the Call before the refresh body runs, so under load the previous refresh of this + // subscription may still be in progress; the SDK refuses the overlap (its rule, not ours). + refusedAsInProgress++; + } + await Task.Delay(TimeSpan.FromMilliseconds(80), ct); } @@ -311,16 +478,17 @@ public sealed class ConditionRefreshDeferralWireTests Volatile.Read(ref deferredSeen).ShouldBeGreaterThan(1, "the stress must actually have run with deferred events"); await RecordedEvents.WaitUntilAsync(() => rig.Nm.DeferredEventReports == 0); - var last = $"s{Volatile.Read(ref counter)}"; + var newest = $"s{Volatile.Read(ref counter)}"; await folderSubscription.SetPublishingModeAsync(true, ct); - await RecordedEvents.WaitUntilAsync(() => serverEvents.Has(onServer, last)); + await RecordedEvents.WaitUntilAsync(() => serverEvents.Has(onServer, newest)); await Task.Delay(TimeSpan.FromMilliseconds(500), ct); - // Walk the Server-object stream: after each RefreshEnd, a condition event must not be older than what that - // refresh replayed (the refresh replays the then-current state, so the floor is its replayed counter). - var floor = -1; + // Walk the Server-object stream (one condition). Outside a refresh its states never go backwards; and each refresh + // ends on the newest state the item has seen — its replay is never older than an event already delivered, and + // nothing older than it follows RefreshEnd. + var max = -1; + var last = -1; var replaying = false; - var replayedInThisRefresh = -1; var violations = new List(); foreach (var f in serverEvents.Of(onServer)) { @@ -328,33 +496,139 @@ public sealed class ConditionRefreshDeferralWireTests if (type == ObjectTypeIds.RefreshStartEventType) { replaying = true; - replayedInThisRefresh = -1; continue; } if (type == ObjectTypeIds.RefreshEndEventType) { replaying = false; - floor = Math.Max(floor, replayedInThisRefresh); + if (last < max) violations.Add($"a refresh ended on s{last} after s{max} had been delivered"); continue; } if (!f[RecordedEvents.AckedField].TryGetValue(out bool _)) continue; var n = int.Parse(RecordedEvents.Message(f)![1..], System.Globalization.CultureInfo.InvariantCulture); - if (replaying) - { - replayedInThisRefresh = n; - continue; - } - - if (n < floor) violations.Add($"s{n} after a refresh that replayed s{floor}"); + if (!replaying && n < max) violations.Add($"s{n} delivered after s{max}"); + max = Math.Max(max, n); + last = n; } violations.ShouldBeEmpty(); + refusedAsInProgress.ShouldBeLessThan(40, "most refreshes must actually have run"); rig.Nm.RefreshSupersededDeliveries.ShouldBeGreaterThan(0, "some refresh must actually have overtaken deferred events"); RecordedEvents.Message(serverEvents.Of(onServer).Last(f => f[RecordedEvents.AckedField].TryGetValue(out bool _))) - .ShouldBe(last, "the stream still converges on the newest state"); - await RecordedEvents.WaitUntilAsync(() => folderEvents.Has(onA, last)); + .ShouldBe(newest, "the stream still converges on the newest state"); + await RecordedEvents.WaitUntilAsync(() => folderEvents.Has(onA, newest)); + } + + /// + /// Review of 1622d07f: a Server-object refresh must not hold the node manager's Lock across a walk of a large + /// address space — ScadaBridge refreshes on every reconnect, and any session may loop ConditionRefresh. Measures + /// the longest any other thread waits for Lock while a refresh of 3000 retained conditions runs, with + /// nothing deferred (the common case) and with a deferral in progress, and pins that the common case does not + /// build or queue its replay under Lock. + /// + /// Whether a deferral episode is in progress when the refresh runs. + [Theory] + [InlineData(false)] + [InlineData(true)] + public async Task A_refresh_of_a_large_address_space_keeps_the_node_manager_lock_hold_short(bool deferring) + { + const int Conditions = 3000; + var ct = TestContext.Current.CancellationToken; + await using var rig = await ServerEventRig.StartAsync("Refresh854Lock", encrypted: false, ct); + rig.KeyRoles = []; + rig.MaterialiseNative(); + rig.Enforce(false, [ServerEventRig.Key()], KeyEquipmentGrant(EqA)); + for (var i = 0; i < Conditions; i++) + { + var id = $"{ServerEventRig.RawDeviceFolder}/c{i}"; + rig.Sink.MaterialiseAlarmCondition(id, ServerEventRig.RawDeviceFolder, $"c{i}", "OffNormalAlarm", 700, AddressSpaceRealm.Raw, isNative: true); + rig.Sink.WriteAlarmCondition(id, Active($"c{i}"), DateTime.UtcNow, AddressSpaceRealm.Raw); + } + + using var folderSession = await rig.ConnectAsync(anonymous: false, ct); + var folderEvents = new RecordedEvents(); + var folderSubscription = await RecordedEvents.SubscribeAsync(folderSession, folderEvents, ct); + var onA = RecordedEvents.AddEventItem(folderSubscription, ServerEventRig.UnsNodeId(folderSession, EqA)); + await folderSubscription.ApplyChangesAsync(ct); + + using var serverSession = await rig.ConnectAsync(anonymous: false, ct); + var serverEvents = new RecordedEvents(); + var serverSubscription = await RecordedEvents.SubscribeAsync(serverSession, serverEvents, ct); + var onServer = RecordedEvents.AddEventItem(serverSubscription, ObjectIds.Server); + onServer.QueueSize = 10_000; + await serverSubscription.ApplyChangesAsync(ct); + + rig.Project(native: true, Active("seed")); + await RecordedEvents.WaitUntilAsync(() => folderEvents.Has(onA, "seed") && serverEvents.Has(onServer, "seed")); + await folderSubscription.SetPublishingModeAsync(false, ct); + + var stall = deferring ? ItemLockStall.Hold(FindServerItem(rig.Nm, onA.Status.Id)) : null; + try + { + if (deferring) + { + for (var i = 0; i < FloodSize; i++) rig.Project(native: true, Active($"f{i}")); + rig.Nm.DeferredEventReports.ShouldBeGreaterThan(0); + } + + var probes = new System.Collections.Concurrent.ConcurrentQueue<(bool Cut, bool LockHeld)>(); + rig.Nm.ServerObjectRefreshProbe = (cut, held) => probes.Enqueue((cut, held)); + var waits = new List(); + for (var round = 0; round < 5; round++) + { + var ends = serverEvents.Of(onServer).Count(f => f[RecordedEvents.EventTypeField].GetNodeId() == ObjectTypeIds.RefreshEndEventType); + using var stop = new CancellationTokenSource(); + var maxWait = 0.0; + var prober = Task.Factory.StartNew(() => + { + var sw = new Stopwatch(); + while (!stop.IsCancellationRequested) + { + sw.Restart(); + lock (rig.Nm.Lock) { } + maxWait = Math.Max(maxWait, sw.Elapsed.TotalMilliseconds); + Thread.Yield(); + } + }, CancellationToken.None, TaskCreationOptions.LongRunning, TaskScheduler.Default); + + await serverSession.CallAsync( + ObjectTypeIds.ConditionType, MethodIds.ConditionType_ConditionRefresh, ct, serverSubscription.Id); + await RecordedEvents.WaitUntilAsync(() => serverEvents.Of(onServer) + .Count(f => f[RecordedEvents.EventTypeField].GetNodeId() == ObjectTypeIds.RefreshEndEventType) > ends); + await stop.CancelAsync(); + await prober; + waits.Add(maxWait); + } + + TestContext.Current.TestOutputHelper?.WriteLine( + $"#854 lock-hold deferring={deferring} conditions={Conditions}: max Lock wait per refresh (ms) = " + + string.Join(", ", waits.Select(w => w.ToString("F1", System.Globalization.CultureInfo.InvariantCulture)))); + // Neither path queues its replay under Lock; only a deferral makes a refresh take the cut path. + probes.ShouldNotBeEmpty(); + probes.ShouldAllBe(p => !p.LockHeld); + if (deferring) probes.ShouldContain(p => p.Cut); + else probes.ShouldAllBe(p => !p.Cut); + } + finally + { + stall?.Dispose(); + } + } + + /// The server-side Server-object (all-events) item behind a client item. + private static IEventMonitoredItem FindAllEventsItem(OtOpcUaNodeManager nm, uint id) + { + var map = (IDictionary)typeof(OtOpcUaNodeManager) + .GetField("_allEventsItems", BindingFlags.Instance | BindingFlags.NonPublic)! + .GetValue(nm)!; + foreach (var key in map.Keys) + { + if (key is IEventMonitoredItem item && item.Id == id) return item; + } + + throw new InvalidOperationException($"no server-side all-events item {id}"); } /// The server-side event item behind a client item, from the node manager's direct-item map.