lmxopcua/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/FolderConditionRefreshWireTests.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;
using MonitoredItem = Opc.Ua.Client.MonitoredItem;

namespace ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests;

/// <summary>
///     Gitea #855: a <c>ConditionRefresh</c> of an event item on an equipment / raw device FOLDER gets the guarantee the
///     Server object got in #854 — no condition event older than what the refresh replayed reaches the item after
///     RefreshEnd, per-condition order is kept, the per-folder ACL (#826) decides every event and every replayed condition,
///     and neither a reporter nor the refresh ever waits on a stalled delivery.
/// </summary>
/// <remarks>
///     <para>
///         Folder items are now served per item by the node manager (no SDK channel), so in production no notifier defers
///         at all; <see cref="Production_overload_never_defers_and_never_overtakes_a_folder_refresh"/> pins that under a
///         flood with concurrent refreshes. Every other test produces the deferral the cut exists for by stalling delivery
///         at a notifier (<see cref="NotifierStall"/>), the way a full SDK channel used to.
///     </para>
///     <para>
///         The rig's native condition is wired into EqA and EqB and hangs under the raw device folder, so its walk is
///         condition → EqA → … → device folder; a stall at EqA suspends it before EqA's items and every later hop.
///     </para>
/// </remarks>
public sealed class FolderConditionRefreshWireTests
{
    private const string EqA = ServerEventRig.EqA;

    // Every event after the stalled one is deferred; < 1024 pending, so nothing coalesces.
    private const int FloodSize = 500;

