lmxopcua/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect.Tests/MTConnectSampleLagTests.cs

using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;

namespace ZB.MOM.WW.OtOpcUa.Driver.MTConnect.Tests;

/// <summary>
///     Review 2026-10-21 R06:T-5 — <c>/sample</c> throughput was hard-capped at <c>SampleCount / SampleIntervalMs</c>
///     with no lag detection: a machine producing more fell behind the Agent's ring buffer forever, re-baselining
///     every buffer-length with every value backlog-stale and nothing surfacing it. The pump now measures the lag
///     (<c>lastSequence + 1 − nextSequence</c>, plus chunk saturation), surfaces it on the diagnostics bag, and raises
///     the per-stream <c>count</c> when it is persistently behind.
/// </summary>
public sealed class MTConnectSampleLagTests
{
    private const long Instance = 1655000000L;
    private const long Primed = 108L;
    private static readonly TimeSpan Watchdog = TimeSpan.FromSeconds(30);

    private static CancellationToken Ct => TestContext.Current.CancellationToken;

    /// <summary>A chunk of <paramref name="observations"/> dev1_pos values from <paramref name="first"/>.</summary>
    private static MTConnectStreamsResult Chunk(long first, int observations, long? last)
    {
        var obs = Enumerable.Range(0, observations)
            .Select(i => new MTConnectObservation(
                "dev1_pos", (first + i).ToString(System.Globalization.CultureInfo.InvariantCulture),
                new DateTime(2026, 10, 22, 12, 0, 0, DateTimeKind.Utc).AddSeconds(first + i)))
            .ToList();
        return new MTConnectStreamsResult(Instance, first + observations, first, obs, last);
    }

    private static async Task<(MTConnectDriver Driver, CannedAgentClient Client)> SubscribedDriverAsync(int sampleCount)
    {
        var client = CannedAgentClient.FromFixtures();
        var driver = new MTConnectDriver(
            new MTConnectDriverOptions
            {
                AgentUri = "http://fixture-agent:5000",
                RequestTimeoutMs = 5000,
                SampleCount = sampleCount,
                Tags = [new MTConnectTagDefinition("dev1_pos", DriverDataType.Float64)],
            },
            "mt-t5",
            _ => client);
        await driver.InitializeAsync("{}", Ct);
        await driver.SubscribeAsync(["dev1_pos"], TimeSpan.FromMilliseconds(50), Ct);
        await client.WaitForSampleCallsAsync(1).WaitAsync(Watchdog, Ct);
        return (driver, client);
    }

    /// <summary>Three full chunks with a positive lag reopen the stream, at the cursor reached, with count doubled.</summary>
    [Fact]
    public async Task A_persistently_behind_stream_reopens_with_a_doubled_count()
    {
        var (driver, client) = await SubscribedDriverAsync(sampleCount: 2);
        try
        {
            await client.PumpAsync(Chunk(Primed, 2, last: 900)).WaitAsync(Watchdog, Ct);
            await client.PumpAsync(Chunk(Primed + 2, 2, last: 900)).WaitAsync(Watchdog, Ct);
            await client.PumpAsync(Chunk(Primed + 4, 2, last: 900)).WaitAsync(Watchdog, Ct);

            await client.WaitForSampleCallsAsync(2).WaitAsync(Watchdog, Ct);
            client.SampleCounts.ShouldBe([2, 4]);
            client.LastSampleFrom.ShouldBe(Primed + 6, "the raised stream resumes where the behind one got to");

            var diag = driver.GetDiagnostics();
            diag["sampleCount"].ShouldBe("4");
            diag["sampleLag"].ShouldBe((900 + 1 - (Primed + 6)).ToString(System.Globalization.CultureInfo.InvariantCulture));
            diag["sampleSaturatedChunks"].ShouldBe("3");
            diag["sampleCountRaises"].ShouldBe("1");
        }
        finally
        {
            await driver.ShutdownAsync(CancellationToken.None);
        }
    }

    /// <summary>An Agent that omits lastSequence: persistent saturation alone is enough.</summary>
    [Fact]
    public async Task Persistent_saturation_without_a_lastSequence_still_raises_the_count()
    {
        var (driver, client) = await SubscribedDriverAsync(sampleCount: 2);
        try
        {
            for (var i = 0; i < 3; i++)
                await client.PumpAsync(Chunk(Primed + (2 * i), 2, last: null)).WaitAsync(Watchdog, Ct);

            await client.WaitForSampleCallsAsync(2).WaitAsync(Watchdog, Ct);
            client.SampleCounts.ShouldBe([2, 4]);
            driver.GetDiagnostics().ContainsKey("sampleLag").ShouldBeFalse("an unknown lag is not reported as a number");
        }
        finally
        {
            await driver.ShutdownAsync(CancellationToken.None);
        }
    }

    /// <summary>
    ///     Full chunks with NO lag (the Agent happened to have exactly count observations) are not "behind": the stream
    ///     is kept. Positive control: the first test, through the same pump, does reopen.
    /// </summary>
    [Fact]
    public async Task Full_chunks_with_no_lag_keep_the_stream()
    {
        var (driver, client) = await SubscribedDriverAsync(sampleCount: 2);
        try
        {
            for (var i = 0; i < 4; i++)
            {
                var first = Primed + (2 * i);
                await client.PumpAsync(Chunk(first, 2, last: first + 1)).WaitAsync(Watchdog, Ct);
            }

            client.SampleCallCount.ShouldBe(1, "a caught-up stream must not be reopened");
            driver.GetDiagnostics()["sampleLag"].ShouldBe("0");
            driver.GetDiagnostics()["sampleCountRaises"].ShouldBe("0");
        }
        finally
        {
            await driver.ShutdownAsync(CancellationToken.None);
        }
    }

    /// <summary>A raised count the Agent refuses (its stream delivers nothing) steps back to the last count that worked.</summary>
    [Fact]
    public async Task A_refused_raised_count_steps_back_to_the_last_count_that_delivered()
    {
        var (driver, client) = await SubscribedDriverAsync(sampleCount: 2);
        try
        {
            // The NEXT enumeration (the raised one) is refused the way a real Agent refuses a count above its buffer.
            // Enqueued only once the first stream has consumed a chunk — i.e. is past its own one-shot-failure check —
            // so the failure can only ever hit the raised stream (enqueueing earlier raced the first enumeration).
            await client.PumpAsync(Chunk(Primed, 2, last: 900)).WaitAsync(Watchdog, Ct);
            client.SampleFailures.Enqueue(new InvalidDataException("MTConnectError: INVALID_REQUEST count too large"));
            for (var i = 1; i < 3; i++)
                await client.PumpAsync(Chunk(Primed + (2 * i), 2, last: 900)).WaitAsync(Watchdog, Ct);

            await client.WaitForSampleCallsAsync(3).WaitAsync(Watchdog, Ct);
            client.SampleCounts.Take(3).ShouldBe([2, 4, 2]);
            driver.GetDiagnostics()["sampleCount"].ShouldBe("2");
        }
        finally
        {
            await driver.ShutdownAsync(CancellationToken.None);
        }
    }

    /// <summary>The ceiling is 16× the configured count, capped at 65 536, and never below the configured count.</summary>
    [Theory]
    [InlineData(1000, 16000)]
    [InlineData(10_000, 65_536)]
    [InlineData(100_000, 100_000)]
    public void The_count_ceiling_is_bounded(int configured, int expected) =>
        MTConnectDriver.SampleCountCeiling(configured).ShouldBe(expected);
}