From d2b1545659bd1420764ce16d1fcee76eece0bd61 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Wed, 11 Nov 2026 12:31:02 -0500 Subject: [PATCH] =?UTF-8?q?fix(opcua):=20#855=20review=201=20=E2=80=94=20r?= =?UTF-8?q?e-decide=20frozen=20replay=20entries;=20a=20failed=20delivery?= =?UTF-8?q?=20is=20not=20a=20deferral?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - QueueFrozenReplay re-asks each item's verdict right before each entry's gate hold (Server-object and folder cut paths): a grant revoked between the pre-Lock verdicts and the replay no longer lets the replay through. A denied entry is also removed from the cut's re-sent set. - ReportEventWithoutBlocking logs and returns on a delivery that faulted inline instead of booking it as a deferral episode (which ran the drain and ClearStaleRefreshCuts under the event-report gate while Lock was held). - OnConditionReportEvent isolates failures per Server-object item. - Tests: revocation between cut and replay (folder + Server object), throwing hop not booked as a deferral; production-overload sampling inside Lock. Claude-Session: https://claude.ai/code/session_01GT67AD99zQQqpLCVm3p7xk --- docs/security.md | 6 +- .../OtOpcUaNodeManager.cs | 79 ++++++++++-- .../FolderConditionRefreshWireTests.cs | 112 +++++++++++++++++- 3 files changed, 182 insertions(+), 15 deletions(-) diff --git a/docs/security.md b/docs/security.md index 1e09644b1..1d02687b5 100644 --- a/docs/security.md +++ b/docs/security.md @@ -924,7 +924,11 @@ Behaviours worth knowing before enabling it: `QueueSize`) and the verdict runs on the reporter's thread, which already holds `Lock` — so a burst under one `Lock` hold completes with nothing deferred, and the #838 deadlock shape is gone for condition events. Folder items therefore see condition events in report order. The cost moves with it: the per-event verdict and filter for - folder items now run under `Lock` (as for Server-object items), where the SDK used to cache the verdict off it. + folder items now run under `Lock` (as for Server-object items), where the SDK used to cache the verdict off it — + including the SDK item queue's own per-event costs (its duplicate scan, and at a full queue a list shift), so many + folder items on one notifier whose clients stop publishing make each report proportionally slower under `Lock`. + With auditing on (off in this repo), a condition's SDK method **audit** event still takes the folder's channel, so + at a folder item it may now arrive after the condition event that method produced (they were one FIFO before). - **The deferral queue and the cut stay as the safety net**, for a hop that is not ours (none exists today: conditions are never event notifiers and variables cannot host event items) — and folder items share the cut: a `ConditionRefresh`/`ConditionRefresh2` of a folder item takes the same two paths as the Server object (plain walk diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs index c3d96cc82..56e116e51 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs @@ -1529,6 +1529,16 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3 if (delivery.IsCompletedSuccessfully) return; + // #855 review: a delivery that FAILED inline is over, not suspended. Booking it as a deferral used to open and close + // a deferral episode synchronously (moving the episode a concurrent plain refresh checks, and clearing cuts under + // the event-report gate while Lock was held). Only an incomplete delivery is deferred. + if (delivery.IsCompleted) + { + _logger.LogError(delivery.AsTask().Exception?.GetBaseException(), + "OtOpcUaNodeManager: failed to report Part 9 condition event for {NodeId}", condition.NodeId); + return; + } + lock (_eventReportGate) { var queue = new DeferredConditionReports(); @@ -1901,6 +1911,9 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3 /// private long _deferralEpisode; + /// Deferral episodes ended since start (test observation, #855 review). + internal long DeferralEpisodes => Interlocked.Read(ref _deferralEpisode); + /// What a Server-object item's latest ConditionRefresh conveyed (#854). /// Every condition event with a sequence at or below this was reported before the replay was read. /// The deferral episode the cut was taken in; once it ends the cut is stale and inert. @@ -5782,17 +5795,26 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3 foreach (var item in _allEventsItems.Keys) { - if (e is AuditEventState && !IsAuditDeliverableTo(item)) continue; - if (scopes is not null && !IsConditionVisibleTo(item, scopes)) continue; - if (ServiceResult.IsBad(ValidateEventRolePermissions(item, e))) continue; - if (reportingSession is { IsNull: false } reportingId - && item.Session?.Id is { } own - && !own.Equals(reportingId)) + try { - continue; - } + if (e is AuditEventState && !IsAuditDeliverableTo(item)) continue; + if (scopes is not null && !IsConditionVisibleTo(item, scopes)) continue; + if (ServiceResult.IsBad(ValidateEventRolePermissions(item, e))) continue; + if (reportingSession is { IsNull: false } reportingId + && item.Session?.Id is { } own + && !own.Equals(reportingId)) + { + continue; + } - QueueConditionEventToItem(item, e, deferred, DeferredServerObjectDeliveryProbe); + QueueConditionEventToItem(item, e, deferred, DeferredServerObjectDeliveryProbe); + } + catch (Exception ex) + { + // #855 review: one item's failure (an item disposed under us, the SDK's operation guard at shutdown) must + // not cost the other Server-object items, nor abort the walk before the folder hops. + _logger.LogWarning(ex, "OtOpcUaNodeManager: failed to deliver a condition event to Server-object item {ItemId}", item.Id); + } } } @@ -6671,7 +6693,24 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3 } ServerObjectRefreshProbe?.Invoke(true, Monitor.IsEntered(Lock)); - QueueFrozenReplay(items, states, perItem.Select(entries => entries.Select(x => (x.Event, x.Condition.NodeId)).ToList()).ToArray(), generation); + // #855 review: the verdicts above were taken before Lock; a deploy may have revoked a grant since. Each replay entry + // is re-decided right before it is queued (scopes resolved once per condition). + var scopeCache = new Dictionary>(); + QueueFrozenReplay( + items, states, perItem.Select(entries => entries.Select(x => (x.Event, x.Condition.NodeId)).ToList()).ToArray(), + generation, + (item, e) => + { + if (enforced) + { + if (ConditionOf(e) is not { } condition) return false; + if (!scopeCache.TryGetValue(condition.NodeId, out var scopes)) + scopeCache[condition.NodeId] = scopes = ConditionVisibilityScopes(condition); + if (!IsConditionVisibleTo(item, scopes)) return false; + } + + return !ServiceResult.IsBad(ValidateEventRolePermissions(item, e)); + }); } finally { @@ -6685,8 +6724,15 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3 /// replay of its condition, so that replay entry is dropped rather than queued after it (the newer event is already /// inside RefreshStart..RefreshEnd). A later refresh of the item supersedes this one. /// + /// + /// The item's verdict for one entry, asked right before the entry's gate hold (never under it: it takes Lock), + /// against the CURRENT generation — the verdicts the cut was built on were taken earlier (#855 review). An entry + /// denied here is dropped AND removed from the cut's re-sent set, so the condition counts as "omitted for the + /// grants" and its later events are left to the ACL rather than withheld as superseded. + /// private static void QueueFrozenReplay( - List items, List states, List<(IFilterTarget Event, NodeId ConditionId)>[] perItem, int generation) + List items, List states, List<(IFilterTarget Event, NodeId ConditionId)>[] perItem, + int generation, Func allowed) { for (var i = 0; i < items.Count; i++) { @@ -6694,9 +6740,17 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3 var state = states[i]; foreach (var (e, conditionId) in perItem[i]) { + if (state.PendingReplay != generation) break; + var permitted = allowed(item, e); lock (state.Gate) { if (state.PendingReplay != generation) break; + if (!permitted) + { + state.Cut?.Refreshed.Remove(conditionId); + continue; + } + if (state.NewerSinceCut.Contains(conditionId)) continue; item.QueueEvent(e); } @@ -6926,7 +6980,8 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2, INodeManager3 } DirectRefreshProbe?.Invoke(true, Monitor.IsEntered(Lock)); - QueueFrozenReplay(items, states, perItem, generation); + // #855 review: re-decided per entry right before queueing — a deploy may have revoked since the verdicts. + QueueFrozenReplay(items, states, perItem, generation, (item, e) => IsDirectRefreshEventAllowed(item, e, enforced)); } finally { diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/FolderConditionRefreshWireTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/FolderConditionRefreshWireTests.cs index b01a93427..0a9bc98b9 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/FolderConditionRefreshWireTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/FolderConditionRefreshWireTests.cs @@ -358,6 +358,110 @@ public sealed class FolderConditionRefreshWireTests keyEvents.Has(keyA, "after-revoke").ShouldBeFalse(); } + /// + /// #855 review: the cut path decides its replay before taking Lock; a grant revoked between that decision + /// and the queueing of the frozen replay must still withhold the replay (it is re-decided right before each entry). + /// The revocation lands from the refresh probe, which runs after the cut is installed and before the replay is queued. + /// + /// Refresh a Server-object item instead of a folder item. + [Theory] + [InlineData(false)] + [InlineData(true)] + public async Task A_grant_revoked_between_the_cut_and_its_replay_withholds_the_replay(bool serverObject) + { + var ct = TestContext.Current.CancellationToken; + await using var rig = await ServerEventRig.StartAsync("Refresh855RevokeCut", encrypted: false, ct); + rig.KeyRoles = []; + rig.MaterialiseNative(); + var anonGrant = ServerEventRig.EquipmentGrant(EqA, NodePermissions.ReadOnly); + var state = rig.Enforce(true, [ServerEventRig.Key()], KeyEquipmentGrant(EqA), anonGrant); + + // An anonymous subscriber keeps EqA a notifier fan-out the stall applies to (and stays granted throughout). + using var anonymous = await rig.ConnectAsync(anonymous: true, ct); + var anonEvents = new RecordedEvents(); + var anonSub = await RecordedEvents.SubscribeAsync(anonymous, anonEvents, ct); + RecordedEvents.AddEventItem(anonSub, ServerEventRig.UnsNodeId(anonymous, EqA)); + await anonSub.ApplyChangesAsync(ct); + + using var keyed = await rig.ConnectAsync(anonymous: false, ct); + var keyEvents = new RecordedEvents(); + var keySub = await RecordedEvents.SubscribeAsync(keyed, keyEvents, ct); + var item = RecordedEvents.AddEventItem(keySub, serverObject ? ObjectIds.Server : ServerEventRig.UnsNodeId(keyed, EqA)); + await keySub.ApplyChangesAsync(ct); + item.Status.Error.ShouldBeNull(); + + rig.Project(native: true, Active("seed")); + await RecordedEvents.WaitUntilAsync(() => keyEvents.Has(item, "seed")); + + var revoked = 0; + void Revoke(bool cut, bool _) + { + if (!cut || Interlocked.Exchange(ref revoked, 1) != 0) return; + state.Apply( + ServerEventRig.Cluster, "rev-2", [ServerEventRig.Key()], [anonGrant], ServerEventRig.ScopePaths(), + ServerEventRig.EquipmentOwnerPaths(), ServerEventRig.ScriptedAlarmOwners(), ServerEventRig.FolderScopes(), + ServerEventRig.TagReferences()); + } + + if (serverObject) rig.Nm.ServerObjectRefreshProbe = Revoke; + else rig.Nm.DirectRefreshProbe = Revoke; + + using (var stall = NotifierStall.Hold(rig.Nm, EqA)) + { + for (var i = 0; i < FloodSize; i++) rig.Project(native: true, Active($"f{i}")); + rig.Nm.DeferredEventReports.ShouldBeGreaterThan(0); + await keyed.CallAsync(ObjectTypeIds.ConditionType, MethodIds.ConditionType_ConditionRefresh, ct, keySub.Id); + await RecordedEvents.WaitUntilAsync(() => keyEvents.HasType(item, ObjectTypeIds.RefreshEndEventType)); + stall.Release(); + } + + rig.Nm.ServerObjectRefreshProbe = null; + rig.Nm.DirectRefreshProbe = null; + Volatile.Read(ref revoked).ShouldBe(1, "the refresh took the cut path"); + await RecordedEvents.WaitUntilAsync(() => rig.Nm.DeferredEventReports == 0); + await Task.Delay(TimeSpan.FromMilliseconds(500), ct); + + var (replayed, after) = Window(keyEvents.Of(item)); + replayed.ShouldBeEmpty("the grant was revoked before the replay was queued"); + after.ShouldBeEmpty("and nothing reaches the item after RefreshEnd either"); + } + + /// + /// #855 review: a delivery that fails inline (here: a hop that throws) is over, not suspended — it must not be booked + /// as a deferral episode (which moved the episode a concurrent plain refresh checks, and cleared cuts under the + /// event-report gate while Lock was held). The Server-object item, served before the failing hop, still gets + /// the event, and the next event is delivered normally. + /// + [Fact] + public async Task A_delivery_that_throws_is_not_booked_as_a_deferral() + { + var ct = TestContext.Current.CancellationToken; + await using var rig = await ServerEventRig.StartAsync("Refresh855Fault", encrypted: false, ct); + rig.KeyRoles = []; + rig.MaterialiseNative(); + rig.Enforce(false, [ServerEventRig.Key()], KeyEquipmentGrant(EqA)); + + using var session = await rig.ConnectAsync(anonymous: false, ct); + var events = new RecordedEvents(); + var subscription = await RecordedEvents.SubscribeAsync(session, events, ct); + var onA = RecordedEvents.AddEventItem(subscription, ServerEventRig.UnsNodeId(session, EqA)); + var onServer = RecordedEvents.AddEventItem(subscription, ObjectIds.Server); + await subscription.ApplyChangesAsync(ct); + + var episodes = rig.Nm.DeferralEpisodes; + rig.Nm.NotifierDeliveryStall = node => + node.NodeId.IdentifierAsString == EqA ? throw new InvalidOperationException("hop fault (test)") : null; + rig.Project(native: true, Active("faulted")); + rig.Nm.NotifierDeliveryStall = null; + + rig.Nm.DeferredEventReports.ShouldBe(0); + rig.Nm.DeferralEpisodes.ShouldBe(episodes, "a failed delivery is not a deferral episode"); + rig.Project(native: true, Active("next")); + await RecordedEvents.WaitUntilAsync(() => + events.Has(onServer, "faulted") && events.Has(onServer, "next") && events.Has(onA, "next")); + events.Has(onA, "faulted").ShouldBeFalse("the faulted hop delivered nothing"); + } + /// /// Production shape, no stall: bursts reported under one Lock hold (an apply pass) and single reports, against /// folder items that refresh repeatedly while their client is slow to publish. Folder items are served per item, so @@ -398,10 +502,14 @@ public sealed class FolderConditionRefreshWireTests { lock (rig.Nm.Lock) { - for (var i = 0; i < 1000; i++) rig.Project(native: true, Active($"s{Interlocked.Increment(ref counter)}")); + for (var i = 0; i < 1000; i++) + { + rig.Project(native: true, Active($"s{Interlocked.Increment(ref counter)}")); + // Sampled inside Lock, after each report: a deferral cannot open and drain unseen (#855 review). + peakDeferred = Math.Max(peakDeferred, rig.Nm.DeferredEventReports); + } } - peakDeferred = Math.Max(peakDeferred, rig.Nm.DeferredEventReports); rig.Project(native: true, Active($"s{Interlocked.Increment(ref counter)}")); Thread.Sleep(2); }