lmxopcua/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbLegacy.Tests/AbLegacyDeviceIsolationTests.cs

using System.Diagnostics;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
using ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport;

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

/// <summary>
///     Review 07 P7 (High): reads were strictly serial across every device with no per-device backoff, so one dead
///     device cost <c>Timeout</c> per tag per poll and starved every healthy device — long enough to trip the
///     comms-loss watchdog against a reachable one. Devices are now read in parallel (bounded), each device pass
///     stops paying for a dead device after two consecutive comms faults and opens a per-device
///     <see cref="ConnectionBackoff"/> window, and cold connects are single-flight per device.
/// </summary>
[Trait("Category", "Unit")]
public sealed class AbLegacyDeviceIsolationTests
{
    private const string HostA = "ab://10.0.0.5/1,0";
    private const string HostB = "ab://10.0.0.6/1,0";

    /// <summary>A tag on a dead device: each wire operation burns <paramref name="cost"/> and then fails.</summary>
    private sealed class DeadTag(AbLegacyTagCreateParams p, TimeSpan cost) : FakeAbLegacyTag(p)
    {
        public static int WireOps;

        public override async Task InitializeAsync(CancellationToken ct)
        {
            Interlocked.Increment(ref WireOps);
            await Task.Delay(cost, ct);
            throw new System.Net.Sockets.SocketException(10060);
        }

        public override async Task ReadAsync(CancellationToken ct)
        {
            Interlocked.Increment(ref WireOps);
            await Task.Delay(cost, ct);
            throw new System.Net.Sockets.SocketException(10060);
        }
    }

    private static AbLegacyDriver NewDriver(FakeAbLegacyTagFactory factory, params AbLegacyTagDefinition[] tags) =>
        new(new AbLegacyDriverOptions
        {
            Devices = [new AbLegacyDeviceOptions(HostA), new AbLegacyDeviceOptions(HostB)],
            RawTags = AbLegacyRawTags.Entries(tags),
            Probe = new AbLegacyProbeOptions { Enabled = false },
            CoalesceReads = false, // per-tag isolation is under test here; coalescing has its own tests
        }, "drv-p7", factory);

    private static AbLegacyTagDefinition[] Tags(string prefix, string host, int n, string file = "N7") =>
        [.. Enumerable.Range(0, n).Select(i => new AbLegacyTagDefinition($"{prefix}{i}", host, $"{file}:{i * 3}", AbLegacyDataType.Int))];

    [Fact]
    public async Task Dead_device_costs_at_most_two_timeouts_then_fails_fast_and_never_stalls_the_healthy_one()
    {
        var cost = TimeSpan.FromMilliseconds(150);
        var factory = new FakeAbLegacyTagFactory
        {
            Customise = p => p.Gateway == "10.0.0.6" ? new DeadTag(p, cost) : new FakeAbLegacyTag(p) { Value = 7 },
        };
        var a = Tags("A", HostA, 5);
        var b = Tags("B", HostB, 20);
        var drv = NewDriver(factory, [.. a, .. b]);
        await drv.InitializeAsync("{}", CancellationToken.None);
        var refs = a.Concat(b).Select(t => t.Name).ToList();
        var before = drv.GetHealth().LastSuccessfulRead;
        Interlocked.Exchange(ref DeadTag.WireOps, 0);

        var sw = Stopwatch.StartNew();
        var first = await drv.ReadAsync(refs, CancellationToken.None);
        sw.Stop();

        // Serial-with-no-backoff would pay 20 x 150 ms = 3 s on device B alone.
        sw.Elapsed.ShouldBeLessThan(TimeSpan.FromSeconds(1.5));
        Volatile.Read(ref DeadTag.WireOps).ShouldBeLessThanOrEqualTo(2);
        first.Take(5).ShouldAllBe(s => s.StatusCode == AbLegacyStatusMapper.Good); // positive control: A is served
        first.Skip(5).ShouldAllBe(s => s.StatusCode == AbLegacyStatusMapper.BadCommunicationError);
        var health = drv.GetHealth();
        health.State.ShouldBe(DriverState.Degraded);
        health.LastSuccessfulRead.ShouldNotBeNull();
        health.LastSuccessfulRead!.Value.ShouldBeGreaterThan(before ?? DateTime.MinValue); // A's success advances it

        // Inside B's backoff window the next poll touches B's wire not at all.
        var opsAfterFirst = Volatile.Read(ref DeadTag.WireOps);
        sw.Restart();
        var second = await drv.ReadAsync(refs, CancellationToken.None);
        sw.Stop();
        Volatile.Read(ref DeadTag.WireOps).ShouldBe(opsAfterFirst);
        sw.Elapsed.ShouldBeLessThan(TimeSpan.FromMilliseconds(500));
        second.Take(5).ShouldAllBe(s => s.StatusCode == AbLegacyStatusMapper.Good);
        await drv.ShutdownAsync(CancellationToken.None);
    }

