lmxopcua/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.FOCAS.Tests/FocasConcurrentDeviceReadTests.cs

using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Driver.FOCAS;

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

/// <summary>
///     #740 / R07:P9 — <see cref="FocasDriver.ReadAsync"/> walked the whole batch sequentially, so a SLOW CNC
///     (TCP-accept-then-stall, the live 31i-B case) cost its timeout per tag per poll during which no other
///     device was read. Devices are now read concurrently, each still serialised on its own socket.
/// </summary>
[Trait("Category", "Unit")]
public sealed class FocasConcurrentDeviceReadTests
{
    private const string HostA = "focas://10.0.0.5:8193";
    private const string HostB = "focas://10.0.0.6:8193";

    /// <summary>
    ///     A client that records which device it dialled. Device A's read only completes once device B's read has
    ///     STARTED — so a sequential walk (A first) can never finish A, while a concurrent one does. Also counts
    ///     the peak number of reads in flight on any one client.
    /// </summary>
    private sealed class OrderingClient(TaskCompletionSource bStarted, int[] global) : FakeFocasClient
    {
        private string _host = "";
        private int _inFlight;
        public int PeakInFlight;
        public int Reads;

        public override Task ConnectAsync(FocasHostAddress address, TimeSpan timeout, CancellationToken ct)
        {
            _host = address.ToString();
            return base.ConnectAsync(address, timeout, ct);
        }

        public override async Task<(object? value, uint status)> ReadAsync(
            FocasAddress address, FocasDataType type, CancellationToken ct)
        {
            var now = Interlocked.Increment(ref _inFlight);
            InterlockedMax(ref PeakInFlight, now);
            Interlocked.Increment(ref Reads);
            // global[0] = reads in flight across EVERY client, global[1] = its peak.
            InterlockedMax(ref global[1], Interlocked.Increment(ref global[0]));
            try
            {
                if (_host.Contains("10.0.0.6", StringComparison.Ordinal)) bStarted.TrySetResult();
                else await bStarted.Task.WaitAsync(TimeSpan.FromSeconds(2), ct).ConfigureAwait(false);
                await Task.Delay(20, ct).ConfigureAwait(false);
                return ((sbyte)1, FocasStatusMapper.Good);
            }
            finally
            {
                Interlocked.Decrement(ref _inFlight);
                Interlocked.Decrement(ref global[0]);
            }
        }

        private static void InterlockedMax(ref int target, int value)
        {
            int seen;
            while ((seen = Volatile.Read(ref target)) < value
                   && Interlocked.CompareExchange(ref target, value, seen) != seen) { }
        }
    }

    /// <summary>A stalled device no longer holds every other device's read hostage — and each device stays serialised.</summary>
    [Fact]
    public async Task Devices_are_read_concurrently_and_each_stays_serialised()
    {
        var bStarted = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
        var global = new int[2];
        var factory = new FakeFocasClientFactory { Customise = () => new OrderingClient(bStarted, global) };
        var drv = new FocasDriver(new FocasDriverOptions
        {
            Devices = [new FocasDeviceOptions(HostA, DeviceName: "A"), new FocasDeviceOptions(HostB, DeviceName: "B")],
            RawTags =
            [
                FocasRawTags.Entry(new FocasTagDefinition("raw/A1", "A", "R100", FocasDataType.Byte)),
                FocasRawTags.Entry(new FocasTagDefinition("raw/A2", "A", "R101", FocasDataType.Byte)),
                FocasRawTags.Entry(new FocasTagDefinition("raw/B1", "B", "R100", FocasDataType.Byte)),
            ],
            Probe = new FocasProbeOptions { Enabled = false },
            // Off so every tag takes the per-tag path — this pins device-level concurrency, not coalescing.
            PmcCoalesce = new FocasPmcCoalesceOptions { Enabled = false },
        }, "focas-p9", factory);
        await drv.InitializeAsync("{}", CancellationToken.None);

        var snaps = await drv.ReadAsync(["raw/A1", "raw/A2", "raw/B1"], CancellationToken.None);

        snaps.ShouldAllBe(s => s.StatusCode == FocasStatusMapper.Good,
            "device A's read can only finish once device B's has started — a sequential walk times A out");
        factory.Clients.Count.ShouldBe(2);
        var clients = factory.Clients.Cast<OrderingClient>().ToList();
        // Every client really served reads (2 on A, 1 on B), so "peak 1" is measured, not vacuous …
        clients.Select(c => c.Reads).OrderBy(n => n).ShouldBe([1, 2]);
        clients.ShouldAllBe(c => c.PeakInFlight == 1,
            "one device's socket is still request→response: never two reads in flight on it");
        // … and the two devices overlapped: two reads were in flight at once across the clients.
        global[1].ShouldBe(2, "the devices must be read at the same time, not one after the other");
        await drv.ShutdownAsync(CancellationToken.None);
    }

