From ca6530c7f5c4610cad78568879183978e4cde0db Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Mon, 13 Jul 2026 11:51:50 -0400 Subject: [PATCH] =?UTF-8?q?feat(core):=20PollGroupEngine=20v2=20=E2=80=94?= =?UTF-8?q?=20failure=20backoff=20+=20caller-token-filtered=20OCE=20exit?= =?UTF-8?q?=20(CONV-1/STAB-8;=20STAB-14-class=20engine=20guard)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- ...2-09-driver-fleet-batch-plan.md.tasks.json | 2 +- .../PollGroupEngine.cs | 47 +++++- .../PollGroupEngineTests.cs | 137 ++++++++++++++++++ 3 files changed, 177 insertions(+), 9 deletions(-) diff --git a/archreview/plans/R2-09-driver-fleet-batch-plan.md.tasks.json b/archreview/plans/R2-09-driver-fleet-batch-plan.md.tasks.json index 17ddfd7e..886840b4 100644 --- a/archreview/plans/R2-09-driver-fleet-batch-plan.md.tasks.json +++ b/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" ] diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/PollGroupEngine.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/PollGroupEngine.cs index deb33e9a..e0d29342 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/PollGroupEngine.cs +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/PollGroupEngine.cs @@ -35,6 +35,7 @@ public sealed class PollGroupEngine : IAsyncDisposable private readonly Action _onChange; private readonly Action? _onError; private readonly TimeSpan _minInterval; + private readonly TimeSpan? _backoffCap; private readonly ConcurrentDictionary _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 onError handler that /// itself throws is silently absorbed so a buggy forwarder cannot crash the poll loop. + /// 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. null (default) keeps the + /// byte-identical fixed-interval cadence for consumers that have not opted in. public PollGroupEngine( Func, CancellationToken, Task>> reader, Action onChange, TimeSpan? minInterval = null, - Action? onError = null) + Action? 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; } /// Register a new polled subscription and start its background loop. @@ -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); } } diff --git a/tests/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions.Tests/PollGroupEngineTests.cs b/tests/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions.Tests/PollGroupEngineTests.cs index 5e6d0065..577cdbf4 100644 --- a/tests/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions.Tests/PollGroupEngineTests.cs +++ b/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) ---- + + /// + /// Sustained reader failures must stretch the poll cadence (capped exponential backoff) + /// instead of hammering a dead device at the base interval every tick. + /// + [Fact] + public async Task Backoff_SustainedReaderFailures_StretchesCadenceToCap() + { + var readCount = 0; + Task> Reader(IReadOnlyList 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); + } + + /// A successful poll resets the failure count so the cadence snaps back to the base interval. + [Fact] + public async Task Backoff_SuccessResetsToInterval() + { + var readCount = 0; + var generation = 0; + Task> Reader(IReadOnlyList 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>( + refs.Select(_ => new DataValueSnapshot(gen, 0u, now, now)).ToList()); + } + + var events = new ConcurrentQueue(); + 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); + } + + /// + /// STAB-14-class engine guard: a reader that throws + /// 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 catch (OCE) { return; } killed the whole subscription on the first + /// such OCE. + /// + [Fact] + public async Task ReaderOperationCanceled_WithoutCallerCancellation_TreatedAsFailureNotTeardown() + { + var observed = new ConcurrentQueue(); + var events = new ConcurrentQueue(); + var readCount = 0; + Task> Reader(IReadOnlyList 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>( + 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); + } + + /// Caller (loop-token) cancellation still exits the loop promptly — well under the drain timeout. + [Fact] + public async Task CallerCancellation_StillExitsPromptly() + { + // Reader blocks until the loop token is cancelled (a well-behaved cancellable read). + Task> Reader(IReadOnlyList refs, CancellationToken ct) + => Task.Delay(Timeout.Infinite, ct).ContinueWith( + _ => (IReadOnlyList)new List(), 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 { /// Gets a diagnostic identifier for this handle.