    [Fact]
    public async Task Devices_are_read_in_parallel()
    {
        // Device A's read cannot complete until device B's read has STARTED: a serial walk deadlocks here.
        var bStarted = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
        var factory = new FakeAbLegacyTagFactory
        {
            Customise = p => p.Gateway == "10.0.0.5"
                ? new GatedTag(p, wait: bStarted.Task) { Value = 1 }
                : new GatedTag(p, signal: bStarted) { Value = 2 },
        };
        var drv = NewDriver(factory,
            new AbLegacyTagDefinition("A", HostA, "N7:0", AbLegacyDataType.Int),
            new AbLegacyTagDefinition("B", HostB, "N7:0", AbLegacyDataType.Int));
        await drv.InitializeAsync("{}", CancellationToken.None);

        var r = await drv.ReadAsync(["A", "B"], CancellationToken.None).WaitAsync(TimeSpan.FromSeconds(5));

        r[0].Value.ShouldBe(1);
        r[1].Value.ShouldBe(2);
        await drv.ShutdownAsync(CancellationToken.None);
    }

    private sealed class GatedTag(AbLegacyTagCreateParams p, Task? wait = null, TaskCompletionSource? signal = null)
        : FakeAbLegacyTag(p)
    {
        public override async Task ReadAsync(CancellationToken ct)
        {
            signal?.TrySetResult();
            if (wait is not null) await wait.WaitAsync(ct);
            await base.ReadAsync(ct);
        }
    }

    [Fact]
    public async Task One_timing_out_tag_does_not_blacken_the_rest_of_its_device()
    {
        var factory = new FakeAbLegacyTagFactory
        {
            Customise = p => p.TagName == "N7:0"
                ? new FakeAbLegacyTag(p) { ThrowOnRead = true, Exception = new TimeoutException() }
                : new FakeAbLegacyTag(p) { Value = 5 },
        };
        var drv = NewDriver(factory, Tags("A", HostA, 4));
        await drv.InitializeAsync("{}", CancellationToken.None);

        for (var poll = 0; poll < 3; poll++)
        {
            var r = await drv.ReadAsync(["A0", "A1", "A2", "A3"], CancellationToken.None);
            r[0].StatusCode.ShouldBe(AbLegacyStatusMapper.BadCommunicationError);
            r.Skip(1).ShouldAllBe(s => s.StatusCode == AbLegacyStatusMapper.Good, $"poll {poll}");
        }
        drv.GetHealth().State.ShouldBe(DriverState.Degraded);
        await drv.ShutdownAsync(CancellationToken.None);
    }

    [Fact]
    public async Task Device_refusal_is_per_tag_Bad_advances_LastSuccessfulRead_and_keeps_the_handle()
    {
        var logger = new AbLegacyTestLogger();
        var factory = new FakeAbLegacyTagFactory
        {
            Customise = p => new FakeAbLegacyTag(p) { Status = (int)AbStatus.NotFound },
        };
        var drv = new AbLegacyDriver(new AbLegacyDriverOptions
        {
            Devices = [new AbLegacyDeviceOptions(HostA)],
            RawTags = AbLegacyRawTags.Entries(new AbLegacyTagDefinition("X", HostA, "N7:0", AbLegacyDataType.Int)),
            Probe = new AbLegacyProbeOptions { Enabled = false },
        }, "drv-p7", factory, logger);
        await drv.InitializeAsync("{}", CancellationToken.None);
        var initHealth = drv.GetHealth();
        await Task.Delay(20);

        var r1 = await drv.ReadAsync(["X"], CancellationToken.None);
        var r2 = await drv.ReadAsync(["X"], CancellationToken.None);

        r1[0].StatusCode.ShouldBe(AbLegacyStatusMapper.BadNodeIdUnknown);
        r2[0].StatusCode.ShouldBe(AbLegacyStatusMapper.BadNodeIdUnknown);
        var health = drv.GetHealth();
        health.State.ShouldBe(DriverState.Healthy); // the device answered: not a comms fault
        health.LastSuccessfulRead!.Value.ShouldBeGreaterThan(initHealth.LastSuccessfulRead!.Value);
        factory.Tags["N7:0"].InitializeCount.ShouldBe(1);
        factory.Tags["N7:0"].ReadCount.ShouldBe(2); // positive control: both polls really asked the device
        logger.Warnings.Count(w => w.Contains("Reference=X")).ShouldBe(1);
        await drv.ShutdownAsync(CancellationToken.None);
    }

