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

using System.Diagnostics;
using System.Globalization;
using System.Text;
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 cost measurement: since #855 a condition event is queued to every event item on an equipment folder by
///     the reporter itself, under the node manager's <c>Lock</c> (per-item verdict, <c>QueueEvent</c> with the SDK's
///     filter evaluation, duplicate scan over the last 1000 queued events and a <c>List.RemoveAt(0)</c> at a full queue).
///     Before #855 that work ran on the SDK's per-node channel consumer, off <c>Lock</c>. This harness reports, per alarm
///     report, the <c>Lock</c> hold time and the end-to-end delivery latency, for one equipment folder carrying N event
///     items, ACL on / off, with the folder's subscribers publishing or stalled (publishing disabled, item queues full and
///     discarding oldest) — and the latency of a Server-object (<c>i=2253</c>) subscriber while the folder's subscribers
///     are stalled (the ScadaBridge shape).
/// </summary>
/// <remarks>
///     <para>Measurement only: nothing is asserted beyond the harness's own sanity checks. Opt-in and never part of a
///     default run: every case skips unless <c>OTOPCUA_EVENT_PERF=1</c>.</para>
///     <para>Run: <c>OTOPCUA_EVENT_PERF=1 dotnet test tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests -c Release
///     --filter "FullyQualifiedName~FolderEventDeliveryPerfTests" --logger "console;verbosity=detailed"</c>.
///     <c>OTOPCUA_EVENT_PERF_REPORTS</c> sets the measured reports per case (default 1000) and
///     <c>OTOPCUA_EVENT_PERF_OUT</c> appends each result row to a file.</para>
///     <para>The rig is the same one <see cref="FolderEventFloodWireTests"/> uses and existed before #855, so this file
///     compiles against the pre-#855 tree for a before/after comparison.</para>
/// </remarks>
[Trait("Category", "Perf")]
public sealed class FolderEventDeliveryPerfTests
{
    private const string OptInVariable = "OTOPCUA_EVENT_PERF";
    private const string EqA = ServerEventRig.EqA;
    private const uint DefaultQueueSize = 1000;
    private const double PublishingIntervalMs = 50;

    private static int Reports =>
        int.TryParse(Environment.GetEnvironmentVariable("OTOPCUA_EVENT_PERF_REPORTS"), out var r) && r > 0 ? r : 1000;

    private static void RequireOptIn()
    {
        if (Environment.GetEnvironmentVariable(OptInVariable) != "1")
            Assert.Skip($"perf harness: set {OptInVariable}=1 to run (not part of a default run)");
    }

    /// <summary>
    ///     Lock hold per report and (publishing only) the time until the LAST of the N folder items received the event.
    /// </summary>
    [Theory]
    [InlineData(1, false, false)]
    [InlineData(10, false, false)]
    [InlineData(50, false, false)]
    [InlineData(200, false, false)]
    [InlineData(1, true, false)]
    [InlineData(10, true, false)]
    [InlineData(50, true, false)]
    [InlineData(200, true, false)]
    [InlineData(1, false, true)]
    [InlineData(10, false, true)]
    [InlineData(50, false, true)]
    [InlineData(200, false, true)]
    [InlineData(1, true, true)]
    [InlineData(10, true, true)]
    [InlineData(50, true, true)]
    [InlineData(200, true, true)]
    public Task Folder_items(int n, bool acl, bool stalled) => RunAsync(n, acl, stalled, serverObjectProbe: false);

    /// <summary>
    ///     A publishing Server-object (<c>i=2253</c>) subscriber on its own session, while N folder items on another
    ///     session are stalled with full queues. N = 0 is the baseline.
    /// </summary>
    [Theory]
    [InlineData(0, false)]
    [InlineData(50, false)]
    [InlineData(200, false)]
    [InlineData(0, true)]
    [InlineData(50, true)]
    [InlineData(200, true)]
    public Task Server_object_subscriber_while_folder_items_stall(int n, bool acl) =>
        RunAsync(n, acl, stalled: true, serverObjectProbe: true);

    /// <summary>
    ///     Full-queue sensitivity: N = 50 stalled items at different <c>QueueSize</c>s. At a full queue every
    ///     <c>QueueEvent</c> scans the last min(QueueSize, 1000) queued events for a duplicate and shifts the whole queue
    ///     (<c>List.RemoveAt(0)</c>), so the per-item cost grows with the queue, not with N.
    /// </summary>
    [Theory]
    [InlineData(10u)]
    [InlineData(100u)]
    [InlineData(1000u)]
    [InlineData(10000u)]
    public Task Stalled_items_queue_size_sensitivity(uint queueSize) =>
        RunAsync(50, acl: false, stalled: true, serverObjectProbe: false, queueSize);

