feat(core): PollGroupEngine v2 — failure backoff + caller-token-filtered OCE exit (CONV-1/STAB-8; STAB-14-class engine guard)

archreview/plans/R2-09-driver-fleet-batch-plan.md.tasks.json

@@ -71,7 +71,7 @@
     {
       "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",
+      "status": "completed",
       "blockedBy": [
         "B3.1"
       ]

src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/PollGroupEngine.cs

@@ -35,6 +35,7 @@ public sealed class PollGroupEngine : IAsyncDisposable
     private readonly Action<ISubscriptionHandle, string, DataValueSnapshot> _onChange;
     private readonly Action<Exception>? _onError;
     private readonly TimeSpan _minInterval;
+    private readonly TimeSpan? _backoffCap;
     private readonly ConcurrentDictionary<long, SubscriptionState> _subscriptions = new();
     private long _nextId;
 
@@ -52,11 +53,17 @@ public sealed class PollGroupEngine : IAsyncDisposable
     ///     internal contract-violation throw) so the owning driver can route the failure to its
     ///     health surface. Defensive: an <c>onError</c> handler that
     ///     itself throws is silently absorbed so a buggy forwarder cannot crash the poll loop.</param>
+    /// <param name="backoffCap">Optional cap enabling capped-exponential failure backoff
+    ///     (05/STAB-8). When supplied, sustained reader failures stretch the retry cadence
+    ///     (interval, 2×, 4×, … saturating at this cap) instead of hammering a dead device every
+    ///     tick; a successful poll resets to the base interval. <c>null</c> (default) keeps the
+    ///     byte-identical fixed-interval cadence for consumers that have not opted in.</param>
     public PollGroupEngine(
         Func<IReadOnlyList<string>, CancellationToken, Task<IReadOnlyList<DataValueSnapshot>>> reader,
         Action<ISubscriptionHandle, string, DataValueSnapshot> onChange,
         TimeSpan? minInterval = null,
-        Action<Exception>? onError = null)
+        Action<Exception>? onError = null,
+        TimeSpan? backoffCap = null)
     {
         ArgumentNullException.ThrowIfNull(reader);
         ArgumentNullException.ThrowIfNull(onChange);
@@ -64,6 +71,7 @@ public sealed class PollGroupEngine : IAsyncDisposable
         _onChange = onChange;
         _onError = onError;
         _minInterval = minInterval ?? DefaultMinInterval;
+        _backoffCap = backoffCap;
     }
 
     /// <summary>Register a new polled subscription and start its background loop.</summary>
@@ -116,31 +124,54 @@ public sealed class PollGroupEngine : IAsyncDisposable
 
     private async Task PollLoopAsync(SubscriptionState state, CancellationToken ct)
     {
+        var consecutiveFailures = 0;
+
         // Initial-data push: every subscribed tag fires once at subscribe time regardless of
         // whether it has changed, satisfying OPC UA Part 4 initial-value semantics.
-        try { await PollOnceAsync(state, forceRaise: true, ct).ConfigureAwait(false); }
-        catch (OperationCanceledException) { return; }
+        try
+        {
+            await PollOnceAsync(state, forceRaise: true, ct).ConfigureAwait(false);
+            consecutiveFailures = 0;
+        }
+        // Only a teardown OCE (the loop's OWN token) exits the loop. A reader that throws OCE with
+        // the loop token un-cancelled — a driver-internal timeout CTS (the STAB-14/STAB-15 class) —
+        // must NOT tear the loop down: it falls through to the generic handler, is reported, and the
+        // loop retries with backoff. This is the guard that makes the S7 fork migration safe.
+        catch (OperationCanceledException) when (ct.IsCancellationRequested) { return; }
         catch (Exception ex) when (!IsFatal(ex))
         {
             // first-read error tolerated — loop continues; forward to driver health surface.
+            consecutiveFailures++;
             ReportError(ex);
         }
 
         while (!ct.IsCancellationRequested)
         {
-            try { await Task.Delay(state.Interval, ct).ConfigureAwait(false); }
+            // Fixed cadence by default; when a backoff cap is configured, sustained failures
+            // stretch the delay (capped exponential) and a success snaps it back to Interval.
+            var delay = _backoffCap is null
+                ? state.Interval
+                : ConnectionBackoff.ComputeDelay(state.Interval, consecutiveFailures, _backoffCap.Value);
+            try { await Task.Delay(delay, ct).ConfigureAwait(false); }
             catch (OperationCanceledException) { return; }
             // Defensive: the CTS may be disposed by Unsubscribe/DisposeAsync between the
             // cancellation check above and the Task.Delay touching the token. Treat that race
             // as a normal cancellation rather than a fatal exception.
             catch (ObjectDisposedException) { return; }
 
-            try { await PollOnceAsync(state, forceRaise: false, ct).ConfigureAwait(false); }
-            catch (OperationCanceledException) { return; }
+            try
+            {
+                await PollOnceAsync(state, forceRaise: false, ct).ConfigureAwait(false);
+                consecutiveFailures = 0;
+            }
+            // Only teardown (the loop token) exits; a reader OCE with ct un-cancelled backs off
+            // like any other failure instead of killing the loop (STAB-14 engine guard).
+            catch (OperationCanceledException) when (ct.IsCancellationRequested) { return; }
             catch (Exception ex) when (!IsFatal(ex))
             {
-                // transient poll error — loop continues, driver health surface logs it
-                // via the supplied onError callback.
+                // transient poll error — loop continues (with backoff when configured); driver
+                // health surface logs it via the supplied onError callback.
+                consecutiveFailures++;
                 ReportError(ex);
             }
         }

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>