From 9cf400f1e8caf5236385e5a06c5a35f875877451 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Sun, 16 Aug 2026 04:26:20 -0400 Subject: [PATCH] perf(drivers): fixed-rate poll pacing with overrun counter; non-blocking unsubscribe --- .../PollGroupEngine.cs | 167 +++++++++++++++--- .../AbCipDriver.cs | 7 +- .../AbLegacyDriver.cs | 7 +- .../FocasDriver.cs | 7 +- .../ModbusDriver.cs | 9 +- .../ZB.MOM.WW.OtOpcUa.Driver.S7/S7Driver.cs | 7 +- .../ZB.MOM.WW.OtOpcUa.Driver.Sql/SqlDriver.cs | 7 +- .../TwinCATDriver.cs | 9 +- .../PollGroupEngineTests.cs | 100 +++++++++++ 9 files changed, 277 insertions(+), 43 deletions(-) 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 8447bda4..fd0e17bc 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/PollGroupEngine.cs +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/PollGroupEngine.cs @@ -1,5 +1,6 @@ using System.Collections; using System.Collections.Concurrent; +using System.Diagnostics; namespace ZB.MOM.WW.OtOpcUa.Core.Abstractions; @@ -33,6 +34,21 @@ namespace ZB.MOM.WW.OtOpcUa.Core.Abstractions; /// behaves byte-identically to before: no batch is collected and no per-cycle allocation is /// made. /// +/// Pacing is fixed-rate, not fixed-delay: the post-poll wait is +/// max(interval - elapsed, 0), so a read that takes 80 ms on a 100 ms interval is +/// followed by a 20 ms wait rather than a full 100 ms one, and the observed cadence stays at +/// the configured interval instead of drifting out to (read + interval). When a poll takes +/// LONGER than the interval the next one starts immediately, the overrun is counted in +/// , and the optional onOverrun callback is invoked with the +/// measured duration. The failure/backoff branch is deliberately exempt: once +/// consecutiveFailures > 0 with a backoffCap configured, the capped-exponential +/// schedule wins, because a failing device must not be hammered at fixed rate. +/// +/// No Meter lives here. Core.Abstractions deliberately references +/// nothing, so plus the onOverrun callback ARE the +/// instrumentation surface — a driver that wants a metric wires its own meter to the callback. +/// (Driver-level meters are a recorded follow-up, not part of this type.) +/// /// Programmer errors and obviously-fatal exceptions (, /// , , /// ) are NOT caught — they propagate and tear the poll @@ -44,10 +60,12 @@ public sealed class PollGroupEngine : IAsyncDisposable private readonly Action _onChange; private readonly Action? _onError; private readonly Action>? _onBatch; + private readonly Action? _onOverrun; private readonly TimeSpan _minInterval; private readonly TimeSpan? _backoffCap; private readonly ConcurrentDictionary _subscriptions = new(); private long _nextId; + private long _overrunCount; /// Default floor for publishing intervals — matches the Modbus 100 ms cap. public static readonly TimeSpan DefaultMinInterval = TimeSpan.FromMilliseconds(100); @@ -73,13 +91,19 @@ public sealed class PollGroupEngine : IAsyncDisposable /// order, after all of the cycle's per-tag callbacks. The /// driver forwards it to its own event. /// null (default) collects nothing and allocates nothing per cycle. + /// Optional overrun sink — invoked with the measured poll duration + /// every time a poll takes longer than its subscription's interval (the same events counted + /// by ). Since this assembly carries no metrics dependency, this + /// callback is how a driver surfaces the condition to its own meter/log. Defensive: a + /// handler that throws is absorbed, exactly like onError. public PollGroupEngine( Func, CancellationToken, Task>> reader, Action onChange, TimeSpan? minInterval = null, Action? onError = null, TimeSpan? backoffCap = null, - Action>? onBatch = null) + Action>? onBatch = null, + Action? onOverrun = null) { ArgumentNullException.ThrowIfNull(reader); ArgumentNullException.ThrowIfNull(onChange); @@ -89,8 +113,16 @@ public sealed class PollGroupEngine : IAsyncDisposable _minInterval = minInterval ?? DefaultMinInterval; _backoffCap = backoffCap; _onBatch = onBatch; + _onOverrun = onOverrun; } + /// + /// Total number of poll cycles, across every subscription of this engine, whose read took + /// longer than the subscription's publishing interval. A non-zero and climbing value means + /// the device cannot sustain the configured rate — the loop is already running flat out. + /// + public long OverrunCount => Interlocked.Read(ref _overrunCount); + /// Register a new polled subscription and start its background loop. /// The list of tag references to poll. /// The desired polling interval; will be clamped to the configured minimum. @@ -108,43 +140,94 @@ public sealed class PollGroupEngine : IAsyncDisposable return handle; } - /// Cancel the background loop for a handle returned by . + /// + /// Cancel the background loop for a handle returned by , BLOCKING the + /// calling thread until the in-flight poll drains. Retained for compatibility; prefer + /// on any path that runs inside an actor mailbox turn or a + /// request pipeline, where blocking a thread for up to 5 s is never acceptable. + /// /// true when the handle was known to the engine and has been torn down. /// The subscription handle to cancel. public bool Unsubscribe(ISubscriptionHandle handle) { - if (handle is PollSubscriptionHandle h && _subscriptions.TryRemove(h.Id, out var state)) + if (!TryTake(handle, out var state)) return false; + StopState(state); + return true; + } + + /// + /// Non-blocking counterpart of : identical teardown semantics (the + /// loop is cancelled and awaited before the CTS is disposed, so no callback can fire after + /// this completes) but it awaits the loop task instead of blocking a thread on it. + /// + /// The subscription handle to cancel. + /// true when the handle was known to the engine and has been torn down. + public async ValueTask UnsubscribeAsync(ISubscriptionHandle handle) + { + if (!TryTake(handle, out var state)) return false; + await StopStateAsync(state).ConfigureAwait(false); + return true; + } + + /// + /// Shared front half of both teardown paths: recognise the handle and remove it from the + /// registry exactly once, so a double unsubscribe (or a race between the sync and async + /// overloads) tears one subscription down once and reports false for the loser. + /// + private bool TryTake(ISubscriptionHandle handle, out SubscriptionState state) + { + if (handle is PollSubscriptionHandle h && _subscriptions.TryRemove(h.Id, out var found)) { - StopState(state); + state = found; return true; } + state = null!; return false; } + // Cancel + drain + dispose, shared by both teardown paths. The loop task is awaited (with a + // generous timeout) before the CTS is disposed so: + // (a) no _onChange callback fires after the caller considers the subscription torn down, and + // (b) the CTS is not disposed while Task.Delay is still holding a reference to its token, + // which can turn OperationCanceledException into ObjectDisposedException. private static void StopState(SubscriptionState state) { try { state.Cts.Cancel(); } catch { } - // Await the loop task (with a generous timeout) before disposing the CTS so: - // (a) no _onChange callback fires after the caller considers the engine torn down, and - // (b) the CTS is not disposed while Task.Delay is still holding a reference to its token, - // which can turn OperationCanceledException into ObjectDisposedException. var task = state.LoopTask; if (task is not null) { - try { task.Wait(TimeSpan.FromSeconds(5)); } catch { } + try { task.Wait(DrainTimeout); } catch { } } state.Cts.Dispose(); } + private static async ValueTask StopStateAsync(SubscriptionState state) + { + try { state.Cts.Cancel(); } catch { } + var task = state.LoopTask; + if (task is not null) + { + try { await task.WaitAsync(DrainTimeout).ConfigureAwait(false); } catch { } + } + state.Cts.Dispose(); + } + + /// Ceiling on how long a teardown waits for an in-flight poll to unwind. + private static readonly TimeSpan DrainTimeout = TimeSpan.FromSeconds(5); + /// Snapshot of active subscription count — exposed for driver diagnostics. public int ActiveSubscriptionCount => _subscriptions.Count; private async Task PollLoopAsync(SubscriptionState state, CancellationToken ct) { var consecutiveFailures = 0; + // Duration of the poll just performed — the fixed-rate pacing subtracts it from the + // interval so the cadence is the interval, not (interval + read time). + TimeSpan lastPollDuration; // 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. + var started = Stopwatch.GetTimestamp(); try { await PollOnceAsync(state, forceRaise: true, ct).ConfigureAwait(false); @@ -161,21 +244,49 @@ public sealed class PollGroupEngine : IAsyncDisposable consecutiveFailures++; ReportError(ex); } + finally { lastPollDuration = Stopwatch.GetElapsedTime(started); } + NoteOverrun(state.Interval, lastPollDuration); while (!ct.IsCancellationRequested) { - // 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; } + // Fixed RATE, not fixed delay: the wait is what remains of the interval after the poll + // that just ran, so a slow-but-succeeding device keeps the configured cadence and a + // poll that overran starts the next one immediately. + // + // The failure branch is deliberately exempt: once a backoff cap is configured AND the + // device is failing, the capped-exponential schedule wins outright — a dead device must + // not be hammered at fixed rate. With no failures ComputeDelay returns the base interval + // anyway, so it is only consulted when it can actually change the answer. + TimeSpan delay; + if (_backoffCap is not null && consecutiveFailures > 0) + { + delay = ConnectionBackoff.ComputeDelay(state.Interval, consecutiveFailures, _backoffCap.Value); + } + else + { + var remaining = state.Interval - lastPollDuration; + delay = remaining > TimeSpan.Zero ? remaining : TimeSpan.Zero; + } + if (delay > TimeSpan.Zero) + { + 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; } + } + else + { + // Overrun: no wait is owed. Still yield, so a reader that completes synchronously + // can never turn a zero delay into a thread-pool-starving tight loop, and so the + // cancellation check below is reached. + await Task.Yield(); + if (ct.IsCancellationRequested) return; + } + + started = Stopwatch.GetTimestamp(); try { await PollOnceAsync(state, forceRaise: false, ct).ConfigureAwait(false); @@ -191,9 +302,25 @@ public sealed class PollGroupEngine : IAsyncDisposable consecutiveFailures++; ReportError(ex); } + finally { lastPollDuration = Stopwatch.GetElapsedTime(started); } + NoteOverrun(state.Interval, lastPollDuration); } } + /// + /// Record a poll that took longer than its interval: bump and + /// hand the measured duration to the optional onOverrun sink. Defensive against a + /// sink that throws, for the same reason is. + /// + private void NoteOverrun(TimeSpan interval, TimeSpan elapsed) + { + if (elapsed <= interval) return; + Interlocked.Increment(ref _overrunCount); + if (_onOverrun is null) return; + try { _onOverrun(elapsed); } + catch { /* never let a buggy overrun sink stop the poll loop */ } + } + /// /// Programmer-error / process-fatal exception classification: anything that cannot be /// safely "swallowed and retry on the next tick" must escape the poll loop instead. 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 5216fb19..510c245e 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip/AbCipDriver.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip/AbCipDriver.cs @@ -448,10 +448,11 @@ public sealed class AbCipDriver : IDriver, IReadable, IWritable, ITagDiscovery, Task.FromResult(_poll.Subscribe(fullReferences, publishingInterval)); /// - public Task UnsubscribeAsync(ISubscriptionHandle handle, CancellationToken cancellationToken) + public async Task UnsubscribeAsync(ISubscriptionHandle handle, CancellationToken cancellationToken) { - _poll.Unsubscribe(handle); - return Task.CompletedTask; + // Async teardown: the engine drains the in-flight poll without blocking the caller's + // thread — this runs inside a DriverInstanceActor mailbox turn. + await _poll.UnsubscribeAsync(handle).ConfigureAwait(false); } // ---- IAlarmSource (ALMD projection) ---- diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbLegacy/AbLegacyDriver.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbLegacy/AbLegacyDriver.cs index 3d6aed9b..5767d489 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbLegacy/AbLegacyDriver.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbLegacy/AbLegacyDriver.cs @@ -489,10 +489,11 @@ public sealed class AbLegacyDriver : IDriver, IReadable, IWritable, ITagDiscover Task.FromResult(_poll.Subscribe(fullReferences, publishingInterval)); /// - public Task UnsubscribeAsync(ISubscriptionHandle handle, CancellationToken cancellationToken) + public async Task UnsubscribeAsync(ISubscriptionHandle handle, CancellationToken cancellationToken) { - _poll.Unsubscribe(handle); - return Task.CompletedTask; + // Async teardown: the engine drains the in-flight poll without blocking the caller's + // thread — this runs inside a DriverInstanceActor mailbox turn. + await _poll.UnsubscribeAsync(handle).ConfigureAwait(false); } // ---- IHostConnectivityProbe ---- diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.FOCAS/FocasDriver.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.FOCAS/FocasDriver.cs index 67aef9eb..8dc5b847 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.FOCAS/FocasDriver.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.FOCAS/FocasDriver.cs @@ -838,10 +838,11 @@ public sealed class FocasDriver : IDriver, IReadable, IWritable, ITagDiscovery, Task.FromResult(_poll.Subscribe(fullReferences, publishingInterval)); /// - public Task UnsubscribeAsync(ISubscriptionHandle handle, CancellationToken cancellationToken) + public async Task UnsubscribeAsync(ISubscriptionHandle handle, CancellationToken cancellationToken) { - _poll.Unsubscribe(handle); - return Task.CompletedTask; + // Async teardown: the engine drains the in-flight poll without blocking the caller's + // thread — this runs inside a DriverInstanceActor mailbox turn. + await _poll.UnsubscribeAsync(handle).ConfigureAwait(false); } // ---- IHostConnectivityProbe ---- diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus/ModbusDriver.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus/ModbusDriver.cs index 713eb27f..0d34449c 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus/ModbusDriver.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus/ModbusDriver.cs @@ -1341,13 +1341,14 @@ public sealed class ModbusDriver Task.FromResult(_poll.Subscribe(fullReferences, publishingInterval)); /// - public Task UnsubscribeAsync(ISubscriptionHandle handle, CancellationToken cancellationToken) + public async Task UnsubscribeAsync(ISubscriptionHandle handle, CancellationToken cancellationToken) { - _poll.Unsubscribe(handle); + // Async teardown: the engine drains the in-flight poll without blocking the caller's + // thread — this runs inside a DriverInstanceActor mailbox turn. + await _poll.UnsubscribeAsync(handle).ConfigureAwait(false); // Drop any partially-collected cycle buffer so a torn-down subscription leaks nothing. - // Unsubscribe awaits the loop task, so no onChange can re-add after this point. + // The teardown above awaits the loop task, so no onChange can re-add after this point. _cycleBatchItems.TryRemove(handle, out _); - return Task.CompletedTask; } // ---- IHostConnectivityProbe ---- diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.S7/S7Driver.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.S7/S7Driver.cs index 1dfafaf5..c93606ed 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.S7/S7Driver.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.S7/S7Driver.cs @@ -1732,10 +1732,11 @@ public sealed class S7Driver => Task.FromResult(_poll.Subscribe(fullReferences, publishingInterval)); /// - public Task UnsubscribeAsync(ISubscriptionHandle handle, CancellationToken cancellationToken) + public async Task UnsubscribeAsync(ISubscriptionHandle handle, CancellationToken cancellationToken) { - _poll.Unsubscribe(handle); - return Task.CompletedTask; + // Async teardown: the engine drains the in-flight poll without blocking the caller's + // thread — this runs inside a DriverInstanceActor mailbox turn. + await _poll.UnsubscribeAsync(handle).ConfigureAwait(false); } /// diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql/SqlDriver.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql/SqlDriver.cs index e0db65c8..0c7b7631 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql/SqlDriver.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql/SqlDriver.cs @@ -570,10 +570,11 @@ public sealed class SqlDriver } /// - public Task UnsubscribeAsync(ISubscriptionHandle handle, CancellationToken cancellationToken) + public async Task UnsubscribeAsync(ISubscriptionHandle handle, CancellationToken cancellationToken) { - _poll.Unsubscribe(handle); - return Task.CompletedTask; + // Async teardown: the engine drains the in-flight poll without blocking the caller's + // thread — this runs inside a DriverInstanceActor mailbox turn. + await _poll.UnsubscribeAsync(handle).ConfigureAwait(false); } // ---- IHostConnectivityProbe ---- diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.TwinCAT/TwinCATDriver.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.TwinCAT/TwinCATDriver.cs index 1775959f..5e1d75c7 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.TwinCAT/TwinCATDriver.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.TwinCAT/TwinCATDriver.cs @@ -576,17 +576,18 @@ public sealed class TwinCATDriver : IDriver, IReadable, IWritable, ITagDiscovery } /// - public Task UnsubscribeAsync(ISubscriptionHandle handle, CancellationToken cancellationToken) + public async Task UnsubscribeAsync(ISubscriptionHandle handle, CancellationToken cancellationToken) { if (handle is NativeSubscriptionHandle native && _nativeSubs.TryRemove(native.Id, out var sub)) { // NativeRegistration.Dispose removes itself from its device registry (so a later replay // won't touch it) and disposes its live handle. foreach (var r in sub.Registrations) { try { r.Dispose(); } catch { } } - return Task.CompletedTask; + return; } - _poll.Unsubscribe(handle); - return Task.CompletedTask; + // Async teardown: the engine drains the in-flight poll without blocking the caller's + // thread — this runs inside a DriverInstanceActor mailbox turn. + await _poll.UnsubscribeAsync(handle).ConfigureAwait(false); } private sealed record NativeSubscriptionHandle(long Id) : ISubscriptionHandle 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 577cdbf4..fafc3cd1 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 @@ -598,6 +598,106 @@ public sealed class PollGroupEngineTests sw.Elapsed.ShouldBeLessThan(TimeSpan.FromSeconds(2)); } + // ---- Fixed-rate pacing + overrun counter + async unsubscribe (Task 11) ---- + + /// + /// A reader slower than the publishing interval must NOT push the cycle out to + /// (read + interval): the loop paces at a fixed RATE, so the post-poll delay is + /// max(interval - elapsed, 0) and the next poll starts immediately. The overrun is + /// counted and reported through the onOverrun callback. + /// + /// + /// Numbers: interval 150 ms, reader 250 ms. Old (delay-after-completion) cycle = 400 ms → + /// ≈5 polls started in 2 s. Fixed-rate cycle = 250 ms → ≈8. The ≥7 assertion sits clear of + /// both the old shape's ceiling and CI jitter. + /// + [Fact] + public async Task A_slow_reader_overrun_is_counted_and_the_next_poll_starts_promptly() + { + var readCount = 0; + async Task> Reader(IReadOnlyList refs, CancellationToken ct) + { + Interlocked.Increment(ref readCount); + await Task.Delay(250, ct); + var now = DateTime.UtcNow; + return refs.Select(_ => new DataValueSnapshot(1, 0u, now, now)).ToList(); + } + + var overruns = new ConcurrentQueue(); + await using var engine = new PollGroupEngine( + Reader, + (_, _, _) => { }, + minInterval: TimeSpan.FromMilliseconds(50), + onOverrun: elapsed => overruns.Enqueue(elapsed)); + + var handle = engine.Subscribe(["X"], TimeSpan.FromMilliseconds(150)); + await Task.Delay(2000, TestContext.Current.CancellationToken); + var observedReads = Volatile.Read(ref readCount); + var observedOverruns = engine.OverrunCount; + await engine.UnsubscribeAsync(handle); + + observedReads.ShouldBeGreaterThanOrEqualTo(7); + observedOverruns.ShouldBeGreaterThanOrEqualTo(4); + overruns.Count.ShouldBeGreaterThanOrEqualTo(4); + overruns.ShouldAllBe(e => e > TimeSpan.FromMilliseconds(150)); + } + + /// + /// Falsification control for the overrun counter: a reader that completes well inside the + /// interval must never register an overrun (an unconditional increment would fail here). + /// + [Fact] + public async Task A_fast_reader_never_overruns() + { + var src = new FakeSource(); + src.Values["X"] = 1; + + var overruns = new ConcurrentQueue(); + await using var engine = new PollGroupEngine( + src.ReadAsync, + (_, _, _) => { }, + minInterval: TimeSpan.FromMilliseconds(50), + onOverrun: e => overruns.Enqueue(e)); + + var handle = engine.Subscribe(["X"], TimeSpan.FromMilliseconds(100)); + await WaitForAsync(() => Volatile.Read(ref src.ReadCount) >= 4, TimeSpan.FromSeconds(3)); + await engine.UnsubscribeAsync(handle); + + Volatile.Read(ref src.ReadCount).ShouldBeGreaterThanOrEqualTo(4); + engine.OverrunCount.ShouldBe(0); + overruns.ShouldBeEmpty(); + } + + /// + /// The async teardown must be behaviourally identical to the sync one: the handle is known, + /// the loop stops, and no further callback fires. Mirrors + /// through UnsubscribeAsync. + /// + [Fact] + public async Task UnsubscribeAsync_stops_callbacks_like_the_sync_path() + { + var src = new FakeSource(); + src.Values["X"] = 1; + + var events = new ConcurrentQueue<(ISubscriptionHandle, string, DataValueSnapshot)>(); + await using var engine = new PollGroupEngine(src.ReadAsync, + (h, r, s) => events.Enqueue((h, r, s))); + + var handle = engine.Subscribe(["X"], TimeSpan.FromMilliseconds(100)); + await WaitForAsync(() => events.Count >= 1, TimeSpan.FromSeconds(1)); + (await engine.UnsubscribeAsync(handle)).ShouldBeTrue(); + var afterUnsub = events.Count; + engine.ActiveSubscriptionCount.ShouldBe(0); + + src.Values["X"] = 999; + await Task.Delay(400, TestContext.Current.CancellationToken); + events.Count.ShouldBe(afterUnsub); + + // An unknown handle is false through the async path too. + (await engine.UnsubscribeAsync(new DummyHandle())).ShouldBeFalse(); + (await engine.UnsubscribeAsync(handle)).ShouldBeFalse(); + } + private sealed record DummyHandle : ISubscriptionHandle { /// Gets a diagnostic identifier for this handle.