diff --git a/docs/drivers/S7.md b/docs/drivers/S7.md index 448c7689..9491c93c 100644 --- a/docs/drivers/S7.md +++ b/docs/drivers/S7.md @@ -336,6 +336,14 @@ degrades health nor reopens the connection, and because it proves the CPU is rea every tag was refused (a wrong DB number) sat `Degraded` with a frozen timestamp and the comms-loss watchdog reconnected it every 90 s against a perfectly reachable PLC. +**A dead handle ends the pass's wire traffic (R07:P1).** Once a read pass or write batch marks the +connection dead (a wire op blew its deadline on a frozen-but-established peer, a socket dropped), the +remaining spans / multi-var chunks / tags / writes of that pass are stamped `BadCommunicationError` +without I/O; the next call reopens. A multi-var chunk that dies on a transport loss no longer falls back +to one doomed per-tag read per member. Without this an op that ignores its cancellation token would cost +one full deadline per remaining item under the gate, and `Degraded` (and so the comms-loss watchdog's +clock) only became visible after the whole pass. + **Connect backoff.** A dead PLC costs one connect attempt per backoff window (1 s doubling to a 30 s cap), not one per read / poll tick / probe tick: inside the window `EnsureConnectedAsync` fails fast (`BadCommunicationError`, `Degraded`) without a connect. A successful connect clears the window, so a diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.S7/S7Driver.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.S7/S7Driver.cs index f1136908..e413125c 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.S7/S7Driver.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.S7/S7Driver.cs @@ -762,6 +762,13 @@ public sealed class S7Driver perTag.Add(span.Members[0].OriginalIndex); continue; } + // Fail fast once an earlier span marked the handle dead (R07:P1): every + // further wire op on it would pay the full deadline under the gate. + if (_plcDead) + { + foreach (var member in span.Members) results[member.OriginalIndex] = DeadHandleSnapshot(now); + continue; + } await ReadSpanAsync(plc, span, results, now, outcome, cancellationToken).ConfigureAwait(false); } perTag.AddRange(plan.LeftoverIndexes); @@ -790,6 +797,12 @@ public sealed class S7Driver foreach (var i in perTag) { + // Fail fast once the handle is dead (R07:P1) — see the span loop. + if (_plcDead) + { + results[i] = DeadHandleSnapshot(now); + continue; + } try { var value = await ReadOneAsync(plc, resolved[i]!, cancellationToken).ConfigureAwait(false); @@ -1069,6 +1082,13 @@ public sealed class S7Driver { var chunk = batchable.GetRange(start, Math.Min(MultiVarReadMaxItems, batchable.Count - start)); + // Fail fast once the handle is dead (R07:P1) — see ReadAsync's span loop. + if (_plcDead) + { + foreach (var (index, _, _) in chunk) results[index] = DeadHandleSnapshot(now); + continue; + } + IReadOnlyList answered; try { @@ -1094,6 +1114,17 @@ public sealed class S7Driver _plcDead = true; throw; } + catch (Exception ex) when (IsS7TransportLoss(ex)) + { + // A transport loss is the connection's fault, not the batch's: falling back would + // send one more doomed per-tag read per member, each paying the full deadline + // (R07:P1). Mark the handle dead and fail the chunk (and, via the guards, the rest + // of the pass) without I/O. + _plcDead = true; + outcome.RecordFailure(DriverState.Degraded, ex.Message); + foreach (var (index, _, _) in chunk) results[index] = DeadHandleSnapshot(now); + continue; + } catch (Exception ex) { _logger.LogDebug(ex, @@ -1974,6 +2005,13 @@ public sealed class S7Driver results[i] = new WriteResult(StatusBadNotWritable); continue; } + // Fail the rest of the batch fast once a write marked the handle dead (R07:P1): the + // next call reopens; this batch must not serialise one deadline per remaining write. + if (_plcDead) + { + results[i] = new WriteResult(StatusBadCommunicationError); + continue; + } try { await WriteOneAsync(plc, tag, w.Value, cancellationToken).ConfigureAwait(false); @@ -2415,6 +2453,37 @@ public sealed class S7Driver if (IsS7ConnectionFatal(ex)) _plcDead = true; } + /// + /// The result stamped, without any I/O, on an item a pass skipped because an earlier item had + /// already marked the connection dead (R07:P1). The pass's health already carries the failure + /// that marked it. + /// + /// The pass timestamp. + /// A BadCommunicationError snapshot. + private static DataValueSnapshot DeadHandleSnapshot(DateTime now) => new(null, StatusBadCommunicationError, null, now); + + /// + /// The TRANSPORT subset of : a socket / IO / disposed handle / + /// blown deadline / anywhere in the chain. Excludes the + /// framing class, because S7.Net reports a multi-var request with one refused item as + /// PlcException(ReadData, PlcException(WrongNumberReceivedBytes)) (live-probed on the snap7 + /// fixture) over a perfectly healthy stream — the multi-var path falls back per tag for that + /// shape, and a genuine desync still surfaces on the per-tag reads. + /// + /// The exception to classify. + /// for a transport loss. + private static bool IsS7TransportLoss(Exception ex) + { + for (Exception? e = ex; e is not null; e = e.InnerException) + { + if (e is System.Net.Sockets.SocketException or System.IO.IOException or ObjectDisposedException or TimeoutException) + return true; + if (e is PlcException { ErrorCode: ErrorCode.ConnectionError }) + return true; + } + return false; + } + /// /// True when (or any inner exception) is a socket-level / connection /// loss OR an ISO-on-TCP framing/desync fault that a reopen can repair — a diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.S7.IntegrationTests/FreezableTcpProxy.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.S7.IntegrationTests/FreezableTcpProxy.cs new file mode 100644 index 00000000..6c5e48ce --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.S7.IntegrationTests/FreezableTcpProxy.cs @@ -0,0 +1,104 @@ +using System.Net; +using System.Net.Sockets; + +namespace ZB.MOM.WW.OtOpcUa.Driver.S7.IntegrationTests; + +/// +/// An in-test TCP forwarder in front of the S7 fixture that can be FROZEN: while frozen it keeps +/// every socket open but forwards nothing in either direction — the frozen-but-established peer +/// (cable pull, firewall DROP) that only a wire-op deadline can detect, produced without touching +/// the shared fixture. It counts the ISO-on-TCP (TPKT) request frames the client sends while frozen, +/// so a test can assert how many wire operations went out without timing anything. +/// +internal sealed class FreezableTcpProxy : IAsyncDisposable +{ + private readonly string _targetHost; + private readonly int _targetPort; + private readonly TcpListener _listener = new(IPAddress.Loopback, 0); + private readonly CancellationTokenSource _cts = new(); + private readonly List _sockets = []; + private volatile bool _frozen; + private int _framesWhileFrozen; + + /// Starts a proxy forwarding to :. + /// The fixture host. + /// The fixture port. + public FreezableTcpProxy(string targetHost, int targetPort) + { + _targetHost = targetHost; + _targetPort = targetPort; + _listener.Start(); + _ = AcceptLoopAsync(); + } + + /// The local port clients connect to. + public int Port => ((IPEndPoint)_listener.LocalEndpoint).Port; + + /// TPKT request frames the client sent while the proxy was frozen. + public int FramesWhileFrozen => Volatile.Read(ref _framesWhileFrozen); + + /// Stops forwarding (sockets stay open). + public void Freeze() => _frozen = true; + + /// Resumes forwarding for connections made from now on. + public void Unfreeze() => _frozen = false; + + private async Task AcceptLoopAsync() + { + try + { + while (!_cts.IsCancellationRequested) + { + var client = await _listener.AcceptTcpClientAsync(_cts.Token).ConfigureAwait(false); + var upstream = new TcpClient(); + await upstream.ConnectAsync(_targetHost, _targetPort, _cts.Token).ConfigureAwait(false); + lock (_sockets) { _sockets.Add(client); _sockets.Add(upstream); } + // A connection accepted while frozen stays frozen for its whole life (its peer is "gone"). + var frozenAtBirth = _frozen; + _ = PumpAsync(client.GetStream(), upstream.GetStream(), countFrames: true, frozenAtBirth); + _ = PumpAsync(upstream.GetStream(), client.GetStream(), countFrames: false, frozenAtBirth); + } + } + catch { /* disposed */ } + } + + private async Task PumpAsync(NetworkStream from, NetworkStream to, bool countFrames, bool frozenAtBirth) + { + var buffer = new byte[8192]; + var pending = new List(); + try + { + while (!_cts.IsCancellationRequested) + { + var n = await from.ReadAsync(buffer, _cts.Token).ConfigureAwait(false); + if (n == 0) return; + var frozen = _frozen || frozenAtBirth; + if (countFrames && frozen) + { + // Count complete TPKT frames: 0x03 0x00 then a big-endian total length. + pending.AddRange(buffer.AsSpan(0, n).ToArray()); + while (pending.Count >= 4) + { + var len = (pending[2] << 8) | pending[3]; + if (len < 4 || pending.Count < len) break; + pending.RemoveRange(0, len); + Interlocked.Increment(ref _framesWhileFrozen); + } + } + if (frozen) continue; // swallow — the peer never answers + await to.WriteAsync(buffer.AsMemory(0, n), _cts.Token).ConfigureAwait(false); + } + } + catch { /* socket closed / disposed */ } + } + + /// + public ValueTask DisposeAsync() + { + _cts.Cancel(); + _listener.Stop(); + lock (_sockets) foreach (var s in _sockets) s.Dispose(); + _cts.Dispose(); + return ValueTask.CompletedTask; + } +} diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.S7.IntegrationTests/S7_1500/S7_1500FrozenPeerTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.S7.IntegrationTests/S7_1500/S7_1500FrozenPeerTests.cs new file mode 100644 index 00000000..088931de --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.S7.IntegrationTests/S7_1500/S7_1500FrozenPeerTests.cs @@ -0,0 +1,71 @@ +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; + +namespace ZB.MOM.WW.OtOpcUa.Driver.S7.IntegrationTests.S7_1500; + +/// +/// Live gate for R07:P1 over real S7comm: a frozen-but-established peer (produced by an in-test +/// , so the shared fixture is never touched). The first wire op of +/// the pass blows its deadline and marks the connection dead; the rest of the pass must NOT go to +/// the wire. The assertion counts request frames, not seconds. +/// +/// +/// What this gate does and does not prove. Verified by mutation (2026-10-22): on the real +/// stack this test also passes with the driver's dead-handle guard removed, because S7.Net 0.20 +/// honours the deadline's cancellation token and closes its own socket, so the next op fails at once +/// with ConnectionError and sends nothing. The driver guard (proven by +/// S7DeadHandleShortCircuitTests against a fake that does not self-close) is what covers an +/// op that ignores its token — the WaitAsync leg of S7OperationDeadline, where the +/// abandoned socket stays open and every further op would pay the full deadline. This test pins the +/// end-to-end property (one request per frozen pass, then recovery) against S7.Net upgrades. +/// +[Collection(Snap7ServerCollection.Name)] +[Trait("Category", "Integration")] +[Trait("Device", "S7_1500")] +public sealed class S7_1500FrozenPeerTests(Snap7ServerFixture sim) +{ + /// A pass against a frozen peer issues ONE request, then fails the rest fast; it recovers after. + [Fact] + public async Task A_frozen_peer_costs_one_wire_op_per_pass_not_one_per_tag() + { + if (sim.SkipReason is not null) Assert.Skip(sim.SkipReason); + var ct = TestContext.Current.CancellationToken; + + await using var proxy = new FreezableTcpProxy(sim.Host, sim.Port); + var baseOptions = S7_1500Profile.BuildOptions("127.0.0.1", proxy.Port); + var options = new S7DriverOptions + { + Host = baseOptions.Host, Port = baseOptions.Port, CpuType = baseOptions.CpuType, + Rack = baseOptions.Rack, Slot = baseOptions.Slot, + Timeout = TimeSpan.FromSeconds(1), // the per-op deadline that detects the frozen peer + Probe = baseOptions.Probe, + SpanReadMaxGapBytes = -1, // one wire request per tag, so "one per tag" is countable + RawTags = baseOptions.RawTags, + }; + string[] refs = + [ + S7_1500Profile.ProbeTag, S7_1500Profile.SmokeI16Tag, S7_1500Profile.SmokeI32Tag, + S7_1500Profile.SmokeF32Tag, S7_1500Profile.SmokeBoolTag, S7_1500Profile.WriteScratchTag, + ]; + + await using var drv = new S7Driver(options, driverInstanceId: "s7-frozen-peer-live"); + await drv.InitializeAsync("{}", ct); + + // Positive control: through the proxy, every tag reads Good. + (await drv.ReadAsync(refs, ct)).ShouldAllBe(s => s.StatusCode == 0u); + + proxy.Freeze(); + var frozen = await drv.ReadAsync(refs, ct); + frozen.ShouldAllBe(s => s.StatusCode != 0u); + proxy.FramesWhileFrozen.ShouldBe(1, "only the request that discovered the dead connection may go out"); + drv.GetHealth().State.ShouldBe(DriverState.Degraded); + + // Recovery: the next pass reopens (a fresh proxy connection) and reads Good again. + proxy.Unfreeze(); + var recovered = await drv.ReadAsync(refs, ct); + recovered.ShouldAllBe(s => s.StatusCode == 0u); + Convert.ToInt32(recovered[0].Value).ShouldBe(S7_1500Profile.ProbeSeedValue); + drv.GetHealth().State.ShouldBe(DriverState.Healthy); + } +} diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.S7.Tests/S7ArrayWriteAndHealthTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.S7.Tests/S7ArrayWriteAndHealthTests.cs index fab2b496..27c5e20a 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.S7.Tests/S7ArrayWriteAndHealthTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.S7.Tests/S7ArrayWriteAndHealthTests.cs @@ -243,7 +243,9 @@ public sealed class S7ArrayWriteAndHealthTests var (drv, factory) = await TwoTagDriver(); using var _ = drv; factory.Created[0].WriteFaults["DB1.DBW0"] = new PlcException(ErrorCode.WriteData, "Accessing object not allowed"); - factory.Created[0].WriteFaults["DB1.DBW2"] = new IOException("socket glitch"); + // A Degraded failure that is NOT connection-fatal: since R07:P1 a fatal one (an IOException) + // would fail the rest of the batch without I/O, and C must reach the wire to be the success. + factory.Created[0].WriteFaults["DB1.DBW2"] = new PlcException(ErrorCode.SendData, "device glitch"); var res = await drv.WriteAsync( [new WriteRequest("A", (ushort)1), new WriteRequest("B", (ushort)2), new WriteRequest("C", (ushort)3)], diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.S7.Tests/S7DeadHandleShortCircuitTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.S7.Tests/S7DeadHandleShortCircuitTests.cs new file mode 100644 index 00000000..852e9d47 --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.S7.Tests/S7DeadHandleShortCircuitTests.cs @@ -0,0 +1,167 @@ +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; + +namespace ZB.MOM.WW.OtOpcUa.Driver.S7.Tests; + +/// +/// R07:P1 — once a pass has marked the connection dead (a wire op blew its deadline on a frozen +/// peer), the rest of that pass must not go to the wire: each further op would pay the full +/// timeout under the gate, so a 20-span pass took ~100 s and the comms-loss watchdog's clock only +/// started after it. The remaining items are stamped BadCommunicationError without I/O. +/// Every assertion counts wire operations; nothing is timed. +/// +[Trait("Category", "Unit")] +public sealed class S7DeadHandleShortCircuitTests +{ + private const uint BadCommunicationError = 0x80050000u; + + private static S7DriverOptions Options(int maxGap, params S7TagDefinition[] tags) => new() + { + Host = "192.0.2.1", + Timeout = TimeSpan.FromSeconds(5), + Probe = new S7ProbeOptions { Enabled = false }, + SpanReadMaxGapBytes = maxGap, + RawTags = S7RawTags.Entries(tags), + }; + + /// Five isolated words — far enough apart that each is its own span. + private static S7TagDefinition[] FarApartWords() => + [.. Enumerable.Range(0, 5).Select(i => new S7TagDefinition($"W{i}", $"DB{i + 1}.DBW0", S7DataType.UInt16, Writable: true))]; + + private static string[] Refs(int n) => [.. Enumerable.Range(0, n).Select(i => $"W{i}")]; + + private static async Task<(S7Driver, ScriptedS7PlcFactory)> StartAsync(S7DriverOptions options) + { + var factory = new ScriptedS7PlcFactory(); + var drv = new S7Driver(options, "s7-dead-short-circuit", factory); + await drv.InitializeAsync("{}", TestContext.Current.CancellationToken); + return (drv, factory); + } + + /// Per-tag path: the first timeout ends the pass's wire traffic. + [Fact] + public async Task Per_tag_pass_stops_issuing_wire_ops_once_the_handle_is_dead() + { + var ct = TestContext.Current.CancellationToken; + var (drv, factory) = await StartAsync(Options(maxGap: -1, FarApartWords())); + using var _ = drv; + factory.Read = (_, _) => throw new TimeoutException("frozen peer"); + + var before = factory.WireOps; + var r = await drv.ReadAsync(Refs(5), ct); + + (factory.WireOps - before).ShouldBe(1, "only the op that discovered the dead handle may go to the wire"); + r.ShouldAllBe(s => s.StatusCode == BadCommunicationError); + drv.GetHealth().State.ShouldBe(DriverState.Degraded); + + // Positive control: the next pass reopens and serves every tag. + factory.Read = (_, _) => Task.FromResult((ushort)1); + var recovered = await drv.ReadAsync(Refs(5), ct); + recovered.ShouldAllBe(s => s.StatusCode == 0u); + factory.Created.Count.ShouldBe(2); + } + + /// Span path: arrays are block reads even alone, so five arrays are five spans. + [Fact] + public async Task Span_pass_stops_issuing_wire_ops_once_the_handle_is_dead() + { + var ct = TestContext.Current.CancellationToken; + S7TagDefinition[] arrays = + [.. Enumerable.Range(0, 5).Select(i => new S7TagDefinition($"W{i}", $"DB{i + 1}.DBW0", S7DataType.UInt16, Writable: false, ArrayCount: 2))]; + var (drv, factory) = await StartAsync(Options(maxGap: 16, arrays)); + using var _ = drv; + factory.ReadBytes = (_, _, _, _, _) => throw new TimeoutException("frozen peer"); + + var before = factory.WireOps; + var r = await drv.ReadAsync(Refs(5), ct); + + (factory.WireOps - before).ShouldBe(1); + r.ShouldAllBe(s => s.StatusCode == BadCommunicationError); + } + + /// + /// Multi-var path: a chunk that dies on a transport loss used to fall back to one per-tag read + /// per member — each paying the timeout again. Now it marks the handle dead and the fallback is + /// short-circuited. + /// + [Fact] + public async Task Multi_var_transport_loss_does_not_fall_back_to_per_tag_wire_reads() + { + var ct = TestContext.Current.CancellationToken; + var (drv, factory) = await StartAsync(Options(maxGap: 0, FarApartWords())); + using var _ = drv; + factory.ReadMulti = (_, _) => throw new TimeoutException("frozen peer"); + factory.Read = (_, _) => throw new TimeoutException("frozen peer"); + + var before = factory.WireOps; + var r = await drv.ReadAsync(Refs(5), ct); + + (factory.WireOps - before).ShouldBe(1); + r.ShouldAllBe(s => s.StatusCode == BadCommunicationError); + drv.GetHealth().State.ShouldBe(DriverState.Degraded); + } + + /// + /// Positive control for the classification: a multi-var REFUSAL (not a transport loss) still + /// falls back per tag, so the members the CPU will serve are served. + /// + [Fact] + public async Task Multi_var_refusal_still_falls_back_per_tag() + { + var ct = TestContext.Current.CancellationToken; + var (drv, factory) = await StartAsync(Options(maxGap: 0, FarApartWords())); + using var _ = drv; + // The live snap7 shape of a multi-var with one refused item. + factory.ReadMulti = (_, _) => throw new global::S7.Net.PlcException(global::S7.Net.ErrorCode.ReadData, + new global::S7.Net.PlcException(global::S7.Net.ErrorCode.WrongNumberReceivedBytes)); + factory.Read = (_, _) => Task.FromResult((ushort)9); + + var before = factory.WireOps; + var r = await drv.ReadAsync(Refs(5), ct); + + (factory.WireOps - before).ShouldBe(6, "one multi-var request, then one per-tag read each"); + r.ShouldAllBe(s => s.StatusCode == 0u); + factory.Created.Count.ShouldBe(1, "a refusal is not a reason to reopen"); + } + + /// A refusal does not short-circuit the pass: the CPU answered, the handle is fine. + [Fact] + public async Task A_refusal_does_not_short_circuit_the_pass() + { + var ct = TestContext.Current.CancellationToken; + var (drv, factory) = await StartAsync(Options(maxGap: -1, FarApartWords())); + using var _ = drv; + factory.Read = (addr, _) => addr == "DB1.DBW0" + ? throw new Exception("Received error from PLC: Object does not exist.") + : Task.FromResult((ushort)3); + + var before = factory.WireOps; + var r = await drv.ReadAsync(Refs(5), ct); + + (factory.WireOps - before).ShouldBe(5); + r.Skip(1).ShouldAllBe(s => s.StatusCode == 0u); + } + + /// Write batch: the first transport loss fails the rest of the batch without I/O. + [Fact] + public async Task Write_batch_stops_issuing_wire_ops_once_the_handle_is_dead() + { + var ct = TestContext.Current.CancellationToken; + var (drv, factory) = await StartAsync(Options(maxGap: -1, FarApartWords())); + using var _ = drv; + factory.Write = (_, _, _) => throw new System.IO.IOException("connection reset"); + + var before = factory.WireOps; + var w = await drv.WriteAsync([.. Refs(5).Select(r => new WriteRequest(r, (ushort)1))], ct); + + (factory.WireOps - before).ShouldBe(1); + w.ShouldAllBe(x => x.StatusCode == BadCommunicationError); + drv.GetHealth().State.ShouldBe(DriverState.Degraded); + + // Positive control: the next batch reopens and writes. + factory.Write = (_, _, _) => Task.CompletedTask; + (await drv.WriteAsync([.. Refs(5).Select(r => new WriteRequest(r, (ushort)1))], ct)) + .ShouldAllBe(x => x.StatusCode == 0u); + } +}