From 4bb9c453a3a540f28255bd7b842aae2d5deb6d6e Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Wed, 11 Nov 2026 10:35:12 -0500 Subject: [PATCH] test(opcua): event-queue overflow with the per-folder/Server-object ACL active (#838 item 2) EventQueueOverflowAclWireTests drives >4096 condition events at a folder notifier item and a Server-object (i=2253) item with Security:Acl enforcement on, a non-publishing client, and a second session reading a node-manager node throughout: - denied (EqA-owned scripted) events interleaved with allowed native ones, under one Lock hold and per-projection: no deadlock, reads stay bounded, deferral peak stays within 2 x (MaxDeferredEventsPerCondition + 1), delivery <= item QueueSize, oldest dropped / newest kept in report order, one EventQueueOverflowEventType published ahead of the survivors (SDK DiscardOldest, Part 4 5.13.1.5), zero denied events delivered; - a grant revoked (state.Apply -> revalidation, re-entering Lock) while a deferred tail waits behind a full folder channel: neither the tail nor later events reach the revoked session's folder or Server-object item; a still-granted session on the same channel receives all of it. Falsified: with enforcement off (and with the revocation removed) all three cases fail. No production defect found. Claude-Session: https://claude.ai/code/session_01GT67AD99zQQqpLCVm3p7xk --- .claude/rules/data-plane-auth.md | 6 +- .../EventQueueOverflowAclWireTests.cs | 349 ++++++++++++++++++ 2 files changed, 354 insertions(+), 1 deletion(-) create mode 100644 tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/EventQueueOverflowAclWireTests.cs diff --git a/.claude/rules/data-plane-auth.md b/.claude/rules/data-plane-auth.md index 3268b7f18..a93cb46bd 100644 --- a/.claude/rules/data-plane-auth.md +++ b/.claude/rules/data-plane-auth.md @@ -219,7 +219,11 @@ Two things to know before relying on any of this: `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`. + 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 + while a deferred tail waits — is `EventQueueOverflowAclWireTests`: the item-queue overflow is the SDK's DiscardOldest + (newest kept, one `EventQueueOverflowEventType` ahead of the survivors); deferred events are decided at DELIVERY, so a + revocation stops the waiting tail; events already in the item queue when a grant is revoked may still publish (authorized when queued). - **ns=0 standard nodes stay ungated by design, even with `Security:Acl:Enabled`** (#617) — `ServerStatus`, `NamespaceArray`, `ServiceLevel` etc. are served by the SDK's core node manager outside our authorizer and clients read them to connect and to judge redundancy health; they carry no plant data. Every raw/UNS node, diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/EventQueueOverflowAclWireTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/EventQueueOverflowAclWireTests.cs new file mode 100644 index 000000000..1ca3b158a --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/EventQueueOverflowAclWireTests.cs @@ -0,0 +1,349 @@ +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; + +namespace ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests; + +/// +/// Gitea #838 item 2 (residual risk of #826): the event-delivery path holds the node manager's Lock while +/// queuing, and the SDK's bounded queues (the 4096-slot per-notifier MonitoredNode2 channel and the per-item +/// event queue) overflow under it. pins the deadlock fix for a folder item +/// with one allowed condition; these tests drive the same overflow with ACL filtering doing real work — a denied +/// condition interleaved with an allowed one, Server-object and folder items, a non-publishing client, a second +/// session reading throughout, and a grant revoked while a deferred backlog waits — and assert: +/// +/// no deadlock, and another session's Read of a node-manager node stays bounded; +/// memory stays within the documented caps (deferral ≤ 1024 + 1 per condition; item queue ≤ QueueSize); +/// the item-queue overflow is the SDK-documented one (Part 4 5.13.1.5, DiscardOldest = true): oldest +/// dropped, newest kept, one EventQueueOverflowEventType event published AHEAD of the survivors; +/// no ACL-denied event reaches an item during the overflow, including deferred events whose grant was +/// revoked while they waited. +/// +/// +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. + private const int PerCondition = 6000; + + /// + /// : the whole flood under ONE hold of Lock (the deadlock shape — the folder + /// channel's consumer needs Lock for its first verdict, so the native tail defers). : + /// each projection takes Lock itself (the driver/engine shape), so a concurrent Read interleaves with it. + /// + [Theory] + [InlineData(true)] + [InlineData(false)] + public async Task An_overflowing_flood_with_denied_events_interleaved_stays_live_bounded_and_never_leaks(bool singleHold) + { + var ct = TestContext.Current.CancellationToken; + var rig = await ServerEventRig.StartAsync("Overflow838Acl", encrypted: false, ct); + var hung = false; + ISession? keyed = null, anonymous = null; + try + { + rig.KeyRoles = []; + rig.MaterialiseNative(); // wired into EqA and EqB + rig.MaterialiseScripted(); // owned by EqA + // The key sees EqB only: the native alarm (through EqB) is allowed, the EqA-owned scripted alarm is DENIED. + // Anonymous sees EqA: both alarms — the positive control that denied events really were being reported. + rig.Enforce( + true, [ServerEventRig.Key()], + KeyEquipmentGrant(EqB), ServerEventRig.EquipmentGrant(EqA, NodePermissions.ReadOnly)); + + keyed = await rig.ConnectAsync(anonymous: false, ct); + anonymous = await rig.ConnectAsync(anonymous: true, ct); + + var keyEvents = new RecordedEvents(); + var keySub = await RecordedEvents.SubscribeAsync(keyed, keyEvents, ct); + var onB = RecordedEvents.AddEventItem(keySub, ServerEventRig.UnsNodeId(keyed, EqB)); + var keyServer = RecordedEvents.AddEventItem(keySub, ObjectIds.Server); + await keySub.ApplyChangesAsync(ct); + onB.Status.Error.ShouldBeNull(); + keyServer.Status.Error.ShouldBeNull(); + // The slow consumer: nothing is published, so the server-side item queues are the only sink. + await keySub.SetPublishingModeAsync(false, ct); + + var anonEvents = new RecordedEvents(); + var anonSub = await RecordedEvents.SubscribeAsync(anonymous, anonEvents, ct); + var anonServer = RecordedEvents.AddEventItem(anonSub, ObjectIds.Server); + await anonSub.ApplyChangesAsync(ct); + anonServer.Status.Error.ShouldBeNull(); + + // Sample the deferral depth throughout (the memory the node manager itself holds for the overflow). + using var sampling = CancellationTokenSource.CreateLinkedTokenSource(ct); + var peakDeferred = 0; + var sampler = Task.Run(async () => + { + while (!sampling.IsCancellationRequested) + { + var now = rig.Nm.DeferredEventReports; + if (now > Volatile.Read(ref peakDeferred)) Volatile.Write(ref peakDeferred, now); + await Task.Delay(1, CancellationToken.None); + } + }, CancellationToken.None); + + // Another session reads a node-manager node (the EqA folder's DisplayName: Read takes Lock) all through the flood. + var eqA = ServerEventRig.UnsNodeId(anonymous, EqA); + var readLatencies = new List(); + var flooding = true; + var reader = Task.Run(async () => + { + while (Volatile.Read(ref flooding)) + { + readLatencies.Add(await TimedReadAsync(anonymous, eqA, ct)); + await Task.Delay(20, ct); + } + }, ct); + + var flood = Task.Factory.StartNew(() => + { + void Burst() + { + for (var i = 0; i < PerCondition; i++) + { + rig.Project(native: true, State($"n{i}")); + rig.Project(native: false, State($"s{i}")); + } + } + + if (singleHold) + { + lock (rig.Nm.Lock) Burst(); + } + else + { + Burst(); + } + }, ct, TaskCreationOptions.LongRunning, TaskScheduler.Default); + + try + { + await flood.WaitAsync(TimeSpan.FromSeconds(60), ct); + } + catch (TimeoutException) + { + hung = true; + throw new ShouldAssertException("the overflowing flood stalled under the node manager's Lock (deadlock)"); + } + + Volatile.Write(ref flooding, false); + await reader.WaitAsync(TimeSpan.FromSeconds(60), ct); + await RecordedEvents.WaitUntilAsync(() => rig.Nm.DeferredEventReports == 0); + await sampling.CancelAsync(); + await sampler; + + // Liveness: every read issued during the flood came back. While one caller holds Lock across the whole flood + // a read waits for that hold (the caller's choice, not a wedge); otherwise each read is short. + readLatencies.ShouldNotBeEmpty(); + if (!singleHold) + 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"); + + // …let the SDK channel finish draining into the (still unpublished) item queues, then publish. + await Task.Delay(TimeSpan.FromSeconds(2), ct); + await keySub.SetPublishingModeAsync(true, ct); + var lastNative = $"n{PerCondition - 1}"; + await RecordedEvents.WaitUntilAsync(() => keyEvents.Has(onB, lastNative) && keyEvents.Has(keyServer, lastNative)); + await Task.Delay(TimeSpan.FromMilliseconds(500), ct); + + 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".) + received.Where(f => RecordedEvents.Message(f)?.StartsWith('s') == true) + .ShouldBeEmpty($"{item.DisplayName}: an ACL-denied event leaked during the overflow"); + + // SDK overflow (DiscardOldest): bounded by the item queue, oldest dropped, newest kept, in report order… + var natives = Indices(received, 'n'); + natives.Count.ShouldBeGreaterThan(0); + natives.Count.ShouldBeLessThanOrEqualTo((int)item.Status.QueueSize, "bounded by the item's queue, not the flood"); + natives.ShouldBe(natives.Order().ToList(), "survivors leave in report order"); + natives[^1].ShouldBe(PerCondition - 1, "the newest event survives"); + natives[0].ShouldBeGreaterThan(0, "the oldest events were the ones dropped"); + + // …and the loss is announced with an EventQueueOverflowEventType event placed ahead of the survivors. + var overflowAt = IndexOfType(received, ObjectTypeIds.EventQueueOverflowEventType); + overflowAt.ShouldBeGreaterThanOrEqualTo(0, $"{item.DisplayName}: no EventQueueOverflowEvent was published"); + overflowAt.ShouldBeLessThan( + received.ToList().FindIndex(f => RecordedEvents.Message(f)?.StartsWith('n') == true), + "the overflow event precedes the surviving events (DiscardOldest)"); + } + + // The server is live and still deciding per event afterwards — and the positive control: the alarm denied to + // the key reaches the session allowed to see it. (Not asserted on the flood itself: the anonymous item's own + // 1000-slot queue may evict any particular flood event before it publishes, by the same DiscardOldest rule.) + rig.Project(native: false, State("after-s")); + rig.Project(native: true, State("after-n")); + await RecordedEvents.WaitUntilAsync(() => + keyEvents.Has(onB, "after-n") && keyEvents.Has(keyServer, "after-n") && anonEvents.Has(anonServer, "after-s")); + await Task.Delay(TimeSpan.FromMilliseconds(300), ct); + keyEvents.Has(keyServer, "after-s").ShouldBeFalse(); + } + finally + { + // A hung rig cannot be disposed (its Lock is held by a blocked thread); leak it rather than hang the run. + if (!hung) + { + keyed?.Dispose(); + anonymous?.Dispose(); + await rig.DisposeAsync(); + } + } + } + + /// + /// A grant revoked while a deferred backlog waits behind a full notifier channel: 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). + /// + [Fact] + public async Task A_grant_revoked_while_overflowed_events_are_deferred_stops_them_all() + { + var ct = TestContext.Current.CancellationToken; + var rig = await ServerEventRig.StartAsync("Overflow838Revoke", encrypted: false, ct); + var hung = false; + ISession? keyed = null, anonymous = null; + try + { + rig.KeyRoles = []; + rig.MaterialiseNative(); + var anonGrant = ServerEventRig.EquipmentGrant(EqA, NodePermissions.ReadOnly); + var state = rig.Enforce(true, [ServerEventRig.Key()], KeyEquipmentGrant(EqA), anonGrant); + + keyed = await rig.ConnectAsync(anonymous: false, ct); + anonymous = await rig.ConnectAsync(anonymous: true, ct); + + var keyEvents = new RecordedEvents(); + var keySub = await RecordedEvents.SubscribeAsync(keyed, keyEvents, ct); + var keyA = RecordedEvents.AddEventItem(keySub, ServerEventRig.UnsNodeId(keyed, EqA)); + var keyServer = RecordedEvents.AddEventItem(keySub, ObjectIds.Server); + await keySub.ApplyChangesAsync(ct); + keyA.Status.Error.ShouldBeNull(); + keyServer.Status.Error.ShouldBeNull(); + await keySub.SetPublishingModeAsync(false, ct); + + var anonEvents = new RecordedEvents(); + var anonSub = await RecordedEvents.SubscribeAsync(anonymous, anonEvents, ct); + var anonA = RecordedEvents.AddEventItem(anonSub, ServerEventRig.UnsNodeId(anonymous, EqA)); + var anonServer = RecordedEvents.AddEventItem(anonSub, ObjectIds.Server); + await anonSub.ApplyChangesAsync(ct); + anonA.Status.Error.ShouldBeNull(); + + var flood = Task.Factory.StartNew(() => + { + lock (rig.Nm.Lock) + { + for (var i = 0; i < PerCondition; i++) rig.Project(native: true, State($"n{i}")); + rig.Nm.DeferredEventReports.ShouldBeGreaterThan(0, "the flood must actually have deferred events"); + + // The deploy that revokes the key's grant lands while the tail waits (its Changed → revalidation runs + // here, re-entering Lock), then one more event is reported. + state.Apply( + ServerEventRig.Cluster, "rev-2", [ServerEventRig.Key()], [anonGrant], ServerEventRig.ScopePaths(), + ServerEventRig.EquipmentOwnerPaths(), ServerEventRig.ScriptedAlarmOwners(), + ServerEventRig.FolderScopes(), ServerEventRig.TagReferences()); + rig.Project(native: true, State("after-revoke")); + } + }, ct, TaskCreationOptions.LongRunning, TaskScheduler.Default); + + try + { + await flood.WaitAsync(TimeSpan.FromSeconds(60), ct); + } + catch (TimeoutException) + { + hung = true; + throw new ShouldAssertException("revoking a grant under Lock while events were deferred stalled (deadlock)"); + } + + 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. + var lastNative = $"n{PerCondition - 1}"; + await RecordedEvents.WaitUntilAsync(() => + anonEvents.Has(anonA, lastNative) && anonEvents.Has(anonA, "after-revoke") + && anonEvents.Has(anonServer, lastNative) && anonEvents.Has(anonServer, "after-revoke")); + await Task.Delay(TimeSpan.FromSeconds(2), ct); + + foreach (var item in new[] { keyA, keyServer }) + { + // Events queued to the item while the grant held may still be published (they were authorized when + // delivered); the deferred tail and everything after the revocation must not be. + keyEvents.Has(item, lastNative).ShouldBeFalse($"{item.DisplayName}: a deferred event leaked past the revocation"); + keyEvents.Has(item, "after-revoke").ShouldBeFalse($"{item.DisplayName}: an event after the revocation leaked"); + var natives = Indices(keyEvents.Of(item), 'n'); + if (natives.Count > 0) natives.Max().ShouldBeLessThan(PerCondition - 1); + } + + (await TimedReadAsync(anonymous, ServerEventRig.UnsNodeId(anonymous, EqA), ct)) + .ShouldBeLessThan(TimeSpan.FromSeconds(2), "the node manager is free afterwards"); + } + finally + { + if (!hung) + { + keyed?.Dispose(); + anonymous?.Dispose(); + await rig.DisposeAsync(); + } + } + } + + private static async Task TimedReadAsync(ISession session, NodeId node, CancellationToken ct) + { + var watch = Stopwatch.StartNew(); + var response = await session.ReadAsync(null, 0, TimestampsToReturn.Neither, new ReadValueId[] + { + new ReadValueId { NodeId = node, AttributeId = Attributes.DisplayName }, + }, ct); + watch.Stop(); + var result = response.Results.ToList().ShouldHaveSingleItem(); + StatusCode.IsGood(result.StatusCode).ShouldBeTrue($"Read of {node} returned {result.StatusCode}"); + return watch.Elapsed; + } + + 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 int IndexOfType(IReadOnlyList> received, NodeId type) => + received.ToList().FindIndex(f => f[RecordedEvents.EventTypeField].GetNodeId() == type); + + private static AlarmConditionSnapshot State(string message) => + new(true, false, false, 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, + }; +}