From fbe8aaad9d57c68a2bfcc8aeb30db8ec8394ffc1 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Sun, 16 Aug 2026 04:18:03 -0400 Subject: [PATCH] perf(acl): dedup subscription revalidation by (session, node) --- .../OtOpcUaNodeManager.cs | 87 ++++++- .../NodeManagerStage6EnforcementTests.cs | 242 ++++++++++++++++++ 2 files changed, 326 insertions(+), 3 deletions(-) diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs index 3121f84f..3410abeb 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs @@ -3127,6 +3127,18 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2 ServerSystemContext context, NodeHandle handle, ISampledDataChangeMonitoredItem monitoredItem) { base.OnMonitoredItemCreated(context, handle, monitoredItem); + TrackLiveMonitoredItem(handle, monitoredItem); + } + + /// Starts tracking for revalidation, if we own its node. + /// The handle the item was created against. + /// The item. + /// + /// Split out of so tests can register an item without a client + /// session; the SDK reaches it only through that override. + /// + internal void TrackLiveMonitoredItem(NodeHandle handle, ISampledDataChangeMonitoredItem monitoredItem) + { if (OwnsNamespace(handle.NodeId)) _liveItems[monitoredItem] = new LiveItem(handle); } @@ -3193,6 +3205,20 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2 /// subscription, and inventing one would leave the client's own bookkeeping describing items the /// server had silently discarded. /// + /// + /// One verdict per (session, node), not per item. A session commonly holds several items + /// on the same node — one per attribute, or one per downstream consumer sharing a session — and + /// they all ask the identical question. The pass memoizes the verdict locally so N items on a + /// node cost one call rather than N. + /// The authorizer's own decision cache does not remove this work: it memoizes the grant walk, + /// while the per-call key revocation/expiry check in front of it is deliberately outside that + /// cache and runs every time. + /// + /// + /// The per-item lock (Lock) takes are deliberately left alone. Batching them under one + /// hold was considered and rejected: it would stretch the SDK lock across the + /// QueueValue calls between them, and the attach/detach is what actually needs the lock. + /// /// public void RevalidateMonitoredItems() { @@ -3202,6 +3228,11 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2 var stoodDown = 0; var restored = 0; + // Pass-local, so single-threaded by construction. A concurrent deployment can advance the + // generation mid-pass and leave later items decided on the earlier one — exactly as the + // un-memoized walk already could, and the change that advanced it runs its own pass. + var verdicts = new Dictionary<(string Session, AddressSpaceRealm Realm, string Node), bool>(); + foreach (var (item, tracked) in _liveItems) { // The item carries the identity of the session that created it, which is what must be @@ -3209,9 +3240,8 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2 if (item is not IMonitoredItem live) continue; var nodeId = tracked.Handle.NodeId; - var permitted = _authorizer!.IsAllowedCached( - live.EffectiveIdentity as RoleCarryingUserIdentity, - OpcUaOperation.CreateMonitoredItems, nodeId, RealmOf(nodeId)); + var permitted = IsSubscriptionStillPermitted( + live.EffectiveIdentity as RoleCarryingUserIdentity, nodeId, verdicts); if (permitted) { @@ -3244,6 +3274,57 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2 } } + /// + /// The per-item verdict for , memoized across the pass. + /// + /// The identity of the session that created the item. + /// The node the item monitors. + /// The pass-local memo, keyed by session, realm and node identifier. + /// when the item may keep receiving values. + /// + /// + /// The session key is ApiKeyId ?? Principal — the same derivation + /// uses for its own decision cache, so this memo is + /// never coarser than the cache it sits in front of. The realm is part of the key for the same + /// reason it is part of the authorizer's: the raw and UNS namespaces have independent string + /// identifier spaces, so an identifier alone does not name a node. + /// + /// + /// Anything the key cannot be built from — no role-carrying identity, no principal, a non-string + /// node identifier — falls through to the un-memoized call and keeps its existing verdict. Those + /// are the cases the authorizer denies anyway; memoizing them would only make a deny cheaper. + /// + /// + /// This memo does cover the key expiry check that the authorizer's own cache excludes, but only + /// for the duration of one pass — and a subscription is re-checked at a deployment, never on a + /// timer, so an expiry that lands inside a single pass was already going to be caught by the + /// next deployment rather than by this one. + /// + /// + private bool IsSubscriptionStillPermitted( + RoleCarryingUserIdentity? identity, + NodeId nodeId, + Dictionary<(string Session, AddressSpaceRealm Realm, string Node), bool> verdicts) + { + var realm = RealmOf(nodeId); + + if (identity is null + || (identity.ApiKeyId ?? identity.Principal) is not { Length: > 0 } sessionKey + || nodeId.Identifier is not string identifier) + { + return _authorizer!.IsAllowedCached( + identity, OpcUaOperation.CreateMonitoredItems, nodeId, realm); + } + + var key = (sessionKey, realm, identifier); + if (verdicts.TryGetValue(key, out var memoized)) return memoized; + + var permitted = _authorizer!.IsAllowedCached( + identity, OpcUaOperation.CreateMonitoredItems, nodeId, realm); + verdicts[key] = permitted; + return permitted; + } + /// Emits the one operator-facing summary line per gated request. /// The service being filtered. /// The calling session's identity. diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerStage6EnforcementTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerStage6EnforcementTests.cs index f60d7dd2..9d3a0d51 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerStage6EnforcementTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerStage6EnforcementTests.cs @@ -347,8 +347,102 @@ public sealed class NodeManagerStage6EnforcementTests : IDisposable finally { await host.DisposeAsync(); } } + /// + /// Revalidation asks the authorizer once per distinct (session, node), not once per monitored item. + /// A client that subscribes to the same tag from several items — one per attribute, or one per + /// downstream consumer sharing a session — is the ordinary case, and every one of those items used + /// to pay for its own grant walk on every deployment. + /// + /// + /// + /// The observable is the authorizer's . Evaluate reads the clock + /// exactly once per call, for the key revocation/expiry check that is deliberately outside the + /// decision cache — so counting clock reads counts IsAllowedCached entries, which is the + /// work this dedup removes. Counting the permission evaluator instead would be vacuous: the + /// authorizer's own decision cache already memoizes the grant walk, so it would report one + /// either way. + /// + /// + /// is called directly rather than via + /// an authority flip so the count covers exactly one pass — a flip would fire the pass through + /// OnAuthorityChanged as well and the assertion would be about two. + /// + /// + [Fact] + public async Task Revalidation_evaluates_one_verdict_per_session_and_node() + { + var (host, server) = await BootAsync(); + try + { + var nm = server.NodeManager!; + var clock = new CountingTimeProvider(new DateTime(2026, 8, 15, 12, 0, 0, DateTimeKind.Utc)); + nm.Authorizer = SubscribeAuthorizer(clock); + + var nodeId = Variable(nm, "Plant/Modbus/dev1/Speed"); + var identity = Identity(); + var items = Enumerable.Range(0, 3) + .Select(_ => TrackItem(nm, nodeId, identity)) + .ToList(); + + clock.Reset(); + nm.RevalidateMonitoredItems(); + + clock.Calls.ShouldBe(1, "three items on one node share one (session, node) verdict"); + + // Non-vacuity: the grant really permitted, so the memoized verdict is an allow rather than the + // deny every unresolvable case produces. + items.ShouldAllBe(i => i.Denials == 0); + } + finally { await host.DisposeAsync(); } + } + + /// + /// The falsification control for : + /// three items on three DISTINCT nodes still cost three evaluations, so the dedup is keyed on the + /// node and not collapsing every item in the pass to a single verdict. + /// + [Fact] + public async Task Revalidation_still_evaluates_each_distinct_node() + { + var (host, server) = await BootAsync(); + try + { + var nm = server.NodeManager!; + var clock = new CountingTimeProvider(new DateTime(2026, 8, 15, 12, 0, 0, DateTimeKind.Utc)); + nm.Authorizer = SubscribeAuthorizer(clock); + + var identity = Identity(); + foreach (var tag in new[] { "Speed", "Torque", "Pressure" }) + TrackItem(nm, Variable(nm, $"Plant/Modbus/dev1/{tag}"), identity); + + clock.Reset(); + nm.RevalidateMonitoredItems(); + + clock.Calls.ShouldBe(3); + } + finally { await host.DisposeAsync(); } + } + // ───────────────────────────── helpers ───────────────────────────── + /// A cluster-wide Subscribe grant — what CreateMonitoredItems resolves to. + private static DataPlaneAuthorizer SubscribeAuthorizer(TimeProvider clock) + { + var state = new DataPlaneAuthorityState(); + state.Apply(Cluster, "rev-1", [], [GrantOn(NodePermissions.Subscribe)], Scopes()); + return new DataPlaneAuthorizer( + state, new TriePermissionEvaluator(state.Tries), enabled: true, NullLogger.Instance, clock); + } + + /// Registers a stub data-change item against as if the SDK had created it. + private static StubSampledItem TrackItem(OtOpcUaNodeManager nm, NodeId nodeId, IUserIdentity identity) + { + var handle = (NodeHandle)nm.GetManagerHandle(nodeId); + var item = new StubSampledItem(handle, identity); + nm.TrackLiveMonitoredItem(handle, item); + return item; + } + private static NodeId Variable(OtOpcUaNodeManager nm, string rawPath) { nm.EnsureVariable(rawPath, parentFolderNodeId: null, displayName: rawPath, dataType: "Float", @@ -509,6 +603,154 @@ public sealed class NodeManagerStage6EnforcementTests : IDisposable public void Advance(TimeSpan by) => _now = _now.Add(by); } + /// A fixed clock that counts how many times it was read. + /// + /// The authorizer reads its clock exactly once per Evaluate, so this counts authorization + /// evaluations without needing a seam on the authorizer itself. + /// + private sealed class CountingTimeProvider(DateTime utcNow) : TimeProvider + { + private readonly DateTimeOffset _now = new(utcNow, TimeSpan.Zero); + + public int Calls { get; private set; } + + public override DateTimeOffset GetUtcNow() + { + Calls++; + return _now; + } + + public void Reset() => Calls = 0; + } + + /// + /// The minimum of a sampled data-change monitored item that revalidation touches: the identity that + /// created it, and the denial queue it is told through. + /// + /// + /// Everything else throws rather than returning a plausible default, so a revalidation that started + /// depending on more of the item surfaces here instead of being silently absorbed. + /// + private sealed class StubSampledItem(NodeHandle handle, IUserIdentity identity) + : ISampledDataChangeMonitoredItem + { + /// How many BadUserAccessDenied values revalidation queued on this item. + public int Denials { get; private set; } + + public IUserIdentity EffectiveIdentity => identity; + + public object ManagerHandle => handle; + + public NodeId NodeId => handle.NodeId; + + public void QueueValue(DataValue value, ServiceResult error, bool ignoreFilters) + { + if (error?.StatusCode.Code == StatusCodes.BadUserAccessDenied) Denials++; + } + + public void QueueValue(DataValue value, ServiceResult error) => QueueValue(value, error, false); + + public void Dispose() { } + + // ── everything below is untouched by revalidation ── + public INodeManager NodeManager => throw new NotSupportedException(); + + public ISession Session => throw new NotSupportedException(); + + public uint Id => throw new NotSupportedException(); + + public uint SubscriptionId => throw new NotSupportedException(); + + public bool IsDurable => throw new NotSupportedException(); + + public uint ClientHandle => throw new NotSupportedException(); + + public ISubscription SubscriptionCallback + { + get => throw new NotSupportedException(); + set => throw new NotSupportedException(); + } + + public int MonitoredItemType => throw new NotSupportedException(); + + public bool IsReadyToPublish => throw new NotSupportedException(); + + public bool IsReadyToTrigger + { + get => throw new NotSupportedException(); + set => throw new NotSupportedException(); + } + + public bool IsResendData => throw new NotSupportedException(); + + public MonitoringMode MonitoringMode => throw new NotSupportedException(); + + public double SamplingInterval => throw new NotSupportedException(); + + public DataChangeFilter DataChangeFilter => throw new NotSupportedException(); + + public uint AttributeId => throw new NotSupportedException(); + + public NumericRange IndexRange => throw new NotSupportedException(); + + public QualifiedName DataEncoding => throw new NotSupportedException(); + + public bool AlwaysReportUpdates + { + get => throw new NotSupportedException(); + set => throw new NotSupportedException(); + } + + public DiagnosticsMasks DiagnosticsMasks => throw new NotSupportedException(); + + public uint QueueSize => throw new NotSupportedException(); + + public double MinimumSamplingInterval => throw new NotSupportedException(); + + public int TimeToNextSample => throw new NotSupportedException(); + + public void SetupResendDataTrigger() => throw new NotSupportedException(); + + public ServiceResult GetCreateResult(out MonitoredItemCreateResult result) => + throw new NotSupportedException(); + + public ServiceResult GetModifyResult(out MonitoredItemModifyResult result) => + throw new NotSupportedException(); + + public IStoredMonitoredItem ToStorableMonitoredItem() => throw new NotSupportedException(); + + public MonitoringMode SetMonitoringMode(MonitoringMode monitoringMode) => + throw new NotSupportedException(); + + public bool Publish( + OperationContext context, + Queue notifications, + Queue diagnostics, + uint maxNotificationsPerPublish, + Microsoft.Extensions.Logging.ILogger logger) => throw new NotSupportedException(); + + public void SetSemanticsChanged() => throw new NotSupportedException(); + + public void SetStructureChanged() => throw new NotSupportedException(); + + public bool SamplingIntervalExpired() => throw new NotSupportedException(); + + public ReadValueId GetReadValueId() => throw new NotSupportedException(); + + public ServiceResult ModifyAttributes( + DiagnosticsMasks diagnosticsMasks, + TimestampsToReturn timestampsToReturn, + uint clientHandle, + MonitoringFilter originalFilter, + MonitoringFilter filterToUse, + Opc.Ua.Range range, + double samplingInterval, + uint queueSize, + bool discardOldest) => throw new NotSupportedException(); + + public void SetSamplingInterval(double samplingInterval) => throw new NotSupportedException(); + } + /// Cleans up the PKI root directory. public void Dispose() {