lmxopcua/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.IntegrationTests/Differential/PollLatencyComparisonTests.cs

using ZB.MOM.WW.OtOpcUa.Driver.Ab.LibplctagShared;
using System.Globalization;
using Shouldly;
using Xunit;

namespace ZB.MOM.WW.OtOpcUa.Driver.AbCip.IntegrationTests.Differential;

/// <summary>
///     Task 21 latency comparison: a fixed set of atomic tags polled through libplctag, then through the native
///     transport, each pass issuing every read at once. Pass criterion: native p95 ≤ libplctag p95 × 1.10.
///     <para>Gated by the same variables as the differential (<c>AB_DIFF_ENDPOINT</c> absent ⇒ skip), in the same
///     serialized collection, and read-only under the same source/IL guard
///     (<see cref="TransportDifferentialTests.The_harness_is_read_only_by_construction"/>). It runs AFTER the
///     differential's one run, reusing its browse for tag selection, so the target sees one listing per transport,
///     not two. Knobs: <c>AB_LAT_TAGS</c>, <c>AB_LAT_PASSES</c>, <c>AB_LAT_PASS_GAP_MS</c> (see
///     <see cref="DifferentialTarget"/>).</para>
/// </summary>
[Collection(DifferentialCollection.Name)]
[Trait("Category", "LiveIntegration")]
public sealed class PollLatencyComparisonTests(DifferentialRunFixture fixture)
{
    private const double Tolerance = 1.10;
    private static readonly TimeSpan RunBudget = TimeSpan.FromMinutes(10);

