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; }