From 530f1b2f0b3320ea2b66d4de1f82dfd084f0d8a4 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Thu, 22 Oct 2026 15:02:11 -0400 Subject: [PATCH] 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 --- .../ModbusRtuOverTcpTransport.cs | 27 +-- .../ModbusSocketLifecycle.cs | 145 ++++++++++------ .../ModbusTcpTransport.cs | 42 +++-- .../ModbusTransportUnavailableException.cs | 23 +++ .../CommsRecoveryTests.cs | 88 ++++++++++ .../TcpRelay.cs | 101 +++++++++++ .../ModbusTransportSelfHealTests.cs | 139 +++++++++++++++ .../ScriptedModbusTcpServer.cs | 160 ++++++++++++++++++ 8 files changed, 655 insertions(+), 70 deletions(-) create mode 100644 src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus/ModbusTransportUnavailableException.cs create mode 100644 tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus.IntegrationTests/CommsRecoveryTests.cs create mode 100644 tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus.IntegrationTests/TcpRelay.cs create mode 100644 tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus.Tests/ModbusTransportSelfHealTests.cs create mode 100644 tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus.Tests/ScriptedModbusTcpServer.cs diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus/ModbusRtuOverTcpTransport.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus/ModbusRtuOverTcpTransport.cs index d159e24a..69ccf915 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus/ModbusRtuOverTcpTransport.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus/ModbusRtuOverTcpTransport.cs @@ -91,7 +91,7 @@ public sealed class ModbusRtuOverTcpTransport : IModbusTransport public async Task SendAsync(byte unitId, byte[] pdu, CancellationToken ct) { if (_disposed) throw new ObjectDisposedException(nameof(ModbusRtuOverTcpTransport)); - if (CurrentStream is null) throw new InvalidOperationException("Transport not connected"); + if (_life is null && _testStream is null) throw new InvalidOperationException("Transport not connected"); // Single-flight: RTU carries no transaction id, so at most one transaction may be on the bus // at a time — serialise every send/receive under the gate. @@ -101,11 +101,14 @@ public sealed class ModbusRtuOverTcpTransport : IModbusTransport // Proactive idle-disconnect (real socket only): if the socket has been quiet longer than // the configured threshold, tear it down + reconnect before this PDU lands. Defends // against silent NAT / firewall reaping. - if (_life is not null && _life.ShouldReconnectForIdle()) - { + if (_life is not null && _life.Stream is not null && _life.ShouldReconnectForIdle()) await _life.TearDownAsync().ConfigureAwait(false); - await _life.ConnectWithBackoffAsync(ct).ConfigureAwait(false); - } + + // Self-heal (real socket only): a socket torn down by an earlier failure is re-dialled here, + // once, throttled by the backoff window — same contract as ModbusTcpTransport. + var freshlyConnected = _life is not null && _life.Stream is null; + if (freshlyConnected) + await _life!.EnsureConnectedAsync(ct).ConfigureAwait(false); try { @@ -113,14 +116,16 @@ public sealed class ModbusRtuOverTcpTransport : IModbusTransport _life?.MarkSuccess(); return result; } - catch (Exception ex) when (_life is not null && _life.AutoReconnect && ModbusSocketLifecycle.IsSocketLevelFailure(ex)) + catch (Exception ex) when (!freshlyConnected && _life is not null && _life.AutoReconnect + && !ct.IsCancellationRequested + && ModbusSocketLifecycle.IsSocketLevelFailure(ex)) { - // Mid-transaction drop: tear down the dead socket, reconnect (with backoff if - // configured), resend. Single retry — a second failure propagates so health/status - // reflect reality. Never reached in the ForTest seam (a raw injected stream has no - // lifecycle to reconnect). + // Mid-transaction drop: tear down the dead socket, make ONE throttled reconnect attempt + // (never a sleep inside the gate), resend. Single retry — a second failure propagates so + // health/status reflect reality. Never reached in the ForTest seam (a raw injected stream + // has no lifecycle to reconnect). await _life.TearDownAsync().ConfigureAwait(false); - await _life.ConnectWithBackoffAsync(ct).ConfigureAwait(false); + await _life.EnsureConnectedAsync(ct).ConfigureAwait(false); var result = await SendOnceAsync(unitId, pdu, ct).ConfigureAwait(false); _life.MarkSuccess(); return result; diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus/ModbusSocketLifecycle.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus/ModbusSocketLifecycle.cs index 024d28f9..7a032cdd 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus/ModbusSocketLifecycle.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus/ModbusSocketLifecycle.cs @@ -4,8 +4,8 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Modbus; /// /// Owns the raw TCP socket lifecycle shared by every Modbus-over-TCP transport (MBAP and -/// RTU-over-TCP alike): the IPv4-preference connect, OS-level keep-alive, geometric-backoff -/// reconnect, idle-disconnect tracking, and teardown. It exposes the current +/// RTU-over-TCP alike): the IPv4-preference connect, OS-level keep-alive, throttled +/// reconnect (one attempt per geometric-backoff window), idle-disconnect tracking, and teardown. It exposes the current /// so the composing transport can drive its own framing on top — /// the lifecycle knows nothing about MBAP / RTU wire layout. /// @@ -14,7 +14,7 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Modbus; /// Extracted verbatim from so the connect / reconnect / /// keep-alive / idle behaviour is written once and composed by both transports. The /// constructor is connection-free — all socket I/O happens in / -/// . +/// (one throttled attempt per call — never a sleep loop). /// /// /// Why keep-alive matters for DL205/DL260: the AutomationDirect H2-ECOM100 does NOT send @@ -37,6 +37,9 @@ public sealed class ModbusSocketLifecycle : IAsyncDisposable private NetworkStream? _stream; private DateTime _lastSuccessUtc = DateTime.UtcNow; + /// Monotonic millisecond clock for the reconnect throttle (test seam). + internal Func NowMs { get; set; } = () => Environment.TickCount64; + /// Initializes a new instance of the class. /// The host address or hostname of the Modbus server. /// The TCP port of the Modbus server. @@ -71,63 +74,107 @@ public sealed class ModbusSocketLifecycle : IAsyncDisposable /// A task representing the asynchronous connection operation. public async Task ConnectAsync(CancellationToken ct) { - // Resolve the host explicitly + prefer IPv4. .NET's TcpClient default-constructor is - // dual-stack (IPv6 first, fallback to IPv4) — but most Modbus TCP devices (PLCs and - // simulators like pymodbus) bind 0.0.0.0 only, so the IPv6 attempt times out and we - // burn the entire ConnectAsync budget before even trying IPv4. Resolving first + - // dialing the IPv4 address directly sidesteps that. - var addresses = await System.Net.Dns.GetHostAddressesAsync(_host, ct).ConfigureAwait(false); - var ipv4 = System.Linq.Enumerable.FirstOrDefault(addresses, - a => a.AddressFamily == System.Net.Sockets.AddressFamily.InterNetwork); - var target = ipv4 ?? (addresses.Length > 0 ? addresses[0] : System.Net.IPAddress.Loopback); - - _client = new TcpClient(target.AddressFamily); - EnableKeepAlive(_client, _keepAlive); + // A previous socket (failed, torn down or idle-reaped) is released FIRST: overwriting _client + // used to strand the old TcpClient for the finalizer — up to nine per backoff loop. + await TearDownAsync().ConfigureAwait(false); + // DNS resolution runs under the same per-op deadline as the TCP connect: an unresolvable or + // slow resolver must not hold the transport gate for the OS resolver's own (much longer) timeout. using var cts = CancellationTokenSource.CreateLinkedTokenSource(ct); cts.CancelAfter(_timeout); - await _client.ConnectAsync(target, _port, cts.Token).ConfigureAwait(false); - _stream = _client.GetStream(); - _lastSuccessUtc = DateTime.UtcNow; + TcpClient? client = null; + try + { + // Resolve the host explicitly + prefer IPv4. .NET's TcpClient default-constructor is + // dual-stack (IPv6 first, fallback to IPv4) — but most Modbus TCP devices (PLCs and + // simulators like pymodbus) bind 0.0.0.0 only, so the IPv6 attempt times out and we + // burn the entire ConnectAsync budget before even trying IPv4. Resolving first + + // dialing the IPv4 address directly sidesteps that. + var addresses = await System.Net.Dns.GetHostAddressesAsync(_host, cts.Token).ConfigureAwait(false); + var ipv4 = System.Linq.Enumerable.FirstOrDefault(addresses, + a => a.AddressFamily == System.Net.Sockets.AddressFamily.InterNetwork); + var target = ipv4 ?? (addresses.Length > 0 ? addresses[0] : System.Net.IPAddress.Loopback); + + client = new TcpClient(target.AddressFamily); + EnableKeepAlive(client, _keepAlive); + await client.ConnectAsync(target, _port, cts.Token).ConfigureAwait(false); + _stream = client.GetStream(); + _client = client; + client = null; // ownership transferred + _lastSuccessUtc = DateTime.UtcNow; + } + catch (OperationCanceledException) when (!ct.IsCancellationRequested) + { + // The per-op deadline, not the caller: normalise to a socket-level failure so every + // classifier treats a connect timeout as a comms fault. + throw new ModbusTransportUnavailableException( + $"Modbus connect to {_host}:{_port} timed out after {_timeout.TotalMilliseconds:0}ms"); + } + finally + { + // A failed attempt disposes its own client — never leave it for the finalizer. + client?.Dispose(); + } } + // Reconnect throttle state. Guarded by the composing transport's single-flight gate (every caller of + // EnsureConnectedAsync holds it), so no extra lock is needed. + private long _connectNotBeforeMs; + private TimeSpan _failureDelay = TimeSpan.Zero; + /// - /// Connect attempt with the configured geometric backoff. The first attempt fires after - /// (default zero — immediate); each - /// subsequent attempt sleeps for the previous delay times BackoffMultiplier, - /// capped at MaxDelay. Caller's cancellation token aborts the loop. + /// Ensures a live socket for the next transaction with AT MOST ONE connect attempt, throttled by the + /// configured geometric backoff: after a failed attempt, calls inside the backoff window fail fast + /// with and never dial. The window starts at + /// (floored at 1 s) and grows by + /// BackoffMultiplier per consecutive failure up to MaxDelay; a successful connect + /// resets it. Never sleeps — the composing transport calls it inside its gate, and a sleep there + /// used to queue every read, write and probe behind one dead socket for up to ~81 s. /// /// A cancellation token to observe for cancellation. - /// A task representing the asynchronous reconnect operation. - public async Task ConnectWithBackoffAsync(CancellationToken ct) + /// A task that completes when a socket is live. + /// Auto-reconnect is off, or the call is inside the backoff window. + public async Task EnsureConnectedAsync(CancellationToken ct) { - var delay = _reconnect.InitialDelay; - var attempt = 0; - while (true) + if (_stream is not null) return; + if (!_autoReconnect) + throw new ModbusTransportUnavailableException( + $"Modbus transport to {_host}:{_port} is not connected and auto-reconnect is disabled"); + + var now = NowMs(); + if (now < _connectNotBeforeMs) + throw new ModbusTransportUnavailableException( + $"Modbus reconnect to {_host}:{_port} throttled for another {_connectNotBeforeMs - now}ms after a failed connect"); + + try { - if (delay > TimeSpan.Zero) - await Task.Delay(delay, ct).ConfigureAwait(false); - try - { - await ConnectAsync(ct).ConfigureAwait(false); - return; - } - catch (Exception ex) when (IsSocketLevelFailure(ex) && _autoReconnect) - { - attempt++; - // Geometric growth, capped. Use Math.Min on ticks so we don't overflow with - // pathological multipliers / long deployments. - var nextTicks = (long)(Math.Max(delay.Ticks, TimeSpan.FromMilliseconds(100).Ticks) * _reconnect.BackoffMultiplier); - delay = TimeSpan.FromTicks(Math.Min(nextTicks, _reconnect.MaxDelay.Ticks)); - if (attempt >= 10) - { - // Bail after 10 attempts to surface persistent failure to the caller. With - // the default backoff (1s base, 2.0x mult, 30s cap) this is roughly 4 minutes - // of attempts; with InitialDelay=0 it's immediate up to the same cap. - throw; - } - } + await ConnectAsync(ct).ConfigureAwait(false); + _failureDelay = TimeSpan.Zero; + _connectNotBeforeMs = 0; } + catch (Exception) when (!ct.IsCancellationRequested) + { + _failureDelay = NextDelay(_failureDelay); + _connectNotBeforeMs = NowMs() + (long)_failureDelay.TotalMilliseconds; + throw; + } + } + + /// Minimum reconnect window after a failed connect — a dead device is dialled at most once per second. + private static readonly TimeSpan MinReconnectWindow = TimeSpan.FromSeconds(1); + + private TimeSpan NextDelay(TimeSpan previous) + { + if (previous <= TimeSpan.Zero) + { + var first = _reconnect.InitialDelay > MinReconnectWindow ? _reconnect.InitialDelay : MinReconnectWindow; + return first < _reconnect.MaxDelay ? first : _reconnect.MaxDelay > TimeSpan.Zero ? _reconnect.MaxDelay : first; + } + // Geometric growth, capped. Math.Min on ticks so a pathological multiplier cannot overflow. + var multiplier = _reconnect.BackoffMultiplier >= 1.0 ? _reconnect.BackoffMultiplier : 1.0; + var nextTicks = previous.Ticks * multiplier; + var cap = _reconnect.MaxDelay > TimeSpan.Zero ? _reconnect.MaxDelay.Ticks : long.MaxValue; + return TimeSpan.FromTicks(nextTicks >= cap ? cap : (long)nextTicks); } /// diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus/ModbusTcpTransport.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus/ModbusTcpTransport.cs index f6f94539..26534610 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus/ModbusTcpTransport.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus/ModbusTcpTransport.cs @@ -56,6 +56,9 @@ public sealed class ModbusTcpTransport : IModbusTransport _life = new ModbusSocketLifecycle(host, port, timeout, autoReconnect, keepAlive, idleDisconnect, reconnect); } + /// The socket lifecycle this transport composes (test seam for its clock). + internal ModbusSocketLifecycle Lifecycle => _life; + /// public Task ConnectAsync(CancellationToken ct) => _life.ConnectAsync(ct); @@ -63,7 +66,6 @@ public sealed class ModbusTcpTransport : IModbusTransport public async Task SendAsync(byte unitId, byte[] pdu, CancellationToken ct) { if (_disposed) throw new ObjectDisposedException(nameof(ModbusTcpTransport)); - if (_life.Stream is null) throw new InvalidOperationException("Transport not connected"); await _gate.WaitAsync(ct).ConfigureAwait(false); try @@ -72,11 +74,16 @@ public sealed class ModbusTcpTransport : IModbusTransport // threshold, tear it down + reconnect before this PDU lands. Defends against silent // NAT / firewall reaping where the socket looks alive locally but the upstream side // dropped it minutes ago. - if (_life.ShouldReconnectForIdle()) - { + if (_life.Stream is not null && _life.ShouldReconnectForIdle()) await _life.TearDownAsync().ConfigureAwait(false); - await _life.ConnectWithBackoffAsync(ct).ConfigureAwait(false); - } + + // Self-heal: a socket torn down by an earlier failure (two lost replies in a row, a refused + // reconnect) is re-dialled HERE, once, throttled by the backoff window. This used to throw + // "Transport not connected" before the gate, so the transport never recovered on its own and + // every call failed fast until the 90 s comms-loss watchdog re-initialised the whole driver. + var freshlyConnected = _life.Stream is null; + if (freshlyConnected) + await _life.EnsureConnectedAsync(ct).ConfigureAwait(false); try { @@ -84,13 +91,16 @@ public sealed class ModbusTcpTransport : IModbusTransport _life.MarkSuccess(); return result; } - catch (Exception ex) when (_life.AutoReconnect && ModbusSocketLifecycle.IsSocketLevelFailure(ex)) + catch (Exception ex) when (!freshlyConnected && _life.AutoReconnect + && !ct.IsCancellationRequested + && ModbusSocketLifecycle.IsSocketLevelFailure(ex)) { - // Mid-transaction drop: tear down the dead socket, reconnect (with backoff if - // configured), resend. Single retry — if it fails again, let it propagate so - // health/status reflect reality. + // Mid-transaction drop on a socket that had been live: tear down the dead socket, make ONE + // throttled reconnect attempt (never a sleep inside the gate), resend. Single retry — if it + // fails again, let it propagate so health/status reflect reality. A socket this call just + // dialled is not retried: it proved nothing about a stale connection. await _life.TearDownAsync().ConfigureAwait(false); - await _life.ConnectWithBackoffAsync(ct).ConfigureAwait(false); + await _life.EnsureConnectedAsync(ct).ConfigureAwait(false); var result = await SendOnceAsync(unitId, pdu, ct).ConfigureAwait(false); _life.MarkSuccess(); return result; @@ -162,10 +172,22 @@ public sealed class ModbusTcpTransport : IModbusTransport var respPdu = new byte[respLen - 1]; await ReadExactlyAsync(stream, respPdu, cts.Token).ConfigureAwait(false); + // Length check before indexing: a PDU must carry at least its function code, and an exception + // PDU its exception code too. Indexing past the end used to throw IndexOutOfRangeException — + // not an IOException, so no teardown/retry, and a malformed reply read as a driver bug. + if (respPdu.Length < 1) + throw new ModbusTransportDesyncException( + ModbusTransportDesyncException.DesyncReason.TruncatedFrame, + "Modbus response carried no PDU (MBAP length 1)"); + // Exception PDU: function code has high bit set. This is a well-formed protocol-level // error — the socket is still coherent, so it MUST propagate without teardown. if ((respPdu[0] & 0x80) != 0) { + if (respPdu.Length < 2) + throw new ModbusTransportDesyncException( + ModbusTransportDesyncException.DesyncReason.TruncatedFrame, + $"Modbus exception response fc=0x{respPdu[0]:X2} carried no exception code"); var fc = (byte)(respPdu[0] & 0x7F); var ex = respPdu[1]; throw new ModbusException(fc, ex, $"Modbus exception fc={fc} code={ex}"); diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus/ModbusTransportUnavailableException.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus/ModbusTransportUnavailableException.cs new file mode 100644 index 00000000..0df33d1b --- /dev/null +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus/ModbusTransportUnavailableException.cs @@ -0,0 +1,23 @@ +namespace ZB.MOM.WW.OtOpcUa.Driver.Modbus; + +/// +/// Raised when a Modbus transport has no live socket and cannot get one for this call: the one +/// in-gate connect attempt failed, the reconnect is inside its backoff window, or auto-reconnect is +/// disabled. It says the ENDPOINT is unreachable, so the driver fails the rest of the batch fast +/// instead of paying the same failure once per tag. +/// +/// +/// Derives from so every existing socket-level classifier +/// (ModbusSocketLifecycle.IsSocketLevelFailure, the driver's comms-fault mapping) treats it as +/// a communication failure without extra plumbing. +/// +public sealed class ModbusTransportUnavailableException : IOException +{ + /// Initializes a new instance of the class. + /// A human-readable description. + /// The originating exception, if any. + public ModbusTransportUnavailableException(string message, Exception? inner = null) + : base(message, inner) + { + } +} diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus.IntegrationTests/CommsRecoveryTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus.IntegrationTests/CommsRecoveryTests.cs new file mode 100644 index 00000000..37192239 --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus.IntegrationTests/CommsRecoveryTests.cs @@ -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; + +/// +/// Comms-loss → recovery against the real pymodbus standard fixture, with a loopback +/// 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). +/// +[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 WaitFor(Func> probe, Func 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; + } + + /// + /// 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. + /// + [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); + } +} diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus.IntegrationTests/TcpRelay.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus.IntegrationTests/TcpRelay.cs new file mode 100644 index 00000000..416d0e72 --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus.IntegrationTests/TcpRelay.cs @@ -0,0 +1,101 @@ +using System.Net; +using System.Net.Sockets; + +namespace ZB.MOM.WW.OtOpcUa.Driver.Modbus.IntegrationTests; + +/// +/// Loopback TCP relay in front of the real Modbus fixture, so a test can make the device "disappear" +/// (: every relayed connection is closed and new connects are refused) and "come back" +/// (: listen again on the SAME port) without touching the shared container. +/// +internal sealed class TcpRelay : IAsyncDisposable +{ + private readonly string _upstreamHost; + private readonly int _upstreamPort; + private readonly object _lock = new(); + private readonly List _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(); + } + + /// Gets the loopback port clients dial. + public int Port { get; } + + /// Gets the number of client connections the relay has accepted. + public int AcceptedConnections => Volatile.Read(ref _accepted); + + /// Starts (or resumes) relaying on . + 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)); + } + + /// Closes every relayed connection and refuses new ones until . + 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; + } +} diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus.Tests/ModbusTransportSelfHealTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus.Tests/ModbusTransportSelfHealTests.cs new file mode 100644 index 00000000..22e3a93b --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus.Tests/ModbusTransportSelfHealTests.cs @@ -0,0 +1,139 @@ +using Shouldly; +using Xunit; + +namespace ZB.MOM.WW.OtOpcUa.Driver.Modbus.Tests; + +/// +/// Transport self-heal (review 07 P3), single in-gate connect attempt + failed-client disposal (L3) +/// and the response-PDU length check (H9). +/// +[Trait("Category", "Unit")] +public sealed class ModbusTransportSelfHealTests +{ + private static readonly byte[] ReadHr0 = [0x03, 0x00, 0x00, 0x00, 0x01]; + + /// + /// 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. + /// + [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( + () => 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); + } + + /// + /// P3 throttle: after a FAILED connect, calls inside the backoff window fail fast with + /// 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. + /// + [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( + () => transport.SendAsync(1, ReadHr0, TestContext.Current.CancellationToken)); + ModbusSocketLifecycle.IsSocketLevelFailure(first).ShouldBeTrue(); + + server.Start(); + var acceptedBefore = server.AcceptedConnections; + await Should.ThrowAsync( + () => 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); + } + + /// + /// L3: the reconnect never sleeps inside the transport gate. With a 30 s InitialDelay the old + /// ConnectWithBackoffAsync held the gate for 30 s before even dialling; now one attempt is made + /// and the call returns at once. + /// + [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(() => call.WaitAsync(TimeSpan.FromSeconds(10), TestContext.Current.CancellationToken)); + ex.ShouldNotBeOfType("the send must not wait out the 30 s backoff inside the gate"); + } + + /// H9: an MBAP frame carrying no PDU is a truncated-frame desync, not an IndexOutOfRangeException. + [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( + () => transport.SendAsync(1, ReadHr0, TestContext.Current.CancellationToken)); + ex.Reason.ShouldBe(ModbusTransportDesyncException.DesyncReason.TruncatedFrame); + } + + /// H9: an exception frame missing its exception code is a truncated-frame desync. + [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( + () => transport.SendAsync(1, ReadHr0, TestContext.Current.CancellationToken)); + ex.Reason.ShouldBe(ModbusTransportDesyncException.DesyncReason.TruncatedFrame); + } + + /// Positive control for the two H9 tests: a well-formed exception frame still surfaces as a + /// carrying its code. + [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( + () => transport.SendAsync(1, ReadHr0, TestContext.Current.CancellationToken)); + ex.ExceptionCode.ShouldBe((byte)0x02); + } +} diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus.Tests/ScriptedModbusTcpServer.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus.Tests/ScriptedModbusTcpServer.cs new file mode 100644 index 00000000..1fa6a505 --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus.Tests/ScriptedModbusTcpServer.cs @@ -0,0 +1,160 @@ +using System.Net; +using System.Net.Sockets; + +namespace ZB.MOM.WW.OtOpcUa.Driver.Modbus.Tests; + +/// +/// 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. +/// +internal sealed class ScriptedModbusTcpServer : IAsyncDisposable +{ + /// What the server does with one request. + public enum Action + { + /// Answer with an FC03-shaped response of zeroed registers. + Answer, + /// Read the request and never answer it (the client's per-op timeout fires). + Stall, + /// Close the socket without answering. + Drop, + } + + private TcpListener? _listener; + private CancellationTokenSource _stop = new(); + private int _requests; + private int _accepted; + + /// 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 . + public Func? RawReply { get; set; } + + /// Per-request action; defaults to . + public Func Script { get; set; } = _ => Action.Answer; + + /// Gets the port the server listens on (stable across /). + public int Port { get; } + + /// Gets the number of accepted TCP connections. + public int AcceptedConnections => Volatile.Read(ref _accepted); + + /// Gets the number of requests received. + 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(); + } + + /// (Re)starts listening on . + 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)); + } + + /// Stops listening and closes every live connection — new connects are refused. + 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 */ } + } + + /// Builds a reply ADU echoing the request's TxId/unit around . + 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 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; + } +}