    private static async Task RunAsync(
        int n, bool acl, bool stalled, bool serverObjectProbe, uint queueSize = DefaultQueueSize)
    {
        RequireOptIn();
        var ct = TestContext.Current.CancellationToken;
        await using var rig = await ServerEventRig.StartAsync($"Perf855-{n}", encrypted: false, ct);
        rig.KeyRoles = [];
        rig.MaterialiseNative();
        rig.Enforce(acl, [ServerEventRig.Key()], KeyEquipmentGrant(EqA));

        var reports = Reports;
        var reportedAt = new long[reports];
        var receipts = new Receipts(reports);

        ISession? folderSession = null;
        ISession? serverSession = null;
        try
        {
            Subscription? folderSubscription = null;
            var folderHandles = new HashSet<uint>();
            if (n > 0)
            {
                folderSession = await rig.ConnectAsync(anonymous: false, ct);
                folderSubscription = await SubscribeAsync(folderSession, receipts, isServer: false, ct);
                for (var i = 0; i < n; i++)
                {
                    var item = RecordedEvents.AddEventItem(folderSubscription, ServerEventRig.UnsNodeId(folderSession, EqA));
                    item.QueueSize = queueSize;
                    folderHandles.Add(item.ClientHandle);
                }

                await folderSubscription.ApplyChangesAsync(ct);
                folderSubscription.MonitoredItems.ShouldAllBe(m => m.Status.Error == null);
                // The server's revision (MaxEventQueueSize) — what the measured queues actually hold.
                queueSize = folderSubscription.MonitoredItems.First().Status.QueueSize;
            }

            receipts.FolderHandles = folderHandles;

            MonitoredItem? onServer = null;
            if (serverObjectProbe)
            {
                serverSession = await rig.ConnectAsync(anonymous: false, ct);
                var serverSubscription = await SubscribeAsync(serverSession, receipts, isServer: true, ct);
                onServer = RecordedEvents.AddEventItem(serverSubscription, ObjectIds.Server);
                await serverSubscription.ApplyChangesAsync(ct);
                onServer.Status.Error.ShouldBeNull();
                receipts.ServerHandle = onServer.ClientHandle;
            }

            if (stalled && folderSubscription is not null)
            {
                await folderSubscription.SetPublishingModeAsync(false, ct);

                // Fill every folder item's queue to QueueSize (and past it, so the measured reports run at a full queue,
                // discarding oldest), then let any off-Lock consumer (pre-#855: the SDK's channel task) catch up.
                for (var i = 0; i < queueSize + 50; i++) rig.Project(native: true, State($"w{i}"));
                await SettleAsync(rig, n, ct);
            }

            // Warm-up (JIT, first-verdict caches) — not measured.
            for (var i = 0; i < 50; i++) rig.Project(native: true, State($"x{i}"));
            await SettleAsync(rig, n, ct);

            var holds = new double[reports];
            for (var i = 0; i < reports; i++)
            {
                var state = State($"p{i}");
                long t0, t1;
                lock (rig.Nm.Lock)
                {
                    t0 = Stopwatch.GetTimestamp();
                    reportedAt[i] = t0;
                    rig.Project(native: true, state);
                    t1 = Stopwatch.GetTimestamp();
                }

                holds[i] = Stopwatch.GetElapsedTime(t0, t1).TotalMicroseconds;
                // ~1 kHz report rate: an alarm storm, not a single burst, so a consumer that keeps up is not penalised.
                Thread.Sleep(1);
            }

            // Wait for the tail of every probe that is expected to receive.
            var expectFolder = n > 0 && !stalled;
            var deadline = DateTime.UtcNow + TimeSpan.FromSeconds(30);
            while (DateTime.UtcNow < deadline
                   && ((expectFolder && !receipts.FolderComplete(reports - 1, n))
                       || (serverObjectProbe && !receipts.ServerHas(reports - 1))))
            {
                await Task.Delay(50, ct);
            }

            var deferredPeak = rig.Nm.DeferredEventReports;
            var row = new StringBuilder();
            row.Append(CultureInfo.InvariantCulture,
                $"{(serverObjectProbe ? "i=2253" : "folder"),-6} N={n,-3} acl={(acl ? "on " : "off")} {(stalled ? "stalled   " : "publishing")} q={queueSize,-5} | ");
            row.Append(CultureInfo.InvariantCulture, $"lock hold us p50 {Pct(holds, 50),8:F1} p90 {Pct(holds, 90),8:F1} p99 {Pct(holds, 99),8:F1} max {holds.Max(),9:F1} mean {holds.Average(),8:F1}");
            row.Append(CultureInfo.InvariantCulture, $" | per-item mean {(n > 0 ? holds.Average() / n : 0),6:F2} us");
            if (expectFolder)
            {
                var lat = receipts.FolderLatenciesMs(reportedAt, n);
                row.Append(CultureInfo.InvariantCulture,
                    $" | folder all-N latency ms p50 {Pct(lat, 50),6:F1} p99 {Pct(lat, 99),6:F1} max {Max(lat),7:F1} (got {lat.Length}/{reports})");
            }

            if (serverObjectProbe)
            {
                var lat = receipts.ServerLatenciesMs(reportedAt);
                row.Append(CultureInfo.InvariantCulture,
                    $" | i=2253 latency ms p50 {Pct(lat, 50),6:F1} p99 {Pct(lat, 99),6:F1} max {Max(lat),7:F1} (got {lat.Length}/{reports})");
            }

            row.Append(CultureInfo.InvariantCulture, $" | deferred-now {deferredPeak}");
            Report(row.ToString());
        }
        finally
        {
            folderSession?.Dispose();
            serverSession?.Dispose();
        }
    }

