lmxopcua/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/EventQueueOverflowAclWireTests.cs

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,
    };
}