Merge R2-09 Driver fleet batch (arch-review round 2) [PR #437]

Findings 05/STAB-3/4/5/8/12/16/17 + CONV-4 + onError (6 sub-batches, 29 tasks):
Modbus desync teardown, AbCip runtime lock, FOCAS concurrent caches, new
ConnectionBackoff + PollGroupEngine v2 (S7 poll fork deleted onto it, reconciled
with R2-01), fleet connect throttles, ResolveHost via resolver, TwinCAT replay
hardening. 29/29 tasks. S7 254/254 verified post-merge; build clean.

tests/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions.Tests/ConnectionBackoffTests.cs

@@ -0,0 +1,98 @@
+using Shouldly;
+using Xunit;
+using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
+
+namespace ZB.MOM.WW.OtOpcUa.Core.Abstractions.Tests;
+
+/// <summary>
+///     Covers <see cref="ConnectionBackoff"/> — the shared capped-exponential backoff extracted
+///     from the S7 poll fork (05/STAB-8; the seam plan R2-01 wires into the S7 connect throttle).
+///     Two surfaces: the static <see cref="ConnectionBackoff.ComputeDelay"/> schedule and the
+///     per-device attempt-throttle instance.
+/// </summary>
+[Trait("Category", "Unit")]
+public sealed class ConnectionBackoffTests
+{
+    private static readonly TimeSpan Base = TimeSpan.FromSeconds(1);
+    private static readonly TimeSpan Cap = TimeSpan.FromSeconds(30);
+
+    /// <summary>Zero (or negative) consecutive failures returns the base interval unchanged.</summary>
+    [Theory]
+    [InlineData(0)]
+    [InlineData(-1)]
+    public void ComputeDelay_NoFailures_ReturnsBaseInterval(int failures)
+        => ConnectionBackoff.ComputeDelay(Base, failures, Cap).ShouldBe(Base);
+
+    /// <summary>The delay doubles per consecutive failure (1×, 2×, 4×, 8×) until it saturates the cap.</summary>
+    [Theory]
+    [InlineData(1, 1)]
+    [InlineData(2, 2)]
+    [InlineData(3, 4)]
+    [InlineData(4, 8)]
+    [InlineData(5, 16)]
+    public void ComputeDelay_DoublesPerFailure(int failures, int expectedSeconds)
+        => ConnectionBackoff.ComputeDelay(Base, failures, Cap)
+            .ShouldBe(TimeSpan.FromSeconds(expectedSeconds));
+
+    /// <summary>Growth saturates at the cap and never exceeds it, even at a large failure count (overflow guard).</summary>
+    [Theory]
+    [InlineData(6)]     // 32s would exceed 30s cap
+    [InlineData(30)]
+    [InlineData(1000)]  // shift saturates; ticks overflow guard returns cap
+    public void ComputeDelay_SaturatesAtCap(int failures)
+        => ConnectionBackoff.ComputeDelay(Base, failures, Cap).ShouldBe(Cap);
+
+    /// <summary>A fresh throttle permits the first attempt immediately.</summary>
+    [Fact]
+    public void ShouldAttempt_FreshInstance_AllowsImmediately()
+    {
+        var backoff = new ConnectionBackoff(Base, Cap);
+        backoff.ShouldAttempt(DateTime.UtcNow).ShouldBeTrue();
+    }
+
+    /// <summary>After a failure the throttle blocks inside the backoff window and reopens once it elapses.</summary>
+    [Fact]
+    public void RecordFailure_BlocksWithinWindow_ReopensAfter()
+    {
+        var backoff = new ConnectionBackoff(Base, Cap);
+        var t0 = new DateTime(2026, 1, 1, 0, 0, 0, DateTimeKind.Utc);
+
+        backoff.RecordFailure(t0);                                    // window = 1s (first failure)
+        backoff.ShouldAttempt(t0).ShouldBeFalse();                    // still inside window
+        backoff.ShouldAttempt(t0.AddMilliseconds(500)).ShouldBeFalse();
+        backoff.ShouldAttempt(t0.AddSeconds(1)).ShouldBeTrue();       // window elapsed
+    }
+
+    /// <summary>Consecutive failures widen the window (1s then 2s).</summary>
+    [Fact]
+    public void RecordFailure_ConsecutiveFailures_WidenWindow()
+    {
+        var backoff = new ConnectionBackoff(Base, Cap);
+        var t0 = new DateTime(2026, 1, 1, 0, 0, 0, DateTimeKind.Utc);
+
+        backoff.RecordFailure(t0);                                    // 1s
+        backoff.RecordFailure(t0.AddSeconds(1));                      // 2nd failure → 2s window
+        backoff.ShouldAttempt(t0.AddSeconds(2)).ShouldBeFalse();      // within the 2s window
+        backoff.ShouldAttempt(t0.AddSeconds(3)).ShouldBeTrue();
+    }
+
+    /// <summary>Success resets immediately — recovery is never delayed by a residual window.</summary>
+    [Fact]
+    public void RecordSuccess_ResetsWindowImmediately()
+    {
+        var backoff = new ConnectionBackoff(Base, Cap);
+        var t0 = new DateTime(2026, 1, 1, 0, 0, 0, DateTimeKind.Utc);
+
+        backoff.RecordFailure(t0);
+        backoff.RecordFailure(t0);                                    // deep in a widened window
+        backoff.ShouldAttempt(t0).ShouldBeFalse();
+
+        backoff.RecordSuccess();
+        backoff.ShouldAttempt(t0).ShouldBeTrue();                     // reset, no residual delay
+
+        // And the schedule restarts from the base window on the next failure.
+        backoff.RecordFailure(t0);
+        backoff.ShouldAttempt(t0.AddMilliseconds(999)).ShouldBeFalse();
+        backoff.ShouldAttempt(t0.AddSeconds(1)).ShouldBeTrue();
+    }
+}

tests/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions.Tests/PollGroupEngineTests.cs

@@ -461,6 +461,143 @@ public sealed class PollGroupEngineTests
         events.Count.ShouldBeGreaterThanOrEqualTo(1);
     }
 
