fix(abcip): single-flight cold-connect per device — backoff survives the parallel read window

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.
     /// </summary>
+    /// <remarks>
+    ///     The COLD path (no cached handle) runs single-flight per device under
+    ///     <see cref="DeviceState.ConnectGate"/> 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.
+    /// </remarks>
     private async Task<IAbCipTagRuntime> 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];
     }
 
     /// <summary>
@@ -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.
         /// </summary>
         public ConnectionBackoff Backoff { get; } = new(TimeSpan.FromSeconds(1), TimeSpan.FromSeconds(30));
+
+        /// <summary>
+        ///     Per-device connect gate. <see cref="ConnectionBackoff"/> is documented as
+        ///     caller-serialized — it mutates plain fields — so EVERY touch of <see cref="Backoff"/>
+        ///     plus the connect attempt it guards runs under this semaphore. Task 14's bounded-parallel
+        ///     read window otherwise let up to <c>MaxConcurrentReads</c> work items run the
+        ///     ShouldAttempt → create+Initialize → RecordFailure/RecordSuccess sequence at once for
+        ///     DIFFERENT cold tags on the SAME device: every one passed <c>ShouldAttempt</c> before any
+        ///     failure was recorded (N concurrent Forward-Opens at a down device) and the unsynchronized
+        ///     <c>_consecutiveFailures++</c> 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.
+        ///     <para>
+        ///         Deliberately NOT disposed in <see cref="DisposeHandles"/>, unlike the per-tag gates:
+        ///         the probe loop also waits on it, and <see cref="AbCipDriver.ShutdownAsync"/>'s wait
+        ///         for that loop is bounded (10 s), so disposal could race a still-unwinding probe into
+        ///         an <see cref="ObjectDisposedException"/>. A <see cref="SemaphoreSlim"/> whose
+        ///         <c>AvailableWaitHandle</c> was never touched holds no unmanaged resource, and the
+        ///         whole <see cref="DeviceState"/> is dropped on reinitialize.
+        ///     </para>
+        /// </summary>
+        public SemaphoreSlim ConnectGate { get; } = new(1, 1);
         /// <summary>Gets or sets the current host state of this device.</summary>
         public HostState HostState { get; set; } = HostState.Unknown;
         /// <summary>Gets or sets the UTC timestamp when the host state was last changed.</summary>

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);
     }
 
+    /// <summary>
+    ///     Counts + overlaps fake <c>InitializeAsync</c> (the Forward-Open stand-in) calls. Holds
+    ///     every in-flight connect until either <paramref name="target"/> overlap (proving the
+    ///     connects were NOT single-flight) or a short grace period elapses, so a correctly
+    ///     serialised run still completes instead of deadlocking.
+    /// </summary>
+    private sealed class ConnectProbe(int target)
+    {
+        private readonly TaskCompletionSource _release = new(TaskCreationOptions.RunContinuationsAsynchronously);
+        private int _inFlight;
+        private int _attempts;
+
+        /// <summary>Gets the highest number of connect attempts observed in flight simultaneously.</summary>
+        public int MaxConcurrent { get; private set; }
+
+        /// <summary>Gets the total number of connect attempts that reached the wire.</summary>
+        public int Attempts => Volatile.Read(ref _attempts);
+
+        /// <summary>Records one in-flight connect and holds it open until the probe releases.</summary>
+        /// <param name="cancellationToken">The connect's cancellation token.</param>
+        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);
+            }
+        }
+    }
+
+    /// <summary>A fake standing in for a device that is down — every connect attempt fails.</summary>
+    private sealed class DownDeviceFake(AbCipTagCreateParams p, ConnectProbe probe) : FakeAbCipTag(p)
+    {
+        /// <inheritdoc />
+        public override async Task InitializeAsync(CancellationToken cancellationToken)
+        {
+            await probe.HoldAsync(cancellationToken).ConfigureAwait(false);
+            throw new InvalidOperationException("fake Forward-Open failure");
+        }
+    }
+
+    /// <summary>
+    ///     Review fix — cold (never-connected) tags on ONE down device must connect single-flight
+    ///     under the parallel read window. The window let up to <c>MaxConcurrentReads</c> work items
+    ///     run the unsynchronised <c>ShouldAttempt</c> → connect → <c>RecordFailure</c> sequence on
+    ///     the shared per-device <c>ConnectionBackoff</c> concurrently: every cold tag passed
+    ///     <c>ShouldAttempt</c> before any failure was recorded (N simultaneous Forward-Opens — a
+    ///     thundering herd at the device), and the unsynchronised <c>_consecutiveFailures++</c> lost
+    ///     increments, weakening the 05/STAB-8 backoff into a reconnect storm.
+    /// </summary>
+    [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);
+    }
+
     /// <summary>
     ///     Health is written once after the batch: the worst per-item outcome wins, and
     ///     <c>LastSuccessfulRead</c> still advances because at least one tag read succeeded — the

tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.Tests/FakeAbCipTag.cs

@@ -129,8 +129,15 @@ internal sealed class ArrayFakeAbCipTag : FakeAbCipTag
 /// <summary>Test factory that produces <see cref="FakeAbCipTag"/>s and indexes them for assertion.</summary>
 internal sealed class FakeAbCipTagFactory : IAbCipTagFactory
 {
-    /// <summary>Gets a dictionary of created tags indexed by tag name for assertion.</summary>
-    public Dictionary<string, FakeAbCipTag> Tags { get; } = new(StringComparer.OrdinalIgnoreCase);
+    /// <summary>
+    ///     Gets a dictionary of created tags indexed by tag name for assertion. Concurrent because
+    ///     <see cref="AbCipDriver.ReadAsync"/> creates cold runtimes from inside its bounded-parallel
+    ///     read window — two work items for different tags can call <see cref="Create"/> at the same
+    ///     time, which a plain <see cref="Dictionary{TKey,TValue}"/> would answer with a corrupt
+    ///     bucket chain or a lost entry rather than an honest failure.
+    /// </summary>
+    public System.Collections.Concurrent.ConcurrentDictionary<string, FakeAbCipTag> Tags { get; } =
+        new(StringComparer.OrdinalIgnoreCase);
     /// <summary>Gets or sets an optional customization function to override the tag creation.</summary>
     public Func<AbCipTagCreateParams, FakeAbCipTag>? Customise { get; set; }