perf(drivers): fixed-rate poll pacing with overrun counter; non-blocking unsubscribe

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.</para>
 ///
+///     <para>Pacing is <b>fixed-rate, not fixed-delay</b>: the post-poll wait is
+///     <c>max(interval - elapsed, 0)</c>, 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
+///     <see cref="OverrunCount"/>, and the optional <c>onOverrun</c> callback is invoked with the
+///     measured duration. The failure/backoff branch is deliberately exempt: once
+///     <c>consecutiveFailures &gt; 0</c> with a <c>backoffCap</c> configured, the capped-exponential
+///     schedule wins, because a failing device must not be hammered at fixed rate.</para>
+///
+///     <para><b>No <c>Meter</c> lives here.</b> <c>Core.Abstractions</c> deliberately references
+///     nothing, so <see cref="OverrunCount"/> plus the <c>onOverrun</c> 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.)</para>
+///
 ///     <para>Programmer errors and obviously-fatal exceptions (<see cref="OutOfMemoryException"/>,
 ///     <see cref="ThreadAbortException"/>, <see cref="StackOverflowException"/>,
 ///     <see cref="AccessViolationException"/>) are NOT caught — they propagate and tear the poll
@@ -44,10 +60,12 @@ public sealed class PollGroupEngine : IAsyncDisposable
     private readonly Action<ISubscriptionHandle, string, DataValueSnapshot> _onChange;
     private readonly Action<Exception>? _onError;
     private readonly Action<ISubscriptionHandle, IReadOnlyList<DataChangeItem>>? _onBatch;
+    private readonly Action<TimeSpan>? _onOverrun;
     private readonly TimeSpan _minInterval;
     private readonly TimeSpan? _backoffCap;
     private readonly ConcurrentDictionary<long, SubscriptionState> _subscriptions = new();
     private long _nextId;
+    private long _overrunCount;
 
     /// <summary>Default floor for publishing intervals — matches the Modbus 100 ms cap.</summary>
     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 <paramref name="onChange"/> callbacks. The
     ///     driver forwards it to its own <see cref="IBatchSubscribable.OnDataChangeBatch"/> event.
     ///     <c>null</c> (default) collects nothing and allocates nothing per cycle.</param>
+    /// <param name="onOverrun">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 <see cref="OverrunCount"/>). 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 <c>onError</c>.</param>
     public PollGroupEngine(
         Func<IReadOnlyList<string>, CancellationToken, Task<IReadOnlyList<DataValueSnapshot>>> reader,
         Action<ISubscriptionHandle, string, DataValueSnapshot> onChange,
         TimeSpan? minInterval = null,
         Action<Exception>? onError = null,
         TimeSpan? backoffCap = null,
-        Action<ISubscriptionHandle, IReadOnlyList<DataChangeItem>>? onBatch = null)
+        Action<ISubscriptionHandle, IReadOnlyList<DataChangeItem>>? onBatch = null,
+        Action<TimeSpan>? 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;
     }
 
+    /// <summary>
+    ///     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.
+    /// </summary>
+    public long OverrunCount => Interlocked.Read(ref _overrunCount);
+
     /// <summary>Register a new polled subscription and start its background loop.</summary>
     /// <param name="fullReferences">The list of tag references to poll.</param>
     /// <param name="publishingInterval">The desired polling interval; will be clamped to the configured minimum.</param>
@@ -108,43 +140,94 @@ public sealed class PollGroupEngine : IAsyncDisposable
         return handle;
     }
 
-    /// <summary>Cancel the background loop for a handle returned by <see cref="Subscribe"/>.</summary>
+    /// <summary>
+    ///     Cancel the background loop for a handle returned by <see cref="Subscribe"/>, BLOCKING the
+    ///     calling thread until the in-flight poll drains. Retained for compatibility; prefer
+    ///     <see cref="UnsubscribeAsync"/> 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.
+    /// </summary>
     /// <returns><c>true</c> when the handle was known to the engine and has been torn down.</returns>
     /// <param name="handle">The subscription handle to cancel.</param>
     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;
+    }
+
+    /// <summary>
+    ///     Non-blocking counterpart of <see cref="Unsubscribe"/>: 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.
+    /// </summary>
+    /// <param name="handle">The subscription handle to cancel.</param>
+    /// <returns><c>true</c> when the handle was known to the engine and has been torn down.</returns>
+    public async ValueTask<bool> UnsubscribeAsync(ISubscriptionHandle handle)
+    {
+        if (!TryTake(handle, out var state)) return false;
+        await StopStateAsync(state).ConfigureAwait(false);
+        return true;
+    }
+
+    /// <summary>
+    ///     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 <c>false</c> for the loser.
+    /// </summary>
+    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();
+    }
+
+    /// <summary>Ceiling on how long a teardown waits for an in-flight poll to unwind.</summary>
+    private static readonly TimeSpan DrainTimeout = TimeSpan.FromSeconds(5);
+
     /// <summary>Snapshot of active subscription count — exposed for driver diagnostics.</summary>
     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);
         }
     }
 
+    /// <summary>
+    ///     Record a poll that took longer than its interval: bump <see cref="OverrunCount"/> and
+    ///     hand the measured duration to the optional <c>onOverrun</c> sink. Defensive against a
+    ///     sink that throws, for the same reason <see cref="ReportError"/> is.
+    /// </summary>
+    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 */ }
+    }
+
     /// <summary>
     ///     Programmer-error / process-fatal exception classification: anything that cannot be
     ///     safely "swallowed and retry on the next tick" must escape the poll loop instead.

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));
 
     /// <inheritdoc />
-    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) ----

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));
 
     /// <inheritdoc />
-    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 ----

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));
 
     /// <inheritdoc />
-    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 ----

src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus/ModbusDriver.cs

@@ -1341,13 +1341,14 @@ public sealed class ModbusDriver
         Task.FromResult(_poll.Subscribe(fullReferences, publishingInterval));
 
     /// <inheritdoc />
-    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 ----

src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.S7/S7Driver.cs

@@ -1732,10 +1732,11 @@ public sealed class S7Driver
         => Task.FromResult(_poll.Subscribe(fullReferences, publishingInterval));
 
     /// <inheritdoc />
-    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);
     }
 
     /// <summary>

src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql/SqlDriver.cs

@@ -570,10 +570,11 @@ public sealed class SqlDriver
     }
 
     /// <inheritdoc />
-    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 ----

src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.TwinCAT/TwinCATDriver.cs

@@ -576,17 +576,18 @@ public sealed class TwinCATDriver : IDriver, IReadable, IWritable, ITagDiscovery
     }
 
     /// <inheritdoc />
-    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