    [Fact]
    public async Task Cold_connects_on_one_device_are_single_flight()
    {
        var inFlight = 0;
        var maxInFlight = 0;
        var factory = new FakeAbLegacyTagFactory
        {
            Customise = p => new CountingInitTag(p, () =>
            {
                var now = Interlocked.Increment(ref inFlight);
                int seen;
                while (now > (seen = Volatile.Read(ref maxInFlight)) && Interlocked.CompareExchange(ref maxInFlight, now, seen) != seen) { }
            }, () => Interlocked.Decrement(ref inFlight)) { Value = 1 },
        };
        var tags = Tags("A", HostA, 6);
        var drv = NewDriver(factory, tags);
        await drv.InitializeAsync("{}", CancellationToken.None);

        // Three concurrent callers (poll loop, server read, write) each hitting different cold tags of one device.
        await Task.WhenAll(
            drv.ReadAsync(["A0", "A1"], CancellationToken.None),
            drv.ReadAsync(["A2", "A3"], CancellationToken.None),
            drv.WriteAsync([new WriteRequest("A4", 3), new WriteRequest("A5", 4)], CancellationToken.None));

        maxInFlight.ShouldBe(1);
        factory.Tags.Values.Count(t => t.InitializeCount == 1).ShouldBe(6); // positive control: all six connected
        await drv.ShutdownAsync(CancellationToken.None);
    }

    private sealed class CountingInitTag(AbLegacyTagCreateParams p, Action enter, Action exit) : FakeAbLegacyTag(p)
    {
        public override async Task InitializeAsync(CancellationToken ct)
        {
            enter();
            try
            {
                await Task.Delay(30, ct);
                await base.InitializeAsync(ct);
            }
            finally { exit(); }
        }
    }

    [Fact]
    public async Task A_recovered_device_is_read_again_once_its_window_expires()
    {
        var dead = true;
        var creates = 0;
        var factory = new FakeAbLegacyTagFactory
        {
            Customise = p =>
            {
                Interlocked.Increment(ref creates);
                return dead
                    ? new FakeAbLegacyTag(p) { ThrowOnInitialize = true, Exception = new System.Net.Sockets.SocketException(10061) }
                    : new FakeAbLegacyTag(p) { Value = 9 };
            },
        };
        var drv = NewDriver(factory, new AbLegacyTagDefinition("B", HostB, "N7:0", AbLegacyDataType.Int));
        await drv.InitializeAsync("{}", CancellationToken.None);

        // Two consecutive comms faults mark the device down and open its first (1 s) window.
        (await drv.ReadAsync(["B"], CancellationToken.None))[0].StatusCode.ShouldBe(AbLegacyStatusMapper.BadCommunicationError);
        (await drv.ReadAsync(["B"], CancellationToken.None))[0].StatusCode.ShouldBe(AbLegacyStatusMapper.BadCommunicationError);
        var attempts = Volatile.Read(ref creates);
        dead = false;
        // Inside the window: fails fast without a connect attempt, although the device is back.
        (await drv.ReadAsync(["B"], CancellationToken.None))[0].StatusCode.ShouldBe(AbLegacyStatusMapper.BadCommunicationError);
        Volatile.Read(ref creates).ShouldBe(attempts);
        await Task.Delay(TimeSpan.FromMilliseconds(1100));

        var recovered = await drv.ReadAsync(["B"], CancellationToken.None);
        recovered[0].StatusCode.ShouldBe(AbLegacyStatusMapper.Good);
        recovered[0].Value.ShouldBe(9);
        drv.GetHealth().State.ShouldBe(DriverState.Healthy);
        await drv.ShutdownAsync(CancellationToken.None);
    }
}