From d5f573b01997044986fdb07b4ecf49dc1da15c10 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Sun, 16 Aug 2026 04:24:59 -0400 Subject: [PATCH] =?UTF-8?q?fix(abcip):=20single-flight=20cold-connect=20pe?= =?UTF-8?q?r=20device=20=E2=80=94=20backoff=20survives=20the=20parallel=20?= =?UTF-8?q?read=20window?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../AbCipDriver.cs | 136 ++++++++++++++---- .../AbCipParallelReadTests.cs | 97 +++++++++++++ .../FakeAbCipTag.cs | 11 +- 3 files changed, 213 insertions(+), 31 deletions(-) diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip/AbCipDriver.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip/AbCipDriver.cs index b07b76c9..5216fb19 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip/AbCipDriver.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip/AbCipDriver.cs @@ -495,6 +495,11 @@ public sealed class AbCipDriver : IDriver, IReadable, IWritable, ITagDiscovery, while (!ct.IsCancellationRequested) { var success = false; + // The probe's backoff verdict for this tick: true = record success, false = record + // failure, null = no signal (a read that returned a non-zero status without throwing + // never recorded either, and still doesn't). Recorded AFTER the try/catch so the + // per-device connect gate is taken outside a catch block — see below. + bool? backoffVerdict = null; try { probeRuntime ??= _tagFactory.Create(probeParams); @@ -508,7 +513,7 @@ public sealed class AbCipDriver : IDriver, IReadable, IWritable, ITagDiscovery, await probeRuntime.ReadAsync(ct).ConfigureAwait(false); success = probeRuntime.GetStatus() == 0; if (success) - state.Backoff.RecordSuccess(); // 05/STAB-8: device reachable — reset the data-path window + backoffVerdict = true; // 05/STAB-8: device reachable — reset the data-path window } catch (OperationCanceledException) when (ct.IsCancellationRequested) { @@ -525,7 +530,34 @@ public sealed class AbCipDriver : IDriver, IReadable, IWritable, ITagDiscovery, try { probeRuntime?.Dispose(); } catch { } probeRuntime = null; state.ProbeInitialized = false; - state.Backoff.RecordFailure(DateTime.UtcNow); // 05/STAB-8: keep the data-path window open + backoffVerdict = false; // 05/STAB-8: keep the data-path window open + } + + // ConnectionBackoff is caller-serialized, and the data path now records under the + // per-device connect gate — so the probe records under it too, or its reset/increment + // could interleave with a concurrent cold-tag connect record and lose one. The gate is + // held by the data path only across ONE connect attempt, so this wait is bounded by a + // Forward-Open timeout; the data path never waits on anything the probe holds, so no + // deadlock is possible. Cancellation exits the loop as it does anywhere else. + if (backoffVerdict is { } verdict) + { + try + { + await state.ConnectGate.WaitAsync(ct).ConfigureAwait(false); + } + catch (OperationCanceledException) + { + break; + } + try + { + if (verdict) state.Backoff.RecordSuccess(); + else state.Backoff.RecordFailure(DateTime.UtcNow); + } + finally + { + state.ConnectGate.Release(); + } } TransitionDeviceState(state, success ? HostState.Running : HostState.Stopped); @@ -665,6 +697,9 @@ public sealed class AbCipDriver : IDriver, IReadable, IWritable, ITagDiscovery, private static int HealthRank(DriverState state) => state switch { + // No ReadOutcome factory produces Faulted today (a read item is Healthy, Degraded, or no + // signal at all); the tier is future-proofing for an outcome source that reports it, so the + // worst-wins order stays correct without a second edit here. DriverState.Faulted => 2, DriverState.Degraded => 1, _ => 0, @@ -1075,6 +1110,13 @@ public sealed class AbCipDriver : IDriver, IReadable, IWritable, ITagDiscovery, /// + initialises the libplctag Tag; subsequent calls reuse the cached handle for the /// lifetime of the device. /// + /// + /// The COLD path (no cached handle) runs single-flight per device under + /// so the backoff check, the Forward-Open, and the + /// failure/success record form one atomic sequence — see that property for why. The HOT path + /// (cached handle) never touches the gate, so the bounded-parallel read window is undiminished + /// in steady state. + /// private async Task EnsureTagRuntimeAsync( DeviceState device, AbCipTagDefinition def, CancellationToken ct) { @@ -1084,37 +1126,50 @@ public sealed class AbCipDriver : IDriver, IReadable, IWritable, ITagDiscovery, ?? throw new InvalidOperationException( $"AbCip tag '{def.Name}' has malformed TagPath '{def.TagPath}'."); - // 05/STAB-8: inside an open backoff window a data-path caller fails fast — no runtime - // create, no Forward-Open. The malformed-path config error above still surfaces (a permanent - // fault must never be masked by the window); only the expensive create+init is throttled. The - // probe loop bypasses this (its own runtime) and drives the failure/success recording. - if (!device.Backoff.ShouldAttempt(DateTime.UtcNow)) - throw new ConnectionThrottledException( - $"AbCip runtime create for '{def.Name}' on {device.Options.HostAddress} throttled — backoff window open after a recent Forward-Open failure."); - - // Review I-1 — an array tag (the EXPLICIT IsArray flag, incl. a 1-element array) sets - // libplctag's elem_count so the read pulls every element in one CIP transaction; the read - // path then boxes them into a typed CLR array. Scalar tags pass count 1 + IsArray false. - var runtime = _tagFactory.Create( - device.BuildCreateParams( - parsed.ToLibplctagName(), _options.Timeout, def.ElementCount, IsArrayTag(def))); + await device.ConnectGate.WaitAsync(ct).ConfigureAwait(false); try { - await runtime.InitializeAsync(ct).ConfigureAwait(false); - } - catch - { + // Re-check under the gate: a sibling cold tag may have connected (or a concurrent write + // may have materialised this very handle) while this caller waited. + if (device.Runtimes.TryGetValue(def.Name, out var raced)) return raced; + + // 05/STAB-8: inside an open backoff window a data-path caller fails fast — no runtime + // create, no Forward-Open. The malformed-path config error above still surfaces (a permanent + // fault must never be masked by the window); only the expensive create+init is throttled. The + // probe loop uses its own runtime, and records under this same gate. + if (!device.Backoff.ShouldAttempt(DateTime.UtcNow)) + throw new ConnectionThrottledException( + $"AbCip runtime create for '{def.Name}' on {device.Options.HostAddress} throttled — backoff window open after a recent Forward-Open failure."); + + // Review I-1 — an array tag (the EXPLICIT IsArray flag, incl. a 1-element array) sets + // libplctag's elem_count so the read pulls every element in one CIP transaction; the read + // path then boxes them into a typed CLR array. Scalar tags pass count 1 + IsArray false. + var runtime = _tagFactory.Create( + device.BuildCreateParams( + parsed.ToLibplctagName(), _options.Timeout, def.ElementCount, IsArrayTag(def))); + try + { + await runtime.InitializeAsync(ct).ConfigureAwait(false); + } + catch + { + runtime.Dispose(); + device.Backoff.RecordFailure(DateTime.UtcNow); // 05/STAB-8: open the backoff window + throw; + } + device.Backoff.RecordSuccess(); // 05/STAB-8: Forward-Open succeeded — reset the window + // The gate makes a cold create single-flight, but a concurrent EvictRuntime (a failed read + // on this handle) can still land between the re-check and the add; only the first TryAdd + // wins. Dispose the loser so it doesn't leak a native tag handle. + if (device.Runtimes.TryAdd(def.Name, runtime)) + return runtime; runtime.Dispose(); - device.Backoff.RecordFailure(DateTime.UtcNow); // 05/STAB-8: open the backoff window - throw; + return device.Runtimes[def.Name]; + } + finally + { + device.ConnectGate.Release(); } - device.Backoff.RecordSuccess(); // 05/STAB-8: Forward-Open succeeded — reset the window - // Two concurrent callers can both miss the cache + both initialize a runtime; only the - // first TryAdd wins. Dispose the loser so it doesn't leak a native tag handle. - if (device.Runtimes.TryAdd(def.Name, runtime)) - return runtime; - runtime.Dispose(); - return device.Runtimes[def.Name]; } /// @@ -1449,6 +1504,29 @@ public sealed class AbCipDriver : IDriver, IReadable, IWritable, ITagDiscovery, /// (its own runtime) and drives failure/success so a recovered device resets the window. /// public ConnectionBackoff Backoff { get; } = new(TimeSpan.FromSeconds(1), TimeSpan.FromSeconds(30)); + + /// + /// Per-device connect gate. is documented as + /// caller-serialized — it mutates plain fields — so EVERY touch of + /// plus the connect attempt it guards runs under this semaphore. Task 14's bounded-parallel + /// read window otherwise let up to MaxConcurrentReads work items run the + /// ShouldAttempt → create+Initialize → RecordFailure/RecordSuccess sequence at once for + /// DIFFERENT cold tags on the SAME device: every one passed ShouldAttempt before any + /// failure was recorded (N concurrent Forward-Opens at a down device) and the unsynchronized + /// _consecutiveFailures++ lost increments, weakening the 05/STAB-8 window into a + /// reconnect storm. Cold-tag creation is therefore single-flight per device; an + /// already-created runtime bypasses the gate entirely, so the steady-state read path keeps + /// the full window. + /// + /// Deliberately NOT disposed in , unlike the per-tag gates: + /// the probe loop also waits on it, and 's wait + /// for that loop is bounded (10 s), so disposal could race a still-unwinding probe into + /// an . A whose + /// AvailableWaitHandle was never touched holds no unmanaged resource, and the + /// whole is dropped on reinitialize. + /// + /// + public SemaphoreSlim ConnectGate { get; } = new(1, 1); /// Gets or sets the current host state of this device. public HostState HostState { get; set; } = HostState.Unknown; /// Gets or sets the UTC timestamp when the host state was last changed. diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.Tests/AbCipParallelReadTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.Tests/AbCipParallelReadTests.cs index 71630e1f..2ddc1281 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.Tests/AbCipParallelReadTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.Tests/AbCipParallelReadTests.cs @@ -168,6 +168,103 @@ public sealed class AbCipParallelReadTests snapshots.ShouldAllBe(s => s.StatusCode == AbCipStatusMapper.Good); } + /// + /// Counts + overlaps fake InitializeAsync (the Forward-Open stand-in) calls. Holds + /// every in-flight connect until either overlap (proving the + /// connects were NOT single-flight) or a short grace period elapses, so a correctly + /// serialised run still completes instead of deadlocking. + /// + private sealed class ConnectProbe(int target) + { + private readonly TaskCompletionSource _release = new(TaskCreationOptions.RunContinuationsAsynchronously); + private int _inFlight; + private int _attempts; + + /// Gets the highest number of connect attempts observed in flight simultaneously. + public int MaxConcurrent { get; private set; } + + /// Gets the total number of connect attempts that reached the wire. + public int Attempts => Volatile.Read(ref _attempts); + + /// Records one in-flight connect and holds it open until the probe releases. + /// The connect's cancellation token. + public async Task HoldAsync(CancellationToken cancellationToken) + { + Interlocked.Increment(ref _attempts); + var n = Interlocked.Increment(ref _inFlight); + lock (this) + { + if (n > MaxConcurrent) MaxConcurrent = n; + } + try + { + if (n >= target) _release.TrySetResult(); + try + { + await _release.Task.WaitAsync(TimeSpan.FromMilliseconds(150), cancellationToken) + .ConfigureAwait(false); + } + catch (TimeoutException) + { + // No second connect overlapped — the single-flight case. + } + } + finally + { + Interlocked.Decrement(ref _inFlight); + } + } + } + + /// A fake standing in for a device that is down — every connect attempt fails. + private sealed class DownDeviceFake(AbCipTagCreateParams p, ConnectProbe probe) : FakeAbCipTag(p) + { + /// + public override async Task InitializeAsync(CancellationToken cancellationToken) + { + await probe.HoldAsync(cancellationToken).ConfigureAwait(false); + throw new InvalidOperationException("fake Forward-Open failure"); + } + } + + /// + /// Review fix — cold (never-connected) tags on ONE down device must connect single-flight + /// under the parallel read window. The window let up to MaxConcurrentReads work items + /// run the unsynchronised ShouldAttempt → connect → RecordFailure sequence on + /// the shared per-device ConnectionBackoff concurrently: every cold tag passed + /// ShouldAttempt before any failure was recorded (N simultaneous Forward-Opens — a + /// thundering herd at the device), and the unsynchronised _consecutiveFailures++ lost + /// increments, weakening the 05/STAB-8 backoff into a reconnect storm. + /// + [Fact] + public async Task Cold_tags_on_one_down_device_connect_single_flight() + { + var probe = new ConnectProbe(target: 2); + var factory = new FakeAbCipTagFactory { Customise = p => new DownDeviceFake(p, probe) }; + var drv = NewDriver(factory, maxConcurrentReads: 8, + Tag("A"), Tag("B"), Tag("C"), Tag("D"), Tag("E"), Tag("F")); + await drv.InitializeAsync("{}", CancellationToken.None); + + var snapshots = await drv.ReadAsync(["A", "B", "C", "D", "E", "F"], CancellationToken.None); + + probe.MaxConcurrent.ShouldBe(1, + "cold connects against ONE device must be single-flight — the backoff check + connect + " + + "record sequence is not internally synchronized"); + probe.Attempts.ShouldBe(1, + "the one recorded failure must be visible to every sibling still inside the read window, " + + "so only ONE Forward-Open reaches a down device per batch"); + snapshots.Count.ShouldBe(6); + snapshots.ShouldAllBe(s => s.StatusCode == AbCipStatusMapper.BadCommunicationError); + + // The failure increment landed and is durable: the 05/STAB-8 window opened by that single + // recorded failure (base 1 s) is still open, so a second batch makes NO further attempt. + var again = await drv.ReadAsync(["A", "B", "C", "D", "E", "F"], CancellationToken.None); + + probe.Attempts.ShouldBe(1, + "no lost increment — the recorded failure must still hold the backoff window shut"); + again.ShouldAllBe(s => s.StatusCode == AbCipStatusMapper.BadCommunicationError); + } + /// /// Health is written once after the batch: the worst per-item outcome wins, and /// LastSuccessfulRead still advances because at least one tag read succeeded — the diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.Tests/FakeAbCipTag.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.Tests/FakeAbCipTag.cs index 91986a20..7eb01fc5 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.Tests/FakeAbCipTag.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.Tests/FakeAbCipTag.cs @@ -129,8 +129,15 @@ internal sealed class ArrayFakeAbCipTag : FakeAbCipTag /// Test factory that produces s and indexes them for assertion. internal sealed class FakeAbCipTagFactory : IAbCipTagFactory { - /// Gets a dictionary of created tags indexed by tag name for assertion. - public Dictionary Tags { get; } = new(StringComparer.OrdinalIgnoreCase); + /// + /// Gets a dictionary of created tags indexed by tag name for assertion. Concurrent because + /// creates cold runtimes from inside its bounded-parallel + /// read window — two work items for different tags can call at the same + /// time, which a plain would answer with a corrupt + /// bucket chain or a lost entry rather than an honest failure. + /// + public System.Collections.Concurrent.ConcurrentDictionary Tags { get; } = + new(StringComparer.OrdinalIgnoreCase); /// Gets or sets an optional customization function to override the tag creation. public Func? Customise { get; set; }