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.