    [Fact]
    public async Task Native_poll_p95_is_within_ten_percent_of_libplctag()
    {
        var target = DifferentialTarget.FromEnvironment();
        if (target is null) Assert.Skip(DifferentialTarget.SkipReason);
        var differential = await fixture.GetAsync();

        using var cts = new CancellationTokenSource(RunBudget);
        // lmxopcua#769: the shared ab_server fixture occasionally resets a connection under load ("connection reset by
        // peer"). That is a fixture transport fault, not a latency finding, so the run is retried once when it surfaces
        // as a thrown reset or a reported problem; any other failure (or a second reset) still fails the test.
        LatencyResult r;
        try { r = await PollLatencyRun.ExecuteAsync(target, differential, cts.Token); }
        catch (Exception ex) when (IsFixtureReset(ex.ToString()))
        {
            Log($"Fixture transport reset on the first attempt, retrying once: {ex.Message}");
            r = await PollLatencyRun.ExecuteAsync(target, differential, cts.Token);
        }
        if (r.Problems.Any(p => IsFixtureReset(p)))
        {
            Log($"Fixture transport reset reported as a problem on the first attempt, retrying once: {string.Join("; ", r.Problems)}");
            r = await PollLatencyRun.ExecuteAsync(target, differential, cts.Token);
        }

        Log($"## Latency — {r.Target}");
        Log($"Tags per pass: {r.Tags.Count} ({r.DistinctTags} distinct" +
            (r.Repeated ? $", REPEATED cyclically to reach {DifferentialTarget.LatencyTagsVar}={r.Target.LatencyTags}" : "") +
            $"); passes per transport: {r.Target.LatencyPasses}; gap between passes: {r.Target.LatencyPassGapMs} ms.");
        Log("Types: " + string.Join(", ", r.Tags.GroupBy(t => t.Type).Select(g => $"{g.Key} {g.Count()}")));
        if (r.Dropped.Count > 0) Log($"Dropped (libplctag refused at setup, removed from both sets): {string.Join(", ", r.Dropped)}");
        Log($"Native requests spent on setup (one read per tag): {r.NativeSetupRequests}; native Forward Opens: {r.NativeForwardOpens}; " +
            $"native Forward Closes sent/acked: {r.NativeForwardCloses.Sent}/{r.NativeForwardCloses.Acked}; " +
            $"connections after the native phase: {r.ConnectionsAfterNative?.ToString(CultureInfo.InvariantCulture) ?? "unobservable"}.");
        Log($"libplctag orphans destroyed so far: {LibplctagOrphanRelease.Released}; rooted (unreleasable): {LibplctagOrphanRelease.RootedCount}.");
        Log("");

        var libP50 = PollLatencyRun.Percentile(r.Libplctag.Select(p => p.Wall), 50);
        var libP95 = PollLatencyRun.Percentile(r.Libplctag.Select(p => p.Wall), 95);
        var natP50 = PollLatencyRun.Percentile(r.Native.Select(p => p.Wall), 50);
        var natP95 = PollLatencyRun.Percentile(r.Native.Select(p => p.Wall), 95);
        var limit = libP95 * Tolerance;

        Log("| Transport | Passes | p50 ms | p95 ms | min ms | max ms | CIP requests per pass (median) | Tag failures |");
        Log("|---|---|---|---|---|---|---|---|");
        Log(Row("libplctag", r.Libplctag, libP50, libP95));
        Log(Row("native", r.Native, natP50, natP95));
        Log("");
        Log($"Criterion: native p95 {Ms(natP95)} ms ≤ libplctag p95 {Ms(libP95)} ms × {Tolerance:0.00} = {Ms(limit)} ms → " +
            (natP95 <= limit ? "PASS" : "**FAIL**"));
        Log("");
        // Attribution (review r1 M8): GC collections during each pass (gen0/gen1/gen2) on both sides, and the
        // round trip of each native CIP exchange — a slow pass with fast exchanges is a client-side stall.
        Log("| Pass | libplctag ms | libplctag GC 0/1/2 | native ms | native GC 0/1/2 | native requests | native RTT min/mean/max ms |");
        Log("|---|---|---|---|---|---|---|");
        for (var i = 0; i < Math.Max(r.Libplctag.Count, r.Native.Count); i++)
        {
            var lib = i < r.Libplctag.Count ? r.Libplctag[i] : null;
            var nat = i < r.Native.Count ? r.Native[i] : null;
            Log($"| {i + 1} | {(lib is null ? "-" : Ms(lib.Wall.TotalMilliseconds))} | {lib?.Gc.ToString() ?? "-"} | " +
                $"{(nat is null ? "-" : Ms(nat.Wall.TotalMilliseconds))} | {nat?.Gc.ToString() ?? "-"} | " +
                $"{nat?.Requests?.ToString(CultureInfo.InvariantCulture) ?? "-"} | {Rtt(nat?.Rtt)} |");
        }
        foreach (var p in r.Problems) Log($"- PROBLEM: {p}");

        r.Problems.ShouldBeEmpty();
        r.Libplctag.Count.ShouldBe(r.Target.LatencyPasses);
        r.Native.Count.ShouldBe(r.Target.LatencyPasses);
        r.Libplctag.Sum(p => p.TagFailures).ShouldBe(0, "every timed libplctag read must succeed");
        r.Native.Sum(p => p.TagFailures).ShouldBe(0, "every timed native read must succeed");
        if (r.Target.CountOpens)
        {
            r.NativeForwardOpens.ShouldBe(1);
            r.NativeForwardCloses.Sent.ShouldBe(1, "the native phase's one connection must be closed with exactly one Forward Close");
        }
        if (r.ConnectionsAfterNative is { } after && differential.ConnectionsBaseline is { } baseline)
            after.ShouldBeLessThanOrEqualTo(baseline, "the native session must be gone when the run reports");
        // Review r1 M4: the latency setup's refused libplctag tags are orphan candidates too — checked after them.
        LibplctagOrphanRelease.RootedCount.ShouldBe(0);
        // The p95 comparison is a perf criterion against a shared fixture and a loaded host: it is reported always and
        // enforced only when AB_LAT_ENFORCE=1 (like OTOPCUA_SCRIPT_PERF); the correctness checks above always run.
        if (Environment.GetEnvironmentVariable("AB_LAT_ENFORCE") == "1")
            natP95.ShouldBeLessThanOrEqualTo(limit, $"native p95 must be ≤ libplctag p95 × {Tolerance:0.00}");
        else
            Log($"(p95 criterion not enforced: set AB_LAT_ENFORCE=1 to fail on it; outcome above is {(natP95 <= limit ? "PASS" : "FAIL")})");
    }

