fix(modbus): transport self-heals with one throttled in-gate connect; PDU length check

P3: a torn-down socket is re-dialled inside the gate (one attempt per backoff window) instead of failing fast with 'Transport not connected' until the 90 s comms-loss watchdog re-initialised the driver. L3: ConnectAsync disposes the previous/failed TcpClient, DNS runs under the per-op timeout, and the reconnect never sleeps inside the gate (the old 10-attempt backoff held it for ~81 s). H9: an MBAP response with no PDU, or an exception PDU with no code, is a TruncatedFrame desync rather than an IndexOutOfRangeException. Refs #741 Claude-Session: https://claude.ai/code/session_01Ub9hGmYLQvn9XETzw7CMsT
2026-10-22 15:02:11 -04:00
parent 7bcb6331ed
commit 530f1b2f0b
8 changed files with 655 additions and 70 deletions
@@ -0,0 +1,88 @@
+using System.Diagnostics;
+using Shouldly;
+using Xunit;
+using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
+
+namespace ZB.MOM.WW.OtOpcUa.Driver.Modbus.IntegrationTests;
+
+/// <summary>
+///     Comms-loss → recovery against the real pymodbus <c>standard</c> fixture, with a loopback
+///     <see cref="TcpRelay"/> in front of it so the device can disappear and come back without touching the
+///     shared container. Covers the review 07 High findings P3 (transport self-heal) and H2 (deadband
+///     recovery publish).
+/// </summary>
+[Collection(ModbusSimulatorCollection.Name)]
+[Trait("Category", "Integration")]
+[Trait("Device", "Standard")]
+public sealed class CommsRecoveryTests(ModbusSimulatorFixture sim)
+{
+    private const uint StatusGood = 0u;
+
+    private void SkipUnlessStandardLive()
+    {
+        if (sim.SkipReason is not null) Assert.Skip(sim.SkipReason);
+        var profile = Environment.GetEnvironmentVariable("MODBUS_SIM_PROFILE");
+        if (profile is not null && !string.Equals(profile, "standard", StringComparison.OrdinalIgnoreCase))
+            Assert.Skip($"MODBUS_SIM_PROFILE={profile} — these tests need the standard profile (HR[0..31] = address).");
+    }
+
+    private ModbusDriverOptions Options(TcpRelay relay, params ModbusTagDefinition[] tags) => new()
+    {
+        Host = "127.0.0.1",
+        Port = relay.Port,
+        UnitId = 1,
+        Timeout = TimeSpan.FromSeconds(2),
+        RawTags = ModbusRawTags.Entries(tags),
+        Probe = new ModbusProbeOptions { Enabled = false },
+    };
+
+    private static async Task<T> WaitFor<T>(Func<Task<T>> probe, Func<T, bool> done, TimeSpan budget)
+    {
+        var sw = Stopwatch.StartNew();
+        T last;
+        do
+        {
+            last = await probe();
+            if (done(last)) return last;
+            await Task.Delay(200, TestContext.Current.CancellationToken);
+        } while (sw.Elapsed < budget);
+        return last;
+    }
+
+    /// <summary>
+    ///     P3: after the device drops (two failed calls in a row, socket torn down, reconnect refused) the
+    ///     transport must reconnect ON ITS OWN once the device is back — before this fix every later call
+    ///     failed fast with "Transport not connected" until the 90 s comms-loss watchdog re-initialised the
+    ///     driver.
+    /// </summary>
+    [Fact]
+    public async Task Driver_reads_again_after_the_device_drops_and_returns_without_reinit()
+    {
+        SkipUnlessStandardLive();
+        await using var relay = new TcpRelay(sim.Host, sim.Port);
+        var tag = new ModbusTagDefinition("HR5", ModbusRegion.HoldingRegisters, 5, ModbusDataType.UInt16, Writable: false);
+        await using var driver = new ModbusDriver(Options(relay, tag), "modbus-p3");
+        await driver.InitializeAsync("{}", TestContext.Current.CancellationToken);
+
+        var good = await driver.ReadAsync(["HR5"], TestContext.Current.CancellationToken);
+        good[0].StatusCode.ShouldBe(StatusGood);
+        good[0].Value.ShouldBe((ushort)5);
+
+        relay.Cut();
+        for (var i = 0; i < 2; i++)
+        {
+            var bad = await driver.ReadAsync(["HR5"], TestContext.Current.CancellationToken);
+            bad[0].StatusCode.ShouldNotBe(StatusGood);
+        }
+        driver.GetHealth().State.ShouldBe(DriverState.Degraded);
+
+        relay.Restore();
+        var recovered = await WaitFor(
+            () => driver.ReadAsync(["HR5"], TestContext.Current.CancellationToken),
+            r => r[0].StatusCode == StatusGood,
+            TimeSpan.FromSeconds(20));
+        recovered[0].StatusCode.ShouldBe(StatusGood, "the transport must reconnect by itself once the device is back");
+        recovered[0].Value.ShouldBe((ushort)5);
+        driver.GetHealth().State.ShouldBe(DriverState.Healthy);
+    }
+}
@@ -0,0 +1,101 @@
+using System.Net;
+using System.Net.Sockets;
+
+namespace ZB.MOM.WW.OtOpcUa.Driver.Modbus.IntegrationTests;
+
+/// <summary>
+///     Loopback TCP relay in front of the real Modbus fixture, so a test can make the device "disappear"
+///     (<see cref="Cut"/>: every relayed connection is closed and new connects are refused) and "come back"
+///     (<see cref="Restore"/>: listen again on the SAME port) without touching the shared container.
+/// </summary>
+internal sealed class TcpRelay : IAsyncDisposable
+{
+    private readonly string _upstreamHost;
+    private readonly int _upstreamPort;
+    private readonly object _lock = new();
+    private readonly List<TcpClient> _live = [];
+    private TcpListener? _listener;
+    private CancellationTokenSource _stop = new();
+    private int _accepted;
+
+    public TcpRelay(string upstreamHost, int upstreamPort)
+    {
+        _upstreamHost = upstreamHost;
+        _upstreamPort = upstreamPort;
+        var probe = new TcpListener(IPAddress.Loopback, 0);
+        probe.Start();
+        Port = ((IPEndPoint)probe.LocalEndpoint).Port;
+        probe.Stop();
+        Restore();
+    }
+
+    /// <summary>Gets the loopback port clients dial.</summary>
+    public int Port { get; }
+
+    /// <summary>Gets the number of client connections the relay has accepted.</summary>
+    public int AcceptedConnections => Volatile.Read(ref _accepted);
+
+    /// <summary>Starts (or resumes) relaying on <see cref="Port"/>.</summary>
+    public void Restore()
+    {
+        _stop = new CancellationTokenSource();
+        var listener = new TcpListener(IPAddress.Loopback, Port);
+        listener.Server.SetSocketOption(SocketOptionLevel.Socket, SocketOptionName.ReuseAddress, true);
+        listener.Start();
+        _listener = listener;
+        var token = _stop.Token;
+        _ = Task.Run(() => AcceptLoopAsync(listener, token));
+    }
+
+    /// <summary>Closes every relayed connection and refuses new ones until <see cref="Restore"/>.</summary>
+    public void Cut()
+    {
+        try { _stop.Cancel(); } catch { }
+        try { _listener?.Stop(); } catch { }
+        _listener = null;
+        lock (_lock)
+        {
+            foreach (var c in _live) { try { c.Close(); } catch { } }
+            _live.Clear();
+        }
+    }
+
+    private async Task AcceptLoopAsync(TcpListener listener, CancellationToken token)
+    {
+        while (!token.IsCancellationRequested)
+        {
+            TcpClient downstream;
+            try { downstream = await listener.AcceptTcpClientAsync(token); }
+            catch { return; }
+            Interlocked.Increment(ref _accepted);
+            _ = Task.Run(() => RelayAsync(downstream, token));
+        }
+    }
+
+    private async Task RelayAsync(TcpClient downstream, CancellationToken token)
+    {
+        var upstream = new TcpClient(AddressFamily.InterNetwork);
+        lock (_lock) { _live.Add(downstream); _live.Add(upstream); }
+        try
+        {
+            var addr = (await Dns.GetHostAddressesAsync(_upstreamHost, token))
+                .First(a => a.AddressFamily == AddressFamily.InterNetwork);
+            await upstream.ConnectAsync(addr, _upstreamPort, token);
+            var a = downstream.GetStream();
+            var b = upstream.GetStream();
+            await Task.WhenAny(a.CopyToAsync(b, token), b.CopyToAsync(a, token));
+        }
+        catch { /* cut or upstream gone */ }
+        finally
+        {
+            try { downstream.Close(); } catch { }
+            try { upstream.Close(); } catch { }
+        }
+    }
+
+    public ValueTask DisposeAsync()
+    {
+        Cut();
+        return ValueTask.CompletedTask;
+    }
+}
@@ -0,0 +1,139 @@
+using Shouldly;
+using Xunit;
+
+namespace ZB.MOM.WW.OtOpcUa.Driver.Modbus.Tests;
+
+/// <summary>
+///     Transport self-heal (review 07 P3), single in-gate connect attempt + failed-client disposal (L3)
+///     and the response-PDU length check (H9).
+/// </summary>
+[Trait("Category", "Unit")]
+public sealed class ModbusTransportSelfHealTests
+{
+    private static readonly byte[] ReadHr0 = [0x03, 0x00, 0x00, 0x00, 0x01];
+
+    /// <summary>
+    ///     P3: two lost replies in a row leave the socket torn down. The NEXT call must reconnect on its own
+    ///     instead of failing fast with "Transport not connected" until the 90 s comms-loss watchdog.
+    /// </summary>
+    [Fact]
+    public async Task After_two_consecutive_timeouts_the_next_call_reconnects()
+    {
+        await using var server = new ScriptedModbusTcpServer
+        {
+            Script = i => i < 2 ? ScriptedModbusTcpServer.Action.Stall : ScriptedModbusTcpServer.Action.Answer,
+        };
+        await using var transport = new ModbusTcpTransport("127.0.0.1", server.Port, TimeSpan.FromMilliseconds(500));
+        await transport.ConnectAsync(TestContext.Current.CancellationToken);
+
+        // First call: stall → teardown → reconnect → resend → stall again → propagates.
+        await Should.ThrowAsync<ModbusTransportDesyncException>(
+            () => transport.SendAsync(1, ReadHr0, TestContext.Current.CancellationToken));
+
+        // Second call: the socket is gone; the transport must dial again and succeed.
+        var resp = await transport.SendAsync(1, ReadHr0, TestContext.Current.CancellationToken);
+        resp[0].ShouldBe((byte)0x03);
+        server.AcceptedConnections.ShouldBe(3);
+    }
+
+    /// <summary>
+    ///     P3 throttle: after a FAILED connect, calls inside the backoff window fail fast with
+    ///     <see cref="ModbusTransportUnavailableException"/> and make no connect attempt; once the window has
+    ///     passed the next call dials again. Positive control: the device is back before the throttled call,
+    ///     so an un-throttled transport would have reconnected.
+    /// </summary>
+    [Fact]
+    public async Task Failed_connect_is_throttled_then_retried_after_the_window()
+    {
+        await using var server = new ScriptedModbusTcpServer();
+        await using var transport = new ModbusTcpTransport("127.0.0.1", server.Port, TimeSpan.FromSeconds(2));
+        var now = 1_000_000L;
+        transport.Lifecycle.NowMs = () => now;
+        await transport.ConnectAsync(TestContext.Current.CancellationToken);
+        (await transport.SendAsync(1, ReadHr0, TestContext.Current.CancellationToken))[0].ShouldBe((byte)0x03);
+
+        server.Stop();
+        // Dead socket → EOF → teardown → ONE reconnect attempt → refused.
+        var first = await Should.ThrowAsync<Exception>(
+            () => transport.SendAsync(1, ReadHr0, TestContext.Current.CancellationToken));
+        ModbusSocketLifecycle.IsSocketLevelFailure(first).ShouldBeTrue();
+
+        server.Start();
+        var acceptedBefore = server.AcceptedConnections;
+        await Should.ThrowAsync<ModbusTransportUnavailableException>(
+            () => transport.SendAsync(1, ReadHr0, TestContext.Current.CancellationToken));
+        server.AcceptedConnections.ShouldBe(acceptedBefore, "a throttled call must not dial");
+
+        now += 60_000; // past any window (MaxDelay default 30 s)
+        (await transport.SendAsync(1, ReadHr0, TestContext.Current.CancellationToken))[0].ShouldBe((byte)0x03);
+        server.AcceptedConnections.ShouldBe(acceptedBefore + 1);
+    }
+
+    /// <summary>
+    ///     L3: the reconnect never sleeps inside the transport gate. With a 30 s InitialDelay the old
+    ///     <c>ConnectWithBackoffAsync</c> held the gate for 30 s before even dialling; now one attempt is made
+    ///     and the call returns at once.
+    /// </summary>
+    [Fact]
+    public async Task Reconnect_after_drop_does_not_sleep_inside_the_gate()
+    {
+        await using var server = new ScriptedModbusTcpServer();
+        await using var transport = new ModbusTcpTransport("127.0.0.1", server.Port, TimeSpan.FromSeconds(2),
+            reconnect: new ModbusReconnectOptions { InitialDelay = TimeSpan.FromSeconds(30) });
+        await transport.ConnectAsync(TestContext.Current.CancellationToken);
+        server.Stop();
+
+        var call = transport.SendAsync(1, ReadHr0, TestContext.Current.CancellationToken);
+        var ex = await Should.ThrowAsync<Exception>(() => call.WaitAsync(TimeSpan.FromSeconds(10), TestContext.Current.CancellationToken));
+        ex.ShouldNotBeOfType<TimeoutException>("the send must not wait out the 30 s backoff inside the gate");
+    }
+
+    /// <summary>H9: an MBAP frame carrying no PDU is a truncated-frame desync, not an IndexOutOfRangeException.</summary>
+    [Fact]
+    public async Task Empty_response_pdu_is_a_truncated_frame_desync()
+    {
+        await using var server = new ScriptedModbusTcpServer
+        {
+            RawReply = (_, header, _) => ScriptedModbusTcpServer.Adu(header, []),
+        };
+        await using var transport = new ModbusTcpTransport("127.0.0.1", server.Port, TimeSpan.FromSeconds(2), autoReconnect: false);
+        await transport.ConnectAsync(TestContext.Current.CancellationToken);
+
+        var ex = await Should.ThrowAsync<ModbusTransportDesyncException>(
+            () => transport.SendAsync(1, ReadHr0, TestContext.Current.CancellationToken));
+        ex.Reason.ShouldBe(ModbusTransportDesyncException.DesyncReason.TruncatedFrame);
+    }
+
+    /// <summary>H9: an exception frame missing its exception code is a truncated-frame desync.</summary>
+    [Fact]
+    public async Task Exception_frame_without_code_is_a_truncated_frame_desync()
+    {
+        await using var server = new ScriptedModbusTcpServer
+        {
+            RawReply = (_, header, _) => ScriptedModbusTcpServer.Adu(header, [0x83]),
+        };
+        await using var transport = new ModbusTcpTransport("127.0.0.1", server.Port, TimeSpan.FromSeconds(2), autoReconnect: false);
+        await transport.ConnectAsync(TestContext.Current.CancellationToken);
+
+        var ex = await Should.ThrowAsync<ModbusTransportDesyncException>(
+            () => transport.SendAsync(1, ReadHr0, TestContext.Current.CancellationToken));
+        ex.Reason.ShouldBe(ModbusTransportDesyncException.DesyncReason.TruncatedFrame);
+    }
+
+    /// <summary>Positive control for the two H9 tests: a well-formed exception frame still surfaces as a
+    /// <see cref="ModbusException"/> carrying its code.</summary>
+    [Fact]
+    public async Task Well_formed_exception_frame_still_surfaces_as_ModbusException()
+    {
+        await using var server = new ScriptedModbusTcpServer
+        {
+            RawReply = (_, header, _) => ScriptedModbusTcpServer.Adu(header, [0x83, 0x02]),
+        };
+        await using var transport = new ModbusTcpTransport("127.0.0.1", server.Port, TimeSpan.FromSeconds(2), autoReconnect: false);
+        await transport.ConnectAsync(TestContext.Current.CancellationToken);
+
+        var ex = await Should.ThrowAsync<ModbusException>(
+            () => transport.SendAsync(1, ReadHr0, TestContext.Current.CancellationToken));
+        ex.ExceptionCode.ShouldBe((byte)0x02);
+    }
+}
@@ -0,0 +1,160 @@
+using System.Net;
+using System.Net.Sockets;
+
+namespace ZB.MOM.WW.OtOpcUa.Driver.Modbus.Tests;
+
+/// <summary>
+///     Loopback Modbus-TCP stub whose reply to each request is scripted by the test, and whose listener can
+///     be stopped (connection refused) and restarted on the SAME port. Used by the transport self-heal tests
+///     that need a device which disappears and comes back.
+/// </summary>
+internal sealed class ScriptedModbusTcpServer : IAsyncDisposable
+{
+    /// <summary>What the server does with one request.</summary>
+    public enum Action
+    {
+        /// <summary>Answer with an FC03-shaped response of zeroed registers.</summary>
+        Answer,
+        /// <summary>Read the request and never answer it (the client's per-op timeout fires).</summary>
+        Stall,
+        /// <summary>Close the socket without answering.</summary>
+        Drop,
+    }
+
+    private TcpListener? _listener;
+    private CancellationTokenSource _stop = new();
+    private int _requests;
+    private int _accepted;
+
+    /// <summary>Optional scripted reply. Receives the 0-based request index and the request PDU; returns
+    /// the raw bytes to write AFTER the MBAP TxId (i.e. the full ADU is built by the test), or null to fall
+    /// through to <see cref="Script"/>.</summary>
+    public Func<int, byte[], byte[], byte[]?>? RawReply { get; set; }
+
+    /// <summary>Per-request action; defaults to <see cref="Action.Answer"/>.</summary>
+    public Func<int, Action> Script { get; set; } = _ => Action.Answer;
+
+    /// <summary>Gets the port the server listens on (stable across <see cref="Stop"/>/<see cref="Start"/>).</summary>
+    public int Port { get; }
+
+    /// <summary>Gets the number of accepted TCP connections.</summary>
+    public int AcceptedConnections => Volatile.Read(ref _accepted);
+
+    /// <summary>Gets the number of requests received.</summary>
+    public int Requests => Volatile.Read(ref _requests);
+
+    public ScriptedModbusTcpServer()
+    {
+        var probe = new TcpListener(IPAddress.Loopback, 0);
+        probe.Start();
+        Port = ((IPEndPoint)probe.LocalEndpoint).Port;
+        probe.Stop();
+        Start();
+    }
+
+    /// <summary>(Re)starts listening on <see cref="Port"/>.</summary>
+    public void Start()
+    {
+        _stop = new CancellationTokenSource();
+        _listener = new TcpListener(IPAddress.Loopback, Port);
+        _listener.Server.SetSocketOption(SocketOptionLevel.Socket, SocketOptionName.ReuseAddress, true);
+        _listener.Start();
+        var listener = _listener;
+        var token = _stop.Token;
+        _ = Task.Run(() => AcceptLoopAsync(listener, token));
+    }
+
+    /// <summary>Stops listening and closes every live connection — new connects are refused.</summary>
+    public void Stop()
+    {
+        try { _stop.Cancel(); } catch { }
+        try { _listener?.Stop(); } catch { }
+        _listener = null;
+    }
+
+    private async Task AcceptLoopAsync(TcpListener listener, CancellationToken token)
+    {
+        while (!token.IsCancellationRequested)
+        {
+            TcpClient client;
+            try { client = await listener.AcceptTcpClientAsync(token); }
+            catch { return; }
+            Interlocked.Increment(ref _accepted);
+            _ = Task.Run(() => ServeAsync(client, token));
+        }
+    }
+
+    private async Task ServeAsync(TcpClient client, CancellationToken token)
+    {
+        using var _ = client;
+        using var reg = token.Register(() => { try { client.Close(); } catch { } });
+        try
+        {
+            var stream = client.GetStream();
+            while (!token.IsCancellationRequested)
+            {
+                var header = new byte[7];
+                if (!await ReadExactly(stream, header, token)) return;
+                var len = (ushort)((header[4] << 8) | header[5]);
+                var pdu = new byte[len - 1];
+                if (!await ReadExactly(stream, pdu, token)) return;
+                var idx = Interlocked.Increment(ref _requests) - 1;
+
+                var raw = RawReply?.Invoke(idx, header, pdu);
+                if (raw is not null)
+                {
+                    await stream.WriteAsync(raw, token);
+                    await stream.FlushAsync(token);
+                    continue;
+                }
+
+                switch (Script(idx))
+                {
+                    case Action.Stall:
+                        continue;
+                    case Action.Drop:
+                        client.Client.Shutdown(SocketShutdown.Both);
+                        return;
+                }
+
+                var qty = (ushort)((pdu[3] << 8) | pdu[4]);
+                var respPdu = new byte[2 + qty * 2];
+                respPdu[0] = pdu[0];
+                respPdu[1] = (byte)(qty * 2);
+                await stream.WriteAsync(Adu(header, respPdu), token);
+                await stream.FlushAsync(token);
+            }
+        }
+        catch { /* connection torn down */ }
+    }
+
+    /// <summary>Builds a reply ADU echoing the request's TxId/unit around <paramref name="respPdu"/>.</summary>
+    public static byte[] Adu(byte[] requestHeader, byte[] respPdu)
+    {
+        var respLen = (ushort)(1 + respPdu.Length);
+        var adu = new byte[7 + respPdu.Length];
+        adu[0] = requestHeader[0]; adu[1] = requestHeader[1];
+        adu[4] = (byte)(respLen >> 8); adu[5] = (byte)(respLen & 0xFF);
+        adu[6] = requestHeader[6];
+        Buffer.BlockCopy(respPdu, 0, adu, 7, respPdu.Length);
+        return adu;
+    }
+
+    private static async Task<bool> ReadExactly(Stream s, byte[] buf, CancellationToken token)
+    {
+        var read = 0;
+        while (read < buf.Length)
+        {
+            var n = await s.ReadAsync(buf.AsMemory(read), token);
+            if (n == 0) return false;
+            read += n;
+        }
+        return true;
+    }
+
+    public ValueTask DisposeAsync()
+    {
+        Stop();
+        return ValueTask.CompletedTask;
+    }
+}