    /// <param name="enforced">Per-folder ACL on or off (off is the production default).</param>
    /// <param name="finalRetained">Whether the condition is still Retain at refresh time (re-sent) or cleared (omitted).</param>
    /// <param name="refresh2">ConditionRefresh2 on the folder item instead of ConditionRefresh on its subscription.</param>
    /// <param name="modify">
    ///     The folder item is modified (ModifyMonitoredItems re-subscribes it to its notifier) between the refresh and the
    ///     drain: its cut must survive, as the Server object's does across an unsubscribe + resubscribe.
    /// </param>
    [Theory]
    [InlineData(false, true, false, false)]
    [InlineData(true, true, false, false)]
    [InlineData(false, false, false, false)]
    [InlineData(true, false, false, false)]
    [InlineData(false, true, true, false)]
    [InlineData(true, false, true, false)]
    [InlineData(false, true, false, true)]
    [InlineData(true, false, false, true)]
    public async Task Deferred_events_older_than_a_refresh_never_reach_the_refreshed_folder_item_after_RefreshEnd(
        bool enforced, bool finalRetained, bool refresh2, bool modify)
    {
        var ct = TestContext.Current.CancellationToken;
        await using var rig = await ServerEventRig.StartAsync("Refresh855", encrypted: false, ct);
        rig.KeyRoles = [];
        rig.MaterialiseNative();
        rig.Enforce(enforced, [ServerEventRig.Key()], KeyEquipmentGrant(EqA));

        // A second, non-refreshing subscriber on the SAME folder: it must keep the whole deferred tail.
        using var otherSession = await rig.ConnectAsync(anonymous: false, ct);
        var otherEvents = new RecordedEvents();
        var otherSubscription = await RecordedEvents.SubscribeAsync(otherSession, otherEvents, ct);
        var otherOnA = RecordedEvents.AddEventItem(otherSubscription, ServerEventRig.UnsNodeId(otherSession, EqA));
        await otherSubscription.ApplyChangesAsync(ct);
        otherOnA.Status.Error.ShouldBeNull();

        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));
        await subscription.ApplyChangesAsync(ct);
        onA.Status.Error.ShouldBeNull();

        rig.Project(native: true, Active("seed"));
        await RecordedEvents.WaitUntilAsync(() => events.Has(onA, "seed") && otherEvents.Has(otherOnA, "seed"));

        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, "the stall must actually have deferred events");
            rig.Project(native: true, finalRetained ? Active("final") : Cleared("final"));

            var refresh = Stopwatch.StartNew();
            await RefreshAsync(session, subscription, onA, refresh2, ct);
            refresh.Elapsed.ShouldBeLessThan(TimeSpan.FromSeconds(10), "ConditionRefresh must not wait for the stalled delivery");
            await RecordedEvents.WaitUntilAsync(() => events.HasType(onA, ObjectTypeIds.RefreshEndEventType));
            rig.Nm.DeferredEventReports.ShouldBeGreaterThan(0, "the deferred events were still waiting at RefreshEnd");
            rig.Nm.ActiveRefreshCuts.ShouldBe(1, "only the refreshed folder item holds a cut");

            if (modify)
            {
                onA.QueueSize = 1001;
                await subscription.ApplyChangesAsync(ct);
                onA.Status.Error.ShouldBeNull();
            }

            stall.Release();
        }

        await RecordedEvents.WaitUntilAsync(() => rig.Nm.DeferredEventReports == 0);
        await RecordedEvents.WaitUntilAsync(() => rig.Nm.ActiveRefreshCuts == 0);
        rig.Project(native: true, Active("after"));
        await RecordedEvents.WaitUntilAsync(() => events.Has(onA, "after") && otherEvents.Has(otherOnA, "after"));
        await Task.Delay(TimeSpan.FromMilliseconds(300), ct);

        var (replayed, afterEnd) = Window(events.Of(onA));
        if (finalRetained)
        {
            replayed.ShouldBe(["final"], "the refresh re-sends the condition's current state");
            afterEnd.ShouldBe(["after"], "nothing older than the refreshed state may follow RefreshEnd");
        }
        else
        {
            replayed.ShouldBeEmpty("a cleared (Retain=false) condition is not part of a refresh");
            afterEnd.ShouldBe(["final", "after"], "stale Retain=true events would re-add the alarm; the Retain=false one agrees");
            events.Of(onA).Last(f => RecordedEvents.Message(f) == "final")[RecordedEvents.RetainField].GetBoolean().ShouldBeFalse();
        }

        rig.Nm.RefreshSupersededDeliveries.ShouldBeGreaterThan(0);

        // The other subscriber on the same folder did not refresh: it keeps every deferred event, in order.
        var flood = Indices(otherEvents.Of(otherOnA), 'f');
        flood.ShouldBe(Enumerable.Range(0, FloodSize).ToList(), "the non-refreshing folder subscriber keeps the whole tail, in order");
        otherEvents.Has(otherOnA, "final").ShouldBeTrue();
    }

    /// <summary>
    ///     The race the item gate closes, for a folder item: a deferred event is between its supersession check and its
    ///     queueing to the folder item when the refresh arrives. The refresh must wait for that one delivery, so it lands
    ///     before the replay rather than after RefreshEnd.
    /// </summary>
    [Fact]
    public async Task A_folder_refresh_waits_for_a_deferred_delivery_already_past_its_check()
    {
        var ct = TestContext.Current.CancellationToken;
        await using var rig = await ServerEventRig.StartAsync("Refresh855Gate", 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));
        await subscription.ApplyChangesAsync(ct);

        rig.Project(native: true, Active("seed"));
        await RecordedEvents.WaitUntilAsync(() => events.Has(onA, "seed"));

        using var entered = new ManualResetEventSlim();
        using var proceed = new ManualResetEventSlim();
        var probed = 0;
        rig.Nm.DeferredFolderDeliveryProbe = () =>
        {
            if (Interlocked.Increment(ref probed) != 1) return;
            entered.Set();
            proceed.Wait(TimeSpan.FromSeconds(30));
        };

        using (NotifierStall.Hold(rig.Nm, EqA))
        {
            for (var i = 0; i < FloodSize; i++) rig.Project(native: true, Active($"f{i}"));
            rig.Project(native: true, Active("final"));
        }

        // The released walk parks on its first deferred delivery to the folder item, past the check, inside the gate.
        entered.Wait(TimeSpan.FromSeconds(10), ct).ShouldBeTrue("the drain must reach the folder item");
        var refresh = session.CallAsync(
            ObjectTypeIds.ConditionType, MethodIds.ConditionType_ConditionRefresh, ct, subscription.Id);
        await Task.Delay(TimeSpan.FromSeconds(1), ct);
        events.HasType(onA, ObjectTypeIds.RefreshEndEventType)
            .ShouldBeFalse("the refresh must wait for the delivery already inside the item's gate");
        proceed.Set();
        await refresh.WaitAsync(TimeSpan.FromSeconds(10), ct);
        await RecordedEvents.WaitUntilAsync(() => events.HasType(onA, ObjectTypeIds.RefreshEndEventType));
        rig.Nm.DeferredFolderDeliveryProbe = null;

        await RecordedEvents.WaitUntilAsync(() => rig.Nm.DeferredEventReports == 0);
        rig.Project(native: true, Active("after"));
        await RecordedEvents.WaitUntilAsync(() => events.Has(onA, "after"));
        await Task.Delay(TimeSpan.FromMilliseconds(300), ct);

        var (window, afterEnd) = Window(events.Of(onA));
        window.Count.ShouldBe(2, "the parked delivery, then the replay");
        window[0]!.ShouldStartWith("f");
        window[1].ShouldBe("final");
        afterEnd.ShouldBe(["after"]);
    }

    /// <summary>
    ///     The cut path queues its frozen replay OFF <c>Lock</c>. An event newer than the cut that reaches the folder item
    ///     in that window (a condition delivered synchronously, reported from the probe between the cut and the queueing)
    ///     must not be followed by its condition's older frozen replay entry.
    /// </summary>
    [Fact]
    public async Task An_event_newer_than_the_folder_cut_is_never_followed_by_its_older_frozen_replay()
    {
        var ct = TestContext.Current.CancellationToken;
        await using var rig = await ServerEventRig.StartAsync("Refresh855Pending", encrypted: false, ct);
        rig.KeyRoles = [];
        rig.MaterialiseNative();
        rig.Enforce(false, [ServerEventRig.Key()], KeyEquipmentGrant(EqA));
        // A second condition under the device folder only: its walk never reaches the stalled EqA, so it stays synchronous.
        const string Free = ServerEventRig.RawDeviceFolder + "/free";
        rig.Sink.MaterialiseAlarmCondition(Free, ServerEventRig.RawDeviceFolder, "free", "OffNormalAlarm", 700, AddressSpaceRealm.Raw, isNative: true);
        rig.Sink.WriteAlarmCondition(Free, Active("x1"), DateTime.UtcNow, AddressSpaceRealm.Raw);

        // EqA needs an event item of its own to be a notifier fan-out the stall can apply to.
        using var eqASession = await rig.ConnectAsync(anonymous: false, ct);
        var eqAEvents = new RecordedEvents();
        var eqASubscription = await RecordedEvents.SubscribeAsync(eqASession, eqAEvents, ct);
        RecordedEvents.AddEventItem(eqASubscription, ServerEventRig.UnsNodeId(eqASession, EqA));
        await eqASubscription.ApplyChangesAsync(ct);

        // The refreshed item is on the raw device folder, which both conditions report through.
        using var session = await rig.ConnectAsync(anonymous: false, ct);
        var events = new RecordedEvents();
        var subscription = await RecordedEvents.SubscribeAsync(session, events, ct);
        var onDevice = RecordedEvents.AddEventItem(subscription, RawNodeId(session, ServerEventRig.RawDeviceFolder));
        await subscription.ApplyChangesAsync(ct);
        onDevice.Status.Error.ShouldBeNull();

        rig.Project(native: true, Active("seed"));
        await RecordedEvents.WaitUntilAsync(() => events.Has(onDevice, "seed"));

        var injected = 0;
        rig.Nm.DirectRefreshProbe = (cut, _) =>
        {
            if (cut && Interlocked.Exchange(ref injected, 1) == 0)
                rig.Sink.WriteAlarmCondition(Free, Active("x2"), DateTime.UtcNow, AddressSpaceRealm.Raw);
        };

        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, "the cut path runs only while something is deferred");
            await session.CallAsync(ObjectTypeIds.ConditionType, MethodIds.ConditionType_ConditionRefresh, ct, subscription.Id);
            await RecordedEvents.WaitUntilAsync(() => events.HasType(onDevice, ObjectTypeIds.RefreshEndEventType));
            stall.Release();
        }

        rig.Nm.DirectRefreshProbe = null;
        Volatile.Read(ref injected).ShouldBe(1, "the folder refresh took the cut path");
        var stream = events.Of(onDevice).ToList();
        var end = stream.FindIndex(f => f[RecordedEvents.EventTypeField].GetNodeId() == ObjectTypeIds.RefreshEndEventType);
        var free = stream.Take(end).Select(RecordedEvents.Message).Where(m => m is "x1" or "x2").ToList();
        free.ShouldContain("x2");
        free[^1].ShouldBe("x2", "the older frozen replay of the condition must not follow its newer event");
    }

    /// <summary>
    ///     ACL on: a condition the folder refresh omitted because the item may not see it there (its reference into EqA was
    ///     dropped by the generation) is not withheld afterwards — the reference returns before the drain, and the item must
    ///     end on the condition's newest state. And while omitted, the replay itself carried nothing of it.
    /// </summary>
    [Fact]
    public async Task Deferred_events_of_a_condition_the_folder_refresh_omitted_by_the_acl_are_not_withheld()
    {
        var ct = TestContext.Current.CancellationToken;
        await using var rig = await ServerEventRig.StartAsync("Refresh855Acl", encrypted: false, ct);
        rig.KeyRoles = [];
        rig.MaterialiseNative();
        NodeAcl[] grants = [KeyEquipmentGrant(EqA)];
        var state = rig.Enforce(true, [ServerEventRig.Key()], grants);
        Deploy(state, "rev-2", grants, ServerEventRig.EqB); // the native alarm is no longer projected into 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));
        await subscription.ApplyChangesAsync(ct);
        onA.Status.Error.ShouldBeNull();

        rig.Project(native: true, Active("seed"));
        await Task.Delay(TimeSpan.FromMilliseconds(500), ct);
        events.Has(onA, "seed").ShouldBeFalse("the generation does not project the alarm into EqA");

        using (var stall = NotifierStall.Hold(rig.Nm, EqA))
        {
            for (var i = 0; i < FloodSize; i++) rig.Project(native: true, Active($"f{i}"));
            rig.Project(native: true, Active("final"));
            rig.Nm.DeferredEventReports.ShouldBeGreaterThan(0);

            await session.CallAsync(ObjectTypeIds.ConditionType, MethodIds.ConditionType_ConditionRefresh, ct, subscription.Id);
            await RecordedEvents.WaitUntilAsync(() => events.HasType(onA, ObjectTypeIds.RefreshEndEventType));

            // The reference returns while the condition's events still wait.
            Deploy(state, "rev-3", grants, EqA, ServerEventRig.EqB);
            stall.Release();
        }

        await RecordedEvents.WaitUntilAsync(() => rig.Nm.DeferredEventReports == 0);
        await RecordedEvents.WaitUntilAsync(() => events.Has(onA, "final"));
        var (replayed, after) = Window(events.Of(onA));
        replayed.ShouldBeEmpty("the refresh omitted the condition: not visible through EqA at the time");
        after.ShouldNotBeEmpty();
        after[^1].ShouldBe("final", "the item ends on the newest state of a condition it was never refreshed");
        after.Count(m => m!.StartsWith('f')).ShouldBeGreaterThan(0, "its deferred events were delivered, not withheld");
    }

    /// <summary>
    ///     ACL on: a grant revoked while a refreshed folder item's deferred events wait. Neither the replay (taken while
    ///     granted) nor anything after it leaks once revoked; a still-granted subscriber on the same folder gets the tail.
    /// </summary>
    [Fact]
    public async Task A_grant_revoked_after_a_folder_refresh_withholds_every_deferred_event()
    {
        var ct = TestContext.Current.CancellationToken;
        await using var rig = await ServerEventRig.StartAsync("Refresh855Revoke", encrypted: false, ct);
        rig.KeyRoles = [];
        rig.MaterialiseNative();
        var anonGrant = ServerEventRig.EquipmentGrant(EqA, NodePermissions.ReadOnly);
        var state = rig.Enforce(true, [ServerEventRig.Key()], KeyEquipmentGrant(EqA), anonGrant);

        using var keyed = await rig.ConnectAsync(anonymous: false, ct);
        var keyEvents = new RecordedEvents();
        var keySub = await RecordedEvents.SubscribeAsync(keyed, keyEvents, ct);
        var keyA = RecordedEvents.AddEventItem(keySub, ServerEventRig.UnsNodeId(keyed, EqA));
        await keySub.ApplyChangesAsync(ct);

        using var anonymous = await rig.ConnectAsync(anonymous: true, ct);
        var anonEvents = new RecordedEvents();
        var anonSub = await RecordedEvents.SubscribeAsync(anonymous, anonEvents, ct);
        var anonA = RecordedEvents.AddEventItem(anonSub, ServerEventRig.UnsNodeId(anonymous, EqA));
        await anonSub.ApplyChangesAsync(ct);

        rig.Project(native: true, Active("seed"));
        await RecordedEvents.WaitUntilAsync(() => keyEvents.Has(keyA, "seed") && anonEvents.Has(anonA, "seed"));

        using (var stall = NotifierStall.Hold(rig.Nm, EqA))
        {
            for (var i = 0; i < FloodSize; i++) rig.Project(native: true, Active($"f{i}"));
            await keyed.CallAsync(ObjectTypeIds.ConditionType, MethodIds.ConditionType_ConditionRefresh, ct, keySub.Id);
            await RecordedEvents.WaitUntilAsync(() => keyEvents.HasType(keyA, ObjectTypeIds.RefreshEndEventType));

            state.Apply(
                ServerEventRig.Cluster, "rev-2", [ServerEventRig.Key()], [anonGrant], ServerEventRig.ScopePaths(),
                ServerEventRig.EquipmentOwnerPaths(), ServerEventRig.ScriptedAlarmOwners(), ServerEventRig.FolderScopes(),
                ServerEventRig.TagReferences());
            rig.Project(native: true, Active("after-revoke"));
            stall.Release();
        }

        await RecordedEvents.WaitUntilAsync(() => rig.Nm.DeferredEventReports == 0);
        await RecordedEvents.WaitUntilAsync(() =>
            anonEvents.Has(anonA, $"f{FloodSize - 1}") && anonEvents.Has(anonA, "after-revoke"));
        await Task.Delay(TimeSpan.FromSeconds(1), ct);

        var (_, after) = Window(keyEvents.Of(keyA));
        after.ShouldBeEmpty("after RefreshEnd the revoked session gets nothing: older events are superseded, later ones denied");
        keyEvents.Has(keyA, "after-revoke").ShouldBeFalse();
    }

    /// <summary>
    ///     #855 review: the cut path decides its replay before taking <c>Lock</c>; 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.
    /// </summary>
    /// <param name="serverObject">Refresh a Server-object item instead of a folder item.</param>
    [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");
    }

    /// <summary>
    ///     #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 <c>Lock</c> was held). The Server-object item, served before the failing hop, still gets
    ///     the event, and the next event is delivered normally.
    /// </summary>
    [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");
    }

    /// <summary>
    ///     Production shape, no stall: bursts reported under one <c>Lock</c> 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
    ///     nothing is ever deferred; each refresh's replay is never older than an event the item already received, nothing
    ///     after RefreshEnd goes backwards, the stream converges on the newest state, and no replay is queued under <c>Lock</c>.
    /// </summary>
    /// <param name="enforced">Per-folder ACL on or off.</param>
    [Theory]
    [InlineData(false)]
    [InlineData(true)]
    public async Task Production_overload_never_defers_and_never_overtakes_a_folder_refresh(bool enforced)
    {
        var ct = TestContext.Current.CancellationToken;
        await using var rig = await ServerEventRig.StartAsync("Refresh855Load", encrypted: false, ct);
        rig.KeyRoles = [];
        rig.MaterialiseNative();
        rig.Enforce(enforced, [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));
        onA.QueueSize = 100_000; // keep every event so the ordering check sees the whole stream
        await subscription.ApplyChangesAsync(ct);

        rig.Project(native: true, Active("s0"));
        await RecordedEvents.WaitUntilAsync(() => events.Has(onA, "s0"));

        var probes = new System.Collections.Concurrent.ConcurrentQueue<(bool Cut, bool LockHeld)>();
        rig.Nm.DirectRefreshProbe = (cut, held) => probes.Enqueue((cut, held));

        using var stop = CancellationTokenSource.CreateLinkedTokenSource(ct);
        var counter = 0;
        var peakDeferred = 0;
        var reporter = Task.Factory.StartNew(() =>
        {
            while (!stop.IsCancellationRequested)
            {
                lock (rig.Nm.Lock)
                {
                    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);
                    }
                }

                rig.Project(native: true, Active($"s{Interlocked.Increment(ref counter)}"));
                Thread.Sleep(2);
            }
        }, stop.Token, TaskCreationOptions.LongRunning, TaskScheduler.Default);

        var refusedAsInProgress = 0;
        for (var r = 0; r < 30; r++)
        {
            try
            {
                await RefreshAsync(session, subscription, onA, refresh2: r % 3 == 2, ct).WaitAsync(TimeSpan.FromSeconds(20), ct);
            }
            catch (ServiceResultException ex) when (ex.StatusCode == StatusCodes.BadRefreshInProgress)
            {
                refusedAsInProgress++;
            }

            await Task.Delay(TimeSpan.FromMilliseconds(80), ct);
        }

        await stop.CancelAsync();
        await reporter.WaitAsync(TimeSpan.FromSeconds(30), CancellationToken.None);
        rig.Nm.DirectRefreshProbe = null;

        peakDeferred.ShouldBe(0, "no notifier of ours defers: folder items are served per item by the reporter");
        rig.Nm.DiscardedEventReports.ShouldBe(0);
        var newest = $"s{Volatile.Read(ref counter)}";
        await RecordedEvents.WaitUntilAsync(() => events.Has(onA, newest));
        await Task.Delay(TimeSpan.FromMilliseconds(500), ct);

        CheckStream(events.Of(onA)).ShouldBeEmpty();
        refusedAsInProgress.ShouldBeLessThan(30, "most refreshes must actually have run");
        probes.ShouldNotBeEmpty();
        probes.ShouldAllBe(p => !p.Cut && !p.LockHeld);
        RecordedEvents.Message(events.Of(onA).Last(f => f[RecordedEvents.AckedField].TryGetValue(out bool _)))
            .ShouldBe(newest, "the stream converges on the newest state");
    }

    /// <summary>
    ///     Race / stress with deferral: reports (bursts under <c>Lock</c> and single), repeated refreshes of the folder item
    ///     and an intermittent stall of its folder's delivery all run at once. Outside a refresh the folder item's stream
    ///     never goes backwards, each refresh ends on the newest state the item has seen, and it converges. Nothing
    ///     deadlocks, and neither refresh path queues its replay under <c>Lock</c>.
    /// </summary>
    /// <param name="enforced">Per-folder ACL on or off.</param>
    [Theory]
    [InlineData(false)]
    [InlineData(true)]
    public async Task Concurrent_reports_folder_refreshes_and_stalls_never_deliver_a_pre_refresh_state_after_RefreshEnd(bool enforced)
    {
        var ct = TestContext.Current.CancellationToken;
        await using var rig = await ServerEventRig.StartAsync("Refresh855Stress", encrypted: false, ct);
        rig.KeyRoles = [];
        rig.MaterialiseNative();
        rig.Enforce(enforced, [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));
        onA.QueueSize = 100_000;
        await subscription.ApplyChangesAsync(ct);

        rig.Project(native: true, Active("s0"));
        await RecordedEvents.WaitUntilAsync(() => events.Has(onA, "s0"));

        var probes = new System.Collections.Concurrent.ConcurrentQueue<(bool Cut, bool LockHeld)>();
        rig.Nm.DirectRefreshProbe = (cut, held) => probes.Enqueue((cut, held));

        using var stop = CancellationTokenSource.CreateLinkedTokenSource(ct);
        var counter = 0;
        var reporter = Task.Factory.StartNew(() =>
        {
            while (!stop.IsCancellationRequested)
            {
                lock (rig.Nm.Lock)
                {
                    for (var i = 0; i < 300; i++) rig.Project(native: true, Active($"s{Interlocked.Increment(ref counter)}"));
                }

                rig.Project(native: true, Active($"s{Interlocked.Increment(ref counter)}"));
                Thread.Sleep(2);
            }
        }, stop.Token, TaskCreationOptions.LongRunning, TaskScheduler.Default);

        var deferredSeen = 0;
        var staller = Task.Factory.StartNew(() =>
        {
            while (!stop.IsCancellationRequested)
            {
                using (NotifierStall.Hold(rig.Nm, EqA))
                {
                    Thread.Sleep(300);
                    if (rig.Nm.DeferredEventReports > 0) Interlocked.Increment(ref deferredSeen);
                }

                Thread.Sleep(100);
            }
        }, stop.Token, TaskCreationOptions.LongRunning, TaskScheduler.Default);

        var refusedAsInProgress = 0;
        var stressDeadline = DateTime.UtcNow + TimeSpan.FromSeconds(90);
        for (var r = 0; r < 40 || (Volatile.Read(ref deferredSeen) <= 1 && DateTime.UtcNow < stressDeadline); r++)
        {
            try
            {
                await RefreshAsync(session, subscription, onA, refresh2: r % 3 == 2, ct).WaitAsync(TimeSpan.FromSeconds(20), ct);
            }
            catch (ServiceResultException ex) when (ex.StatusCode == StatusCodes.BadRefreshInProgress)
            {
                refusedAsInProgress++;
            }

            await Task.Delay(TimeSpan.FromMilliseconds(80), ct);
        }

        await stop.CancelAsync();
        await reporter.WaitAsync(TimeSpan.FromSeconds(30), CancellationToken.None);
        await staller.WaitAsync(TimeSpan.FromSeconds(30), CancellationToken.None);
        rig.Nm.DirectRefreshProbe = null;
        Volatile.Read(ref deferredSeen).ShouldBeGreaterThan(1, "the stress must actually have run with deferred events");

        await RecordedEvents.WaitUntilAsync(() => rig.Nm.DeferredEventReports == 0);
        var newest = $"s{Volatile.Read(ref counter)}";
        await RecordedEvents.WaitUntilAsync(() => events.Has(onA, newest));
        await Task.Delay(TimeSpan.FromMilliseconds(500), ct);

        CheckStream(events.Of(onA)).ShouldBeEmpty();
        refusedAsInProgress.ShouldBeLessThan(40, "most refreshes must actually have run");
        rig.Nm.RefreshSupersededDeliveries.ShouldBeGreaterThan(0, "some folder refresh must actually have overtaken deferred events");
        probes.ShouldContain(p => p.Cut, "some folder refresh must have taken the cut path");
        probes.ShouldAllBe(p => !p.LockHeld);
        RecordedEvents.Message(events.Of(onA).Last(f => f[RecordedEvents.AckedField].TryGetValue(out bool _)))
            .ShouldBe(newest, "the stream converges on the newest state");
    }

    /// <summary>
    ///     Walks one condition's stream: outside a refresh its states never go backwards, and each refresh ends on the newest
    ///     state the item has seen (its replay is never older than an event already delivered, and nothing older follows).
    /// </summary>
    private static List<string> CheckStream(IReadOnlyList<ArrayOf<Variant>> stream)
    {
        var max = -1;
        var last = -1;
        var replaying = false;
        var violations = new List<string>();
        foreach (var f in stream)
        {
            var type = f[RecordedEvents.EventTypeField].GetNodeId();
            if (type == ObjectTypeIds.RefreshStartEventType)
            {
                replaying = true;
                continue;
            }

            if (type == ObjectTypeIds.RefreshEndEventType)
            {
                replaying = false;
                if (last < max) violations.Add($"a refresh ended on s{last} after s{max} had been delivered");
                continue;
            }

            if (!f[RecordedEvents.AckedField].TryGetValue(out bool _)) continue;
            var n = int.Parse(RecordedEvents.Message(f)![1..], CultureInfo.InvariantCulture);
            if (!replaying && n < max) violations.Add($"s{n} delivered after s{max}");
            max = Math.Max(max, n);
            last = n;
        }

        return violations;
    }

    private static Task RefreshAsync(ISession session, Subscription subscription, MonitoredItem item, bool refresh2, CancellationToken ct) =>
        refresh2
            ? session.CallAsync(ObjectTypeIds.ConditionType, MethodIds.ConditionType_ConditionRefresh2, ct, subscription.Id, item.Status.Id)
            : session.CallAsync(ObjectTypeIds.ConditionType, MethodIds.ConditionType_ConditionRefresh, ct, subscription.Id);

    /// <summary>The condition-event messages inside the item's LAST RefreshStart..RefreshEnd, and those after it.</summary>
    private static (List<string?> Replayed, List<string?> AfterEnd) Window(IReadOnlyList<ArrayOf<Variant>> stream)
    {
        var list = stream.ToList();
        var start = list.FindLastIndex(f => f[RecordedEvents.EventTypeField].GetNodeId() == ObjectTypeIds.RefreshStartEventType);
        var end = list.FindLastIndex(f => f[RecordedEvents.EventTypeField].GetNodeId() == ObjectTypeIds.RefreshEndEventType);
        start.ShouldBeGreaterThanOrEqualTo(0);
        end.ShouldBeGreaterThan(start);
        static bool IsCondition(ArrayOf<Variant> f) => f[RecordedEvents.AckedField].TryGetValue(out bool _);
        return (
            list.Skip(start + 1).Take(end - start - 1).Where(IsCondition).Select(RecordedEvents.Message).ToList(),
            list.Skip(end + 1).Where(IsCondition).Select(RecordedEvents.Message).ToList());
    }

    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 NodeId RawNodeId(ISession session, string id) =>
        new(id, (ushort)session.NamespaceUris.GetIndex(V3NodeIds.RawNamespaceUri));

    private static void Deploy(DataPlaneAuthorityState state, string revision, NodeAcl[] grants, params string[] referencing) =>
        state.Apply(
            ServerEventRig.Cluster, revision, [ServerEventRig.Key()], grants, ServerEventRig.ScopePaths(),
            ServerEventRig.EquipmentOwnerPaths(), ServerEventRig.ScriptedAlarmOwners(), ServerEventRig.FolderScopes(),
            new Dictionary<string, IReadOnlySet<string>> { [ServerEventRig.RawAlarmPath] = referencing.ToHashSet() });

    private static AlarmConditionSnapshot Active(string message) =>
        new(true, false, false, true, AlarmShelvingKind.Unshelved, 700, message);

    private static AlarmConditionSnapshot Cleared(string message) =>
        new(false, true, true, 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,
    };
}