    /// <summary>A device that accepted TCP and then stopped answering: every read hangs until the call timeout.</summary>
    private sealed class StalledClient : FakeFocasClient
    {
        public int Reads;

        public override async Task<(object? value, uint status)> ReadAsync(
            FocasAddress address, FocasDataType type, CancellationToken ct)
        {
            Interlocked.Increment(ref Reads);
            await Task.Delay(Timeout.Infinite, ct).ConfigureAwait(false);
            return (null, FocasStatusMapper.Good);
        }
    }

    /// <summary>
    ///     #740 review I1 — <c>Task.WhenAll</c> waits for the slowest device, so concurrency alone still made a
    ///     stalled device cost (its tag count × the call timeout) per batch, and the healthy device's values waited
    ///     that long. After the first timeout on a device its remaining tags in the batch fail fast: the batch is
    ///     bounded by about ONE call timeout and the healthy device's values are Good.
    /// </summary>
    [Fact]
    public async Task A_stalled_device_costs_one_call_timeout_per_batch_not_one_per_tag()
    {
        var callTimeout = TimeSpan.FromMilliseconds(200);
        var factory = new FakeFocasClientFactory();
        factory.Customise = () => factory.Clients.Count == 0 ? new StalledClient() : new FakeFocasClient { Values = { ["R100"] = (sbyte)9 } };
        var stalledTags = Enumerable.Range(0, 10)
            .Select(i => FocasRawTags.Entry(new FocasTagDefinition($"raw/A{i}", "A", $"R{100 + i}", FocasDataType.Byte)));
        var drv = new FocasDriver(new FocasDriverOptions
        {
            Devices = [new FocasDeviceOptions(HostA, DeviceName: "A"), new FocasDeviceOptions(HostB, DeviceName: "B")],
            RawTags = [.. stalledTags, FocasRawTags.Entry(new FocasTagDefinition("raw/B1", "B", "R100", FocasDataType.Byte))],
            Probe = new FocasProbeOptions { Enabled = false },
            PmcCoalesce = new FocasPmcCoalesceOptions { Enabled = false },
            Timeout = callTimeout,
        }, "focas-stall", factory);
        await drv.InitializeAsync("{}", CancellationToken.None);
        // Connect device A first so it gets the stalled client (connect order = client order).
        await drv.ReadAsync(["raw/A0"], CancellationToken.None);

        var refs = Enumerable.Range(0, 10).Select(i => $"raw/A{i}").Append("raw/B1").ToArray();
        var sw = System.Diagnostics.Stopwatch.StartNew();
        var snaps = await drv.ReadAsync(refs, CancellationToken.None);
        sw.Stop();

        sw.Elapsed.ShouldBeLessThan(callTimeout * 3, "one stalled device must cost ~one call timeout, not ten");
        snaps[10].StatusCode.ShouldBe(FocasStatusMapper.Good);
        snaps[10].Value.ShouldBe((sbyte)9);
        snaps.Take(10).ShouldAllBe(s => s.StatusCode == FocasStatusMapper.BadCommunicationError);
        ((StalledClient)factory.Clients[0]).Reads.ShouldBe(2, "one read in the warm-up batch, one in the measured batch");
        await drv.ShutdownAsync(CancellationToken.None);
    }

    /// <summary>Results land at their own indices whatever order the devices finish in.</summary>
    [Fact]
    public async Task Results_keep_their_request_order_across_devices()
    {
        var factory = new FakeFocasClientFactory();
        var drv = new FocasDriver(new FocasDriverOptions
        {
            Devices = [new FocasDeviceOptions(HostA, DeviceName: "A"), new FocasDeviceOptions(HostB, DeviceName: "B")],
            RawTags =
            [
                FocasRawTags.Entry(new FocasTagDefinition("raw/A1", "A", "R100", FocasDataType.Byte)),
                FocasRawTags.Entry(new FocasTagDefinition("raw/B1", "B", "R200", FocasDataType.Byte)),
            ],
            Probe = new FocasProbeOptions { Enabled = false },
        }, "focas-p9-order", factory)
        ;
        factory.Customise = () => new FakeFocasClient { Values = { ["R100"] = (sbyte)1, ["R200"] = (sbyte)2 } };
        await drv.InitializeAsync("{}", CancellationToken.None);

        var snaps = await drv.ReadAsync(["raw/B1", "nope", "raw/A1"], CancellationToken.None);

        snaps[0].Value.ShouldBe((sbyte)2);
        snaps[1].StatusCode.ShouldBe(FocasStatusMapper.BadNodeIdUnknown);
        snaps[2].Value.ShouldBe((sbyte)1);
        await drv.ShutdownAsync(CancellationToken.None);
    }
}