+    // ---- PollGroupEngine v2: failure backoff + caller-token-filtered OCE (CONV-1 / STAB-8 / STAB-14) ----
+
+    /// <summary>
+    ///     Sustained reader failures must stretch the poll cadence (capped exponential backoff)
+    ///     instead of hammering a dead device at the base interval every tick.
+    /// </summary>
+    [Fact]
+    public async Task Backoff_SustainedReaderFailures_StretchesCadenceToCap()
+    {
+        var readCount = 0;
+        Task<IReadOnlyList<DataValueSnapshot>> Reader(IReadOnlyList<string> refs, CancellationToken ct)
+        {
+            Interlocked.Increment(ref readCount);
+            throw new InvalidOperationException("dead device");
+        }
+
+        await using var engine = new PollGroupEngine(
+            Reader,
+            (_, _, _) => { },
+            minInterval: TimeSpan.FromMilliseconds(20),
+            onError: _ => { },
+            backoffCap: TimeSpan.FromMilliseconds(120));
+
+        var handle = engine.Subscribe(["X"], TimeSpan.FromMilliseconds(20));
+        await Task.Delay(700, TestContext.Current.CancellationToken);
+        engine.Unsubscribe(handle);
+
+        // Backoff schedule from a 20 ms base: 20,40,80,120,120,... → ≈8-9 reads in 700 ms.
+        // A no-backoff 20 ms loop would fire ≈35. Assert the cadence is clearly stretched.
+        readCount.ShouldBeGreaterThanOrEqualTo(2);
+        readCount.ShouldBeLessThan(18);
+    }
+
+    /// <summary>A successful poll resets the failure count so the cadence snaps back to the base interval.</summary>
+    [Fact]
+    public async Task Backoff_SuccessResetsToInterval()
+    {
+        var readCount = 0;
+        var generation = 0;
+        Task<IReadOnlyList<DataValueSnapshot>> Reader(IReadOnlyList<string> refs, CancellationToken ct)
+        {
+            // First few reads fail (winding the backoff up toward the cap); thereafter every read
+            // returns a fresh changing value so each successful poll raises a change event.
+            if (Interlocked.Increment(ref readCount) <= 3)
+                throw new InvalidOperationException("warming up failures");
+            var gen = Interlocked.Increment(ref generation);
+            var now = DateTime.UtcNow;
+            return Task.FromResult<IReadOnlyList<DataValueSnapshot>>(
+                refs.Select(_ => new DataValueSnapshot(gen, 0u, now, now)).ToList());
+        }
+
+        var events = new ConcurrentQueue<string>();
+        await using var engine = new PollGroupEngine(
+            Reader,
+            (_, r, _) => events.Enqueue(r),
+            minInterval: TimeSpan.FromMilliseconds(20),
+            onError: _ => { },
+            backoffCap: TimeSpan.FromMilliseconds(300));
+
+        var handle = engine.Subscribe(["X"], TimeSpan.FromMilliseconds(20));
+        // After recovery the cadence must be the fast 20 ms base, not the wound-up cap — so many
+        // change events accumulate quickly. A non-resetting impl stays near the 300 ms cap (≈3).
+        await WaitForAsync(() => events.Count >= 8, TimeSpan.FromSeconds(3));
+        engine.Unsubscribe(handle);
+
+        events.Count.ShouldBeGreaterThanOrEqualTo(8);
+    }
+
+    /// <summary>
+    ///     STAB-14-class engine guard: a reader that throws <see cref="OperationCanceledException"/>
+    ///     WITHOUT the caller/loop token being cancelled (a driver-internal timeout CTS) must be
+    ///     treated as a normal failure — reported + retried — NOT as a loop teardown. Before the
+    ///     fix the bare <c>catch (OCE) { return; }</c> killed the whole subscription on the first
+    ///     such OCE.
+    /// </summary>
+    [Fact]
+    public async Task ReaderOperationCanceled_WithoutCallerCancellation_TreatedAsFailureNotTeardown()
+    {
+        var observed = new ConcurrentQueue<Exception>();
+        var events = new ConcurrentQueue<string>();
+        var readCount = 0;
+        Task<IReadOnlyList<DataValueSnapshot>> Reader(IReadOnlyList<string> refs, CancellationToken ct)
+        {
+            // A driver-internal timeout OCE — the loop token (ct) is NOT cancelled.
+            if (Interlocked.Increment(ref readCount) <= 2)
+                throw new OperationCanceledException("driver-internal read timeout");
+            var now = DateTime.UtcNow;
+            return Task.FromResult<IReadOnlyList<DataValueSnapshot>>(
+                refs.Select(_ => new DataValueSnapshot(1, 0u, now, now)).ToList());
+        }
+
+        await using var engine = new PollGroupEngine(
+            Reader,
+            (_, r, _) => events.Enqueue(r),
+            minInterval: TimeSpan.FromMilliseconds(30),
+            onError: ex => observed.Enqueue(ex),
+            backoffCap: TimeSpan.FromMilliseconds(120));
+
+        var handle = engine.Subscribe(["X"], TimeSpan.FromMilliseconds(30));
+        // The loop must SURVIVE the reader OCEs and eventually deliver a change once the reader
+        // recovers — proving the OCE was not treated as teardown.
+        await WaitForAsync(() => events.Count >= 1, TimeSpan.FromSeconds(3));
+        engine.Unsubscribe(handle);
+
+        events.Count.ShouldBeGreaterThanOrEqualTo(1);
+        observed.Count.ShouldBeGreaterThanOrEqualTo(1);
+        observed.ShouldAllBe(e => e is OperationCanceledException);
+    }
+
+    /// <summary>Caller (loop-token) cancellation still exits the loop promptly — well under the drain timeout.</summary>
+    [Fact]
+    public async Task CallerCancellation_StillExitsPromptly()
+    {
+        // Reader blocks until the loop token is cancelled (a well-behaved cancellable read).
+        Task<IReadOnlyList<DataValueSnapshot>> Reader(IReadOnlyList<string> refs, CancellationToken ct)
+            => Task.Delay(Timeout.Infinite, ct).ContinueWith(
+                _ => (IReadOnlyList<DataValueSnapshot>)new List<DataValueSnapshot>(), ct);
+
+        await using var engine = new PollGroupEngine(
+            Reader,
+            (_, _, _) => { },
+            minInterval: TimeSpan.FromMilliseconds(30),
+            onError: _ => { },
+            backoffCap: TimeSpan.FromMilliseconds(120));
+
+        var handle = engine.Subscribe(["X"], TimeSpan.FromMilliseconds(30));
+        await Task.Delay(60, TestContext.Current.CancellationToken); // let the reader begin blocking
+
+        var sw = System.Diagnostics.Stopwatch.StartNew();
+        engine.Unsubscribe(handle).ShouldBeTrue();
+        sw.Stop();
+
+        // Prompt exit: the caller-token OCE unwinds the loop immediately — nowhere near the 5 s
+        // StopState drain ceiling.
+        sw.Elapsed.ShouldBeLessThan(TimeSpan.FromSeconds(2));
+    }
+
     private sealed record DummyHandle : ISubscriptionHandle
     {
         /// <summary>Gets a diagnostic identifier for this handle.</summary>