using System.Collections;
using System.Collections.Concurrent;
using System.Diagnostics;
namespace ZB.MOM.WW.OtOpcUa.Core.Abstractions;
///
/// Shared poll-based subscription engine for drivers whose underlying protocol has no
/// native push model (Modbus, AB CIP, S7, FOCAS). Owns one background Task per subscription
/// that periodically invokes the supplied reader, diffs each snapshot against the last
/// known value, and dispatches a change callback per changed tag. Extracted from
/// ModbusDriver (AB CIP PR 1) so poll-based drivers don't each re-ship the loop,
/// floor logic, and lifecycle plumbing.
///
///
/// The engine is read-path agnostic: it calls the supplied reader delegate
/// and trusts the driver to map protocol errors into .
/// Callbacks fire on: (a) the first poll after subscribe (initial-data push per the OPC UA
/// Part 4 convention), (b) any subsequent poll where the boxed value or status code differs
/// from the previously-seen snapshot.
///
/// Exceptions thrown by the reader on the initial poll or any subsequent poll are
/// caught — the loop continues on the next tick. When an onError callback is supplied
/// to the constructor the caught exception is routed to it so the driver's health surface
/// can record the failure. Without an onError callback the exception is silently
/// swallowed (preserves the original behaviour for drivers that have not opted in yet).
///
/// An optional onBatch callback delivers the SAME changes collected per poll
/// cycle: it is raised at most once per cycle, only when that cycle produced at least one
/// change, and always AFTER every per-tag onChange callback of that same cycle. Drivers
/// forward it to so downstream consumers can
/// take one message per cycle instead of one per tag. Both callbacks fire by design — a
/// consumer subscribes to exactly one of the two. Without an onBatch callback the engine
/// 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, and that is still deliberate.
/// Core.Abstractions references nothing, so plus the
/// onOverrun callback ARE the instrumentation surface — a driver that wants a metric
/// wires its own meter to the callback. The driver-level meter this note used to record as a
/// follow-up is delivered: every one of the seven poll drivers now passes an
/// onOverrun closure that bumps OtOpcUaTelemetry.PollOverruns
/// (otopcua.driver.poll.overrun) on the shared OtOpcUa Meter, tagged with the
/// driver's DriverTypeNames constant and its driver-instance id. The counter lives in
/// Commons.Observability, on the consumer side of the reference graph — this type still
/// owns no Meter and takes no metrics dependency.
///
/// Programmer errors and obviously-fatal exceptions (,
/// , ,
/// ) are NOT caught — they propagate and tear the poll
/// loop down rather than spin a silently-broken subscription.
///
public sealed class PollGroupEngine : IAsyncDisposable
{
private readonly Func, CancellationToken, Task>> _reader;
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);
/// Constructs the engine with the driver-supplied reader, change callback, and optional interval floor / error sink.
/// Driver-supplied batch reader; snapshots MUST be returned in the same
/// order as the input references.
/// Callback invoked per changed tag — the driver forwards to its own
/// event.
/// Interval floor; anything below is clamped. Defaults to 100 ms
/// per .
/// Optional error sink — invoked once per caught reader exception (or
/// 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.
/// Units: a ceiling on the post-poll wait — the
/// longest gap the loop will ever leave between a failed poll and the next attempt. It is
/// not a count, a multiplier, or a total-outage budget: a device that stays down waits
/// backoffCap between every retry, forever.
/// Interaction with — it REPLACES the paced delay,
/// it does not add to it. The loop picks exactly one of two branches per cycle. With no
/// failures outstanding (or no cap configured) the wait is the fixed-RATE remainder
/// max(interval − pollDuration, 0). With a cap configured AND
/// consecutiveFailures > 0, the whole remainder computation is bypassed and the
/// wait becomes ConnectionBackoff.ComputeDelay(interval, failures, cap) — the poll's
/// own elapsed time is NOT subtracted, so the failure branch is a fixed DELAY (cycle =
/// poll + backoff), deliberately unlike the fixed-rate success branch.
/// The interval feeding that schedule is the per-subscription interval
/// already clamped to in , so
/// the first failure's wait equals the clamped interval (shift 0 → 1×) and the floor is
/// honoured for free. One asymmetry worth knowing: ComputeDelay saturates at the cap
/// unconditionally, so a set SMALLER than the clamped
/// interval makes failures retry FASTER than the interval floor — the cap outranks
/// on the failure branch. Configure it above the floor
/// (every driver here uses 30 s against a 100 ms floor).
/// Optional per-cycle batch callback — invoked once per poll cycle that
/// produced at least one change, carrying exactly that cycle's changed items in reader
/// 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? onOverrun = null)
{
ArgumentNullException.ThrowIfNull(reader);
ArgumentNullException.ThrowIfNull(onChange);
_reader = reader;
_onChange = onChange;
_onError = onError;
_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.
/// A handle identifying the new subscription, for later use with .
public ISubscriptionHandle Subscribe(IReadOnlyList fullReferences, TimeSpan publishingInterval)
{
ArgumentNullException.ThrowIfNull(fullReferences);
var id = Interlocked.Increment(ref _nextId);
var cts = new CancellationTokenSource();
var interval = publishingInterval < _minInterval ? _minInterval : publishingInterval;
var handle = new PollSubscriptionHandle(id);
var state = new SubscriptionState(handle, [.. fullReferences], interval, cts);
_subscriptions[id] = state;
state.LoopTask = Task.Run(() => PollLoopAsync(state, cts.Token));
return handle;
}
///
/// 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 (!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))
{
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 { }
var task = state.LoopTask;
if (task is not null)
{
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);
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);
}
finally { lastPollDuration = Stopwatch.GetElapsedTime(started); }
NoteOverrun(state.Interval, lastPollDuration);
while (!ct.IsCancellationRequested)
{
// 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);
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 (with backoff when configured); driver
// health surface logs it via the supplied onError callback.
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.
/// Caveat — this CONFLATES a slow success with a slow failure. It is called
/// from the finally of both poll sites, so the only input is the measured elapsed
/// time; a device that answers in 300 ms on a 200 ms interval and one that throws (or times
/// out) after 300 ms both increment the counter and both raise onOverrun with the
/// same duration. A consumer reading otopcua.driver.poll.overrun therefore learns
/// "the loop cannot keep its cadence", NOT "the device is healthy but slow" — pair it with
/// the driver's health surface (which onError already feeds) to tell the two apart.
/// Disambiguating here would mean threading the cycle's outcome into this method and
/// splitting the counter/callback by it; no consumer needs that yet, so it is deliberately
/// not done.
///
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.
///
private static bool IsFatal(Exception ex)
=> ex is OutOfMemoryException
or StackOverflowException
or AccessViolationException
or ThreadAbortException;
///
/// Forward a caught exception to the optional onError callback. Defensive
/// against an onError implementation that itself throws — that would crash the
/// poll loop and re-introduce the silent-stall failure mode this method exists to prevent.
///
private void ReportError(Exception ex)
{
if (_onError is null) return;
try { _onError(ex); }
catch { /* never let a buggy error sink stop the poll loop */ }
}
private async Task PollOnceAsync(SubscriptionState state, bool forceRaise, CancellationToken ct)
{
var snapshots = await _reader(state.TagReferences, ct).ConfigureAwait(false);
// Validate the reader contract before indexing. A reader that
// returns fewer snapshots than references would silently stall the subscription; surface
// the violation immediately with a descriptive exception instead.
if (snapshots.Count != state.TagReferences.Count)
throw new InvalidOperationException(
$"Reader contract violation: expected {state.TagReferences.Count} snapshots but received {snapshots.Count}. " +
"The reader delegate must return one snapshot per input reference in input order.");
// Collected lazily: nothing is allocated for a cycle with no changes, and nothing at all
// when no batch callback is wired.
List? changed = null;
for (var i = 0; i < state.TagReferences.Count; i++)
{
var tagRef = state.TagReferences[i];
var current = snapshots[i];
var lastSeen = state.LastValues.TryGetValue(tagRef, out var prev) ? prev : default;
if (forceRaise || ValuesAreDifferent(lastSeen?.Value, current.Value) || lastSeen?.StatusCode != current.StatusCode)
{
state.LastValues[tagRef] = current;
_onChange(state.Handle, tagRef, current);
// Capacity hint: a cycle can produce at most one item per subscribed reference, so
// sizing the list once avoids the doubling re-allocations a large group would incur.
if (_onBatch is not null)
(changed ??= new List(state.TagReferences.Count)).Add(new DataChangeItem(tagRef, current));
}
}
// One raise per cycle, after every per-tag callback of that cycle, never empty.
if (changed is { Count: > 0 }) _onBatch!(state.Handle, changed);
}
///
/// Returns true when and
/// represent different values. Array values are compared structurally
/// (element-by-element) so that a driver producing a fresh array instance on every poll
/// does not trigger spurious change events when the contents are identical.
///
private static bool ValuesAreDifferent(object? previous, object? current)
{
if (previous is Array prevArr && current is Array currArr)
return !StructuralComparisons.StructuralEqualityComparer.Equals(prevArr, currArr);
return !Equals(previous, current);
}
/// Cancel every active subscription and await all loop tasks. Idempotent.
/// A task that represents the asynchronous operation.
public async ValueTask DisposeAsync()
{
// Cancel all loops first so they can all start winding down in parallel.
foreach (var state in _subscriptions.Values)
{
try { state.Cts.Cancel(); } catch { }
}
// Await every loop task before disposing CTSs, ensuring no callback fires after disposal.
var waitTasks = _subscriptions.Values
.Select(s => s.LoopTask ?? Task.CompletedTask)
.ToArray();
if (waitTasks.Length > 0)
{
try { await Task.WhenAll(waitTasks).WaitAsync(TimeSpan.FromSeconds(5)).ConfigureAwait(false); }
catch { }
}
foreach (var state in _subscriptions.Values)
{
state.Cts.Dispose();
}
_subscriptions.Clear();
}
private sealed record SubscriptionState(
PollSubscriptionHandle Handle,
IReadOnlyList TagReferences,
TimeSpan Interval,
CancellationTokenSource Cts)
{
/// Gets the cache of last-seen values per tag reference.
public ConcurrentDictionary LastValues { get; }
= new(StringComparer.OrdinalIgnoreCase);
///
/// The background poll-loop task. Assigned immediately after creation in
/// ; awaited during /
/// so disposal is deterministic and no
/// _onChange callback can fire after the caller tears down the subscription.
///
public Task? LoopTask { get; set; }
}
private sealed record PollSubscriptionHandle(long Id) : ISubscriptionHandle
{
///
public string DiagnosticId => $"poll-sub-{Id}";
}
}