From 141c143586e48a21e1e1e1bee07689eb859ad3d4 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Mon, 13 Jul 2026 11:35:20 -0400 Subject: [PATCH] test(modbus): failing repro for STAB-3 timeout/framing stream desync --- ...2-09-driver-fleet-batch-plan.md.tasks.json | 262 +++++++++++++++--- .../ModbusTcpReconnectTests.cs | 125 +++++++++ 2 files changed, 352 insertions(+), 35 deletions(-) diff --git a/archreview/plans/R2-09-driver-fleet-batch-plan.md.tasks.json b/archreview/plans/R2-09-driver-fleet-batch-plan.md.tasks.json index 6292c0d7..76641699 100644 --- a/archreview/plans/R2-09-driver-fleet-batch-plan.md.tasks.json +++ b/archreview/plans/R2-09-driver-fleet-batch-plan.md.tasks.json @@ -3,39 +3,231 @@ "lastUpdated": "2026-07-12", "notes": "Sub-batches B1/B2/B5/B6 are mutually independent PRs; B4 depends on B3 tasks B3.1-B3.2. Ordering inside a sub-batch is encoded via blockedBy. Concurrency tasks are high-risk (see plan Classification fields).", "tasks": [ - { "id": "B1.1", "subject": "B1: failing Modbus transport-desync tests (stall / wrong TxId / truncated header / caller-cancel pin) in ModbusTcpReconnectTests", "status": "pending", "blockedBy": [] }, - { "id": "B1.2", "subject": "B1: ModbusTransportDesyncException + teardown-before-propagate in SendOnceAsync (timeout-OCE + InvalidDataException) — STAB-3", "status": "pending", "blockedBy": ["B1.1"] }, - { "id": "B1.3", "subject": "B1: full Modbus suites green + optional exception_injector live gate + merge PR 1", "status": "pending", "blockedBy": ["B1.2"] }, - - { "id": "B2.1", "subject": "B2: failing AbCipRuntimeConcurrencyTests (reentrancy-guarded FakeAbCipTag hammer) — STAB-4 repro", "status": "pending", "blockedBy": [] }, - { "id": "B2.2", "subject": "B2: port _runtimeLocks + GetRuntimeLock + dispose onto AbCip DeviceState (AbLegacy pattern)", "status": "pending", "blockedBy": ["B2.1"] }, - { "id": "B2.3", "subject": "B2: wrap ReadSingleAsync/ReadGroupAsync/WriteAsync critical sections in the per-runtime lock; AbCip suite green", "status": "pending", "blockedBy": ["B2.2"] }, - { "id": "B2.4", "subject": "B2: failing FocasFixedTreeConcurrencyTests (writer-vs-reader stress on the four DeviceState caches) — STAB-5 repro", "status": "pending", "blockedBy": [] }, - { "id": "B2.5", "subject": "B2: convert the four FOCAS caches to ConcurrentDictionary; FOCAS suite green; merge PR 2", "status": "pending", "blockedBy": ["B2.3", "B2.4"] }, - - { "id": "B3.1", "subject": "B3: ConnectionBackoff primitive in Core.Abstractions (static ComputeDelay ported from S7 + instance attempt-throttle) + ConnectionBackoffTests — STAB-8 seam, hand-off to R2-01", "status": "pending", "blockedBy": [] }, - { "id": "B3.2", "subject": "B3: PollGroupEngine v2 — backoffCap + consecutiveFailures + caller-token-filtered OCE catches (STAB-14-class engine guard) + engine tests FAIL->PASS", "status": "pending", "blockedBy": ["B3.1"] }, - { "id": "B3.3", "subject": "B3: wire onError->health + 30s backoffCap into Modbus/AbCip/AbLegacy engine ctors (tests first per driver) — STAB-9", "status": "pending", "blockedBy": ["B3.2"] }, - { "id": "B3.4", "subject": "B3: wire onError->health + 30s backoffCap into TwinCAT/FOCAS engine ctors (tests first per driver) — STAB-9", "status": "pending", "blockedBy": ["B3.2"] }, - { "id": "B3.5", "subject": "B3: failing S7PollEngineMigrationTests (array-cadence PERF-3, rapid sub/unsub STAB-6, sustained-failure health+backoff)", "status": "pending", "blockedBy": ["B3.2"] }, - { "id": "B3.6", "subject": "B3: retire the S7 poll fork onto PollGroupEngine (delete PollLoopAsync/PollOnceAsync/HandlePollFailure/SubscriptionState; ComputeBackoffDelay -> ConnectionBackoff) — CONV-1; cross-ref R2-01", "status": "pending", "blockedBy": ["B3.3", "B3.4", "B3.5"] }, - { "id": "B3.7", "subject": "B3: engine + all five driver suites green; s7 fixture bounce live gate; merge PR 3", "status": "pending", "blockedBy": ["B3.6"] }, - - { "id": "B4.1", "subject": "B4: TwinCAT per-device connect-attempt backoff (data-path gate, probe bypass + instant reset) tests FAIL->PASS — STAB-8", "status": "pending", "blockedBy": ["B3.2"] }, - { "id": "B4.2", "subject": "B4: FOCAS per-device connect-attempt backoff across the three loops, tests FAIL->PASS — STAB-8", "status": "pending", "blockedBy": ["B3.2"] }, - { "id": "B4.3", "subject": "B4: AbCip evict-recreate (Forward Open) create-path throttle, cache-hit unaffected, tests FAIL->PASS — STAB-8", "status": "pending", "blockedBy": ["B3.2"] }, - { "id": "B4.4", "subject": "B4: S7 seam comment pointing R2-01 at ConnectionBackoff; suites green; merge PR 4", "status": "pending", "blockedBy": ["B4.1", "B4.2", "B4.3"] }, - - { "id": "B5.1", "subject": "B5: failing ResolveHost tests in AbCip/AbLegacy/TwinCAT/FOCAS (equipment ref for device B must key B; fallback regression pins) — CONV-4 repro", "status": "pending", "blockedBy": [] }, - { "id": "B5.2", "subject": "B5: rewrite the four multi-device ResolveHost bodies via EquipmentTagRefResolver + empty-host guard", "status": "pending", "blockedBy": ["B5.1"] }, - { "id": "B5.3", "subject": "B5: Modbus — optional unitId in ModbusEquipmentTagParser + ResolveHost via resolver (per-unit breaker key); scope-note vs R2-11", "status": "pending", "blockedBy": ["B5.1"] }, - { "id": "B5.4", "subject": "B5: all five driver suites green; merge PR 5", "status": "pending", "blockedBy": ["B5.2", "B5.3"] }, - - { "id": "B6.1", "subject": "B6: failing ReplayFailure_EmitsBadQuality_AndDegradesHealth in TwinCATReconnectReplayTests — STAB-16 repro", "status": "pending", "blockedBy": [] }, - { "id": "B6.2", "subject": "B6: NativeRegistration HasLiveHandle/MarkHandleDead/OnUnavailable + replay-catch Bad-quality emit + health degrade — STAB-16", "status": "pending", "blockedBy": ["B6.1"] }, - { "id": "B6.3", "subject": "B6: RetryDeadRegistrationsAsync on successful probe tick (under ConnectGate), tests FAIL->PASS — STAB-16", "status": "pending", "blockedBy": ["B6.2"] }, - { "id": "B6.4", "subject": "B6: post-SwapHandle ownership re-check retracts the fresh ADS notification when unsubscribe raced the replay, tests FAIL->PASS — STAB-17", "status": "pending", "blockedBy": ["B6.3"] }, - { "id": "B6.5", "subject": "B6: RecycleClientAsync takes the probe ct (no CancellationToken.None gate wait) + in-catch OCE wrap, tests FAIL->PASS — STAB-12 compounded part", "status": "pending", "blockedBy": ["B6.4"] }, - { "id": "B6.6", "subject": "B6: full TwinCAT suite green (fake-client authoritative, no TC3 fixture); merge PR 6", "status": "pending", "blockedBy": ["B6.5"] } + { + "id": "B1.1", + "subject": "B1: failing Modbus transport-desync tests (stall / wrong TxId / truncated header / caller-cancel pin) in ModbusTcpReconnectTests", + "status": "completed", + "blockedBy": [] + }, + { + "id": "B1.2", + "subject": "B1: ModbusTransportDesyncException + teardown-before-propagate in SendOnceAsync (timeout-OCE + InvalidDataException) \u2014 STAB-3", + "status": "pending", + "blockedBy": [ + "B1.1" + ] + }, + { + "id": "B1.3", + "subject": "B1: full Modbus suites green + optional exception_injector live gate + merge PR 1", + "status": "pending", + "blockedBy": [ + "B1.2" + ] + }, + { + "id": "B2.1", + "subject": "B2: failing AbCipRuntimeConcurrencyTests (reentrancy-guarded FakeAbCipTag hammer) \u2014 STAB-4 repro", + "status": "pending", + "blockedBy": [] + }, + { + "id": "B2.2", + "subject": "B2: port _runtimeLocks + GetRuntimeLock + dispose onto AbCip DeviceState (AbLegacy pattern)", + "status": "pending", + "blockedBy": [ + "B2.1" + ] + }, + { + "id": "B2.3", + "subject": "B2: wrap ReadSingleAsync/ReadGroupAsync/WriteAsync critical sections in the per-runtime lock; AbCip suite green", + "status": "pending", + "blockedBy": [ + "B2.2" + ] + }, + { + "id": "B2.4", + "subject": "B2: failing FocasFixedTreeConcurrencyTests (writer-vs-reader stress on the four DeviceState caches) \u2014 STAB-5 repro", + "status": "pending", + "blockedBy": [] + }, + { + "id": "B2.5", + "subject": "B2: convert the four FOCAS caches to ConcurrentDictionary; FOCAS suite green; merge PR 2", + "status": "pending", + "blockedBy": [ + "B2.3", + "B2.4" + ] + }, + { + "id": "B3.1", + "subject": "B3: ConnectionBackoff primitive in Core.Abstractions (static ComputeDelay ported from S7 + instance attempt-throttle) + ConnectionBackoffTests \u2014 STAB-8 seam, hand-off to R2-01", + "status": "pending", + "blockedBy": [] + }, + { + "id": "B3.2", + "subject": "B3: PollGroupEngine v2 \u2014 backoffCap + consecutiveFailures + caller-token-filtered OCE catches (STAB-14-class engine guard) + engine tests FAIL->PASS", + "status": "pending", + "blockedBy": [ + "B3.1" + ] + }, + { + "id": "B3.3", + "subject": "B3: wire onError->health + 30s backoffCap into Modbus/AbCip/AbLegacy engine ctors (tests first per driver) \u2014 STAB-9", + "status": "pending", + "blockedBy": [ + "B3.2" + ] + }, + { + "id": "B3.4", + "subject": "B3: wire onError->health + 30s backoffCap into TwinCAT/FOCAS engine ctors (tests first per driver) \u2014 STAB-9", + "status": "pending", + "blockedBy": [ + "B3.2" + ] + }, + { + "id": "B3.5", + "subject": "B3: failing S7PollEngineMigrationTests (array-cadence PERF-3, rapid sub/unsub STAB-6, sustained-failure health+backoff)", + "status": "pending", + "blockedBy": [ + "B3.2" + ] + }, + { + "id": "B3.6", + "subject": "B3: retire the S7 poll fork onto PollGroupEngine (delete PollLoopAsync/PollOnceAsync/HandlePollFailure/SubscriptionState; ComputeBackoffDelay -> ConnectionBackoff) \u2014 CONV-1; cross-ref R2-01", + "status": "pending", + "blockedBy": [ + "B3.3", + "B3.4", + "B3.5" + ] + }, + { + "id": "B3.7", + "subject": "B3: engine + all five driver suites green; s7 fixture bounce live gate; merge PR 3", + "status": "pending", + "blockedBy": [ + "B3.6" + ] + }, + { + "id": "B4.1", + "subject": "B4: TwinCAT per-device connect-attempt backoff (data-path gate, probe bypass + instant reset) tests FAIL->PASS \u2014 STAB-8", + "status": "pending", + "blockedBy": [ + "B3.2" + ] + }, + { + "id": "B4.2", + "subject": "B4: FOCAS per-device connect-attempt backoff across the three loops, tests FAIL->PASS \u2014 STAB-8", + "status": "pending", + "blockedBy": [ + "B3.2" + ] + }, + { + "id": "B4.3", + "subject": "B4: AbCip evict-recreate (Forward Open) create-path throttle, cache-hit unaffected, tests FAIL->PASS \u2014 STAB-8", + "status": "pending", + "blockedBy": [ + "B3.2" + ] + }, + { + "id": "B4.4", + "subject": "B4: S7 seam comment pointing R2-01 at ConnectionBackoff; suites green; merge PR 4", + "status": "pending", + "blockedBy": [ + "B4.1", + "B4.2", + "B4.3" + ] + }, + { + "id": "B5.1", + "subject": "B5: failing ResolveHost tests in AbCip/AbLegacy/TwinCAT/FOCAS (equipment ref for device B must key B; fallback regression pins) \u2014 CONV-4 repro", + "status": "pending", + "blockedBy": [] + }, + { + "id": "B5.2", + "subject": "B5: rewrite the four multi-device ResolveHost bodies via EquipmentTagRefResolver + empty-host guard", + "status": "pending", + "blockedBy": [ + "B5.1" + ] + }, + { + "id": "B5.3", + "subject": "B5: Modbus \u2014 optional unitId in ModbusEquipmentTagParser + ResolveHost via resolver (per-unit breaker key); scope-note vs R2-11", + "status": "pending", + "blockedBy": [ + "B5.1" + ] + }, + { + "id": "B5.4", + "subject": "B5: all five driver suites green; merge PR 5", + "status": "pending", + "blockedBy": [ + "B5.2", + "B5.3" + ] + }, + { + "id": "B6.1", + "subject": "B6: failing ReplayFailure_EmitsBadQuality_AndDegradesHealth in TwinCATReconnectReplayTests \u2014 STAB-16 repro", + "status": "pending", + "blockedBy": [] + }, + { + "id": "B6.2", + "subject": "B6: NativeRegistration HasLiveHandle/MarkHandleDead/OnUnavailable + replay-catch Bad-quality emit + health degrade \u2014 STAB-16", + "status": "pending", + "blockedBy": [ + "B6.1" + ] + }, + { + "id": "B6.3", + "subject": "B6: RetryDeadRegistrationsAsync on successful probe tick (under ConnectGate), tests FAIL->PASS \u2014 STAB-16", + "status": "pending", + "blockedBy": [ + "B6.2" + ] + }, + { + "id": "B6.4", + "subject": "B6: post-SwapHandle ownership re-check retracts the fresh ADS notification when unsubscribe raced the replay, tests FAIL->PASS \u2014 STAB-17", + "status": "pending", + "blockedBy": [ + "B6.3" + ] + }, + { + "id": "B6.5", + "subject": "B6: RecycleClientAsync takes the probe ct (no CancellationToken.None gate wait) + in-catch OCE wrap, tests FAIL->PASS \u2014 STAB-12 compounded part", + "status": "pending", + "blockedBy": [ + "B6.4" + ] + }, + { + "id": "B6.6", + "subject": "B6: full TwinCAT suite green (fake-client authoritative, no TC3 fixture); merge PR 6", + "status": "pending", + "blockedBy": [ + "B6.5" + ] + } ] -} +} \ No newline at end of file diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus.Tests/ModbusTcpReconnectTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus.Tests/ModbusTcpReconnectTests.cs index 47641eb4..a6c242f5 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus.Tests/ModbusTcpReconnectTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus.Tests/ModbusTcpReconnectTests.cs @@ -29,6 +29,23 @@ public sealed class ModbusTcpReconnectTests private readonly CancellationTokenSource _stop = new(); private int _txCount; + // --- Scriptable STAB-3 desync behaviors (each fires for the first N transactions, then + // the server behaves normally). Decremented atomically because an old (torn-down) serve + // task and a fresh one can briefly coexist. --- + private int _stallRemaining; + private int _wrongTxRemaining; + private int _invalidLenRemaining; + private int _acceptedConnections; + + /// The first N responses are withheld entirely — the client blocks until its per-op timeout / caller cancellation fires. + public int StallFirstNResponses { set => _stallRemaining = value; } + /// The first N responses echo a corrupted MBAP transaction id (framing violation). + public int WrongTxIdFirstNResponses { set => _wrongTxRemaining = value; } + /// The first N responses claim an MBAP length field of 0 (truncated-length framing violation). + public int InvalidLengthFirstNResponses { set => _invalidLenRemaining = value; } + /// Gets the number of TCP connections the listener has accepted (a reconnect increments this). + public int AcceptedConnectionCount => Volatile.Read(ref _acceptedConnections); + /// Initializes a new instance and starts listening on a loopback port. public FlakeyModbusServer() { @@ -45,6 +62,7 @@ public sealed class ModbusTcpReconnectTests try { client = await _listener.AcceptTcpClientAsync(_stop.Token); } catch { return; } + Interlocked.Increment(ref _acceptedConnections); _ = Task.Run(() => ServeAsync(client!)); } } @@ -63,8 +81,27 @@ public sealed class ModbusTcpReconnectTests var pdu = new byte[len - 1]; if (!await ReadExactly(stream, pdu)) return; + // Stall: read the request but never answer it — simulate a hung / unreachable + // unit so the client's per-op CancelAfter (or caller token) fires. + if (Interlocked.Decrement(ref _stallRemaining) >= 0) + continue; + var fc = pdu[0]; var qty = (ushort)((pdu[3] << 8) | pdu[4]); + + // Truncated-length framing violation: a 7-byte header whose length field is 0 + // (< the mandatory 1 unit-id byte) — trips the transport's `respLen < 1` guard. + if (Interlocked.Decrement(ref _invalidLenRemaining) >= 0) + { + var bad = new byte[7]; + bad[0] = header[0]; bad[1] = header[1]; + bad[4] = 0; bad[5] = 0; // length field = 0 + bad[6] = header[6]; + await stream.WriteAsync(bad); + await stream.FlushAsync(); + continue; + } + var respPdu = new byte[2 + qty * 2]; respPdu[0] = fc; respPdu[1] = (byte)(qty * 2); @@ -73,6 +110,16 @@ public sealed class ModbusTcpReconnectTests var respLen = (ushort)(1 + respPdu.Length); var adu = new byte[7 + respPdu.Length]; adu[0] = header[0]; adu[1] = header[1]; + + // TxId-mismatch framing violation: emit an otherwise valid, full frame but with + // the transaction id flipped, so the transport reads the whole (unexpected) + // response and rejects it on the TxId guard. + if (Interlocked.Decrement(ref _wrongTxRemaining) >= 0) + { + adu[0] = (byte)~header[0]; + adu[1] = (byte)~header[1]; + } + adu[4] = (byte)(respLen >> 8); adu[5] = (byte)(respLen & 0xFF); adu[6] = header[6]; Buffer.BlockCopy(respPdu, 0, adu, 7, respPdu.Length); @@ -149,4 +196,82 @@ public sealed class ModbusTcpReconnectTests await Should.ThrowAsync(async () => await transport.SendAsync(unitId: 1, pdu, TestContext.Current.CancellationToken)); } + + // --- STAB-3: per-op timeout + framing violations must be classified connection-fatal so the + // socket is torn down instead of left desynchronized. With auto-reconnect on, the transport + // transparently reconnects + resends once and returns the CORRECT response (no stale read); + // the reconnect is observable as a second accepted connection on the server. Before the fix + // the timeout OCE / InvalidDataException are not socket-level, so no teardown happens and the + // SendAsync throws (and the socket stays desynchronized forever). --- + + private static readonly byte[] Fc03Qty1 = { 0x03, 0x00, 0x00, 0x00, 0x01 }; + + /// A per-op response timeout tears down the socket and (auto-reconnect) resends cleanly. + [Fact] + public async Task SendAsync_AfterResponseTimeout_TearsDownAndReconnects() + { + await using var server = new FlakeyModbusServer { StallFirstNResponses = 1 }; + await using var transport = new ModbusTcpTransport( + "127.0.0.1", server.Port, TimeSpan.FromMilliseconds(300), autoReconnect: true); + await transport.ConnectAsync(TestContext.Current.CancellationToken); + + var resp = await transport.SendAsync(unitId: 1, Fc03Qty1, TestContext.Current.CancellationToken); + + resp[0].ShouldBe((byte)0x03); // correct FC, not stale bytes + resp.Length.ShouldBe(2 + 1 * 2); // FC + byteCount + qty*2 data + server.AcceptedConnectionCount.ShouldBe(2); // torn down + reconnected + } + + /// A TxId-mismatch framing violation tears down the socket and (auto-reconnect) resends cleanly. + [Fact] + public async Task SendAsync_AfterTxIdMismatch_TearsDownSocket() + { + await using var server = new FlakeyModbusServer { WrongTxIdFirstNResponses = 1 }; + await using var transport = new ModbusTcpTransport( + "127.0.0.1", server.Port, TimeSpan.FromSeconds(2), autoReconnect: true); + await transport.ConnectAsync(TestContext.Current.CancellationToken); + + var resp = await transport.SendAsync(unitId: 1, Fc03Qty1, TestContext.Current.CancellationToken); + + resp[0].ShouldBe((byte)0x03); + server.AcceptedConnectionCount.ShouldBe(2); + } + + /// A truncated-length framing violation tears down the socket and (auto-reconnect) resends cleanly. + [Fact] + public async Task SendAsync_AfterTruncatedHeader_TearsDownSocket() + { + await using var server = new FlakeyModbusServer { InvalidLengthFirstNResponses = 1 }; + await using var transport = new ModbusTcpTransport( + "127.0.0.1", server.Port, TimeSpan.FromSeconds(2), autoReconnect: true); + await transport.ConnectAsync(TestContext.Current.CancellationToken); + + var resp = await transport.SendAsync(unitId: 1, Fc03Qty1, TestContext.Current.CancellationToken); + + resp[0].ShouldBe((byte)0x03); + server.AcceptedConnectionCount.ShouldBe(2); + } + + /// Caller cancellation propagates as OCE and must NOT tear down the socket (a legitimate shutdown is not a desync). + [Fact] + public async Task SendAsync_CallerCancellation_DoesNotTearDown() + { + await using var server = new FlakeyModbusServer { StallFirstNResponses = 1 }; + await using var transport = new ModbusTcpTransport( + "127.0.0.1", server.Port, TimeSpan.FromSeconds(5), autoReconnect: true); + await transport.ConnectAsync(TestContext.Current.CancellationToken); + + using var cts = new CancellationTokenSource(); + var pending = transport.SendAsync(unitId: 1, Fc03Qty1, cts.Token); + await Task.Delay(150, TestContext.Current.CancellationToken); // let the request land + block on the response read + await cts.CancelAsync(); + + await Should.ThrowAsync(async () => await pending); + + // The socket must still be usable — a subsequent transaction succeeds on the SAME connection + // (no reconnect), proving caller-cancel did not trip the desync teardown. + var resp = await transport.SendAsync(unitId: 1, Fc03Qty1, TestContext.Current.CancellationToken); + resp[0].ShouldBe((byte)0x03); + server.AcceptedConnectionCount.ShouldBe(1); + } }