    /// <summary>Offline: the tag pick takes atomic scalars only, and repeats a short set deliberately and visibly.</summary>
    [Fact]
    public void Tag_pick_takes_atomic_scalars_only_and_repeats_a_short_set_visibly()
    {
        var browsed = new TransportObservations();
        browsed.Browse.Add(new AbCipDiscoveredTag("D", null, AbCipDataType.DInt, false));
        browsed.Browse.Add(new AbCipDiscoveredTag("R", null, AbCipDataType.Real, false));
        browsed.Browse.Add(new AbCipDiscoveredTag("S", null, AbCipDataType.String, false));
        browsed.Browse.Add(new AbCipDiscoveredTag("T", null, AbCipDataType.Dt, false));
        browsed.Browse.Add(new AbCipDiscoveredTag("A", null, AbCipDataType.DInt, false, ElementCount: 4, IsArray: true));
        browsed.Browse.Add(new AbCipDiscoveredTag("Sys", null, AbCipDataType.DInt, false, IsSystemTag: true));
        var target = new DifferentialTarget("127.0.0.1", 1, "1,0", AbCipPlcFamily.CompactLogix, null, 200, false, LatencyTags: 5);

        var (tags, distinct, repeated) = PollLatencyRun.PickTags(target, browsed);
        distinct.ShouldBe(2);
        repeated.ShouldBeTrue();
        tags.Select(t => t.Name).ShouldBe(["D", "R", "D", "R", "D"]);

        var (one, oneDistinct, oneRepeated) = PollLatencyRun.PickTags(target with { LatencyTags = 1 }, browsed);
        one.ShouldHaveSingleItem().Name.ShouldBe("D");
        oneDistinct.ShouldBe(1);
        oneRepeated.ShouldBeFalse();

        PollLatencyRun.Percentile([TimeSpan.FromMilliseconds(3), TimeSpan.FromMilliseconds(1), TimeSpan.FromMilliseconds(2)], 50)
            .ShouldBe(2);
        PollLatencyRun.Percentile(Enumerable.Range(1, 20).Select(i => TimeSpan.FromMilliseconds(i)), 95).ShouldBe(19);
    }

    private static string Row(string name, IReadOnlyList<LatencyPass> passes, double p50, double p95)
    {
        var walls = passes.Select(p => p.Wall.TotalMilliseconds).ToList();
        var requests = passes.Where(p => p.Requests is not null).Select(p => p.Requests!.Value).Order().ToList();
        var medianRequests = requests.Count == 0 ? "n/a" : requests[requests.Count / 2].ToString(CultureInfo.InvariantCulture);
        return $"| {name} | {passes.Count} | {Ms(p50)} | {Ms(p95)} | {Ms(walls.Min())} | {Ms(walls.Max())} | {medianRequests} | " +
               $"{passes.Sum(p => p.TagFailures)} |";
    }

    private static string Rtt(ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Cip.ConnectionDiagnostics.ExchangeStats? s) =>
        s is not { Count: > 0 } x ? "-" : $"{Ms(x.MinMs)}/{Ms(x.MeanMs)}/{Ms(x.MaxMs)}";

    private static bool IsFixtureReset(string text) =>
        text.Contains("reset by peer", StringComparison.OrdinalIgnoreCase)
        || text.Contains("connection reset", StringComparison.OrdinalIgnoreCase);

    private static string Ms(double ms) => ms.ToString("0.0", CultureInfo.InvariantCulture);

    private static void Log(string line) => TestContext.Current.TestOutputHelper?.WriteLine(line);
}