dohertj2/lmxopcua
1
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

test(modbus): failing repro for STAB-3 timeout/framing stream desync

Browse Source
This commit is contained in:
Joseph Doherty
2026-07-13 11:35:20 -04:00
parent 1a698cbb97
commit 141c143586
2 changed files with 352 additions and 35 deletions
Download Patch File Download Diff File Expand all files Collapse all files
archreview/plans/R2-09-driver-fleet-batch-plan.md.tasks.json
+227 -35
View File
@@ -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"
]
}
]
}
}
tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus.Tests/ModbusTcpReconnectTests.cs
+125
View File
@@ -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;
/// <summary>The first N responses are withheld entirely — the client blocks until its per-op timeout / caller cancellation fires.</summary>
public int StallFirstNResponses { set => _stallRemaining = value; }
/// <summary>The first N responses echo a corrupted MBAP transaction id (framing violation).</summary>
public int WrongTxIdFirstNResponses { set => _wrongTxRemaining = value; }
/// <summary>The first N responses claim an MBAP length field of 0 (truncated-length framing violation).</summary>
public int InvalidLengthFirstNResponses { set => _invalidLenRemaining = value; }
/// <summary>Gets the number of TCP connections the listener has accepted (a reconnect increments this).</summary>
public int AcceptedConnectionCount => Volatile.Read(ref _acceptedConnections);
/// <summary>Initializes a new instance and starts listening on a loopback port.</summary>
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<Exception>(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 };
/// <summary>A per-op response timeout tears down the socket and (auto-reconnect) resends cleanly.</summary>
[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
}
/// <summary>A TxId-mismatch framing violation tears down the socket and (auto-reconnect) resends cleanly.</summary>
[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);
}
/// <summary>A truncated-length framing violation tears down the socket and (auto-reconnect) resends cleanly.</summary>
[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);
}
/// <summary>Caller cancellation propagates as OCE and must NOT tear down the socket (a legitimate shutdown is not a desync).</summary>
[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<OperationCanceledException>(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);
}
}