dohertj2/lmxopcua
1
0
Fork 0
Code Issues 5 Pull Requests Actions Packages Projects Releases Wiki Activity

test(opcua): event-queue overflow with the per-folder/Server-object ACL active (#838 item 2)

Browse Source 
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
This commit is contained in:
Joseph Doherty
2026-11-11 10:35:12 -05:00
parent 03e5145796
commit 4bb9c453a3
2 changed files with 354 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.claude/rules/data-plane-auth.md
+5 -1
View File
@@ -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,
tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/EventQueueOverflowAclWireTests.cs
+349
View File
@@ -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;
/// <summary>
/// Gitea #838 item 2 (residual risk of #826): the event-delivery path holds the node manager's <c>Lock</c> while
/// queuing, and the SDK's bounded queues (the 4096-slot per-notifier <c>MonitoredNode2</c> channel and the per-item
/// event queue) overflow under it. <see cref="FolderEventFloodWireTests"/> 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:
/// <list type="bullet">
/// <item>no deadlock, and another session's Read of a node-manager node stays bounded;</item>
/// <item>memory stays within the documented caps (deferral ≤ 1024 + 1 per condition; item queue ≤ QueueSize);</item>
/// <item>the item-queue overflow is the SDK-documented one (Part 4 5.13.1.5, DiscardOldest = true): oldest
/// dropped, newest kept, one <c>EventQueueOverflowEventType</c> event published AHEAD of the survivors;</item>
/// <item>no ACL-denied event reaches an item during the overflow, including deferred events whose grant was
/// revoked while they waited.</item>
/// </list>
/// </summary>
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;
/// <param name="singleHold">
/// <see langword="true"/>: the whole flood under ONE hold of <c>Lock</c> (the deadlock shape — the folder
/// channel's consumer needs <c>Lock</c> for its first verdict, so the native tail defers). <see langword="false"/>:
/// each projection takes <c>Lock</c> itself (the driver/engine shape), so a concurrent Read interleaves with it.
/// </param>
[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<TimeSpan>();
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();
}
}
}
/// <summary>
/// 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).
/// </summary>
[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<TimeSpan> 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<int> Indices(IReadOnlyList<ArrayOf<Variant>> 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<ArrayOf<Variant>> 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,
};
}