    /// <summary>
    ///     Gives an off-<c>Lock</c> consumer (pre-#855) time to drain what was just reported; post-#855 nothing is pending.
    /// </summary>
    private static async Task SettleAsync(ServerEventRig rig, int n, CancellationToken ct)
    {
        var deadline = DateTime.UtcNow + TimeSpan.FromSeconds(60);
        while (rig.Nm.DeferredEventReports > 0 && DateTime.UtcNow < deadline) await Task.Delay(50, ct);
        await Task.Delay(TimeSpan.FromMilliseconds(500 + (10 * n)), ct);
    }

    private static async Task<Subscription> SubscribeAsync(ISession session, Receipts receipts, bool isServer, CancellationToken ct)
    {
        var subscription = new Subscription(session.DefaultSubscription)
        {
            PublishingInterval = (int)PublishingIntervalMs,
            MaxNotificationsPerPublish = 0,
        };
        subscription.FastEventCallback = receipts.OnEvents;
        session.AddSubscription(subscription);
        await subscription.CreateAsync(ct);
        return subscription;
    }

    private static double Pct(double[] values, double p)
    {
        if (values.Length == 0) return double.NaN;
        var sorted = values.Order().ToArray();
        var idx = (int)Math.Ceiling(p / 100.0 * sorted.Length) - 1;
        return sorted[Math.Clamp(idx, 0, sorted.Length - 1)];
    }

    private static double Max(double[] values) => values.Length == 0 ? double.NaN : values.Max();

    private static void Report(string text)
    {
        TestContext.Current.TestOutputHelper?.WriteLine(text);
        if (Environment.GetEnvironmentVariable("OTOPCUA_EVENT_PERF_OUT") is { Length: > 0 } path)
            File.AppendAllText(path, text + Environment.NewLine);
    }

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

    /// <summary>Client-side receipt times of the measured (<c>p{i}</c>) events, per probe.</summary>
    private sealed class Receipts(int reports)
    {
        private readonly object _gate = new();
        private readonly int[] _folderCount = new int[reports];
        private readonly long[] _folderLast = new long[reports];
        private readonly long[] _serverFirst = new long[reports];

        public IReadOnlySet<uint> FolderHandles { get; set; } = new HashSet<uint>();

        public uint? ServerHandle { get; set; }

        public void OnEvents(Subscription subscription, EventNotificationList notification, ArrayOf<string> stringTable)
        {
            var now = Stopwatch.GetTimestamp();
            lock (_gate)
            {
                foreach (var e in notification.Events)
                {
                    var message = RecordedEvents.Message(e.EventFields);
                    if (message is not { Length: > 1 } || message[0] != 'p') continue;
                    if (!int.TryParse(message.AsSpan(1), NumberStyles.Integer, CultureInfo.InvariantCulture, out var i)
                        || i < 0 || i >= _folderCount.Length)
                    {
                        continue;
                    }

                    if (ServerHandle == e.ClientHandle)
                    {
                        if (_serverFirst[i] == 0) _serverFirst[i] = now;
                    }
                    else if (FolderHandles.Contains(e.ClientHandle))
                    {
                        _folderCount[i]++;
                        _folderLast[i] = now;
                    }
                }
            }
        }

        public bool FolderComplete(int i, int n)
        {
            lock (_gate) return _folderCount[i] >= n;
        }

        public bool ServerHas(int i)
        {
            lock (_gate) return _serverFirst[i] != 0;
        }

        public double[] FolderLatenciesMs(long[] reportedAt, int n)
        {
            lock (_gate)
            {
                return Enumerable.Range(0, reportedAt.Length)
                    .Where(i => _folderCount[i] >= n)
                    .Select(i => Stopwatch.GetElapsedTime(reportedAt[i], _folderLast[i]).TotalMilliseconds)
                    .ToArray();
            }
        }

        public double[] ServerLatenciesMs(long[] reportedAt)
        {
            lock (_gate)
            {
                return Enumerable.Range(0, reportedAt.Length)
                    .Where(i => _serverFirst[i] != 0)
                    .Select(i => Stopwatch.GetElapsedTime(reportedAt[i], _serverFirst[i]).TotalMilliseconds)
                    .ToArray();
            }
        }
    }
}