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 e802ac57..850aaf6e 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 @@ -103,7 +103,7 @@ { "id": "B3.6", "subject": "B3: retire the S7 poll fork onto PollGroupEngine (delete PollLoopAsync/PollOnceAsync/HandlePollFailure/SubscriptionState; ComputeBackoffDelay -> ConnectionBackoff) \u2014 CONV-1; cross-ref R2-01", - "status": "pending", + "status": "completed", "blockedBy": [ "B3.3", "B3.4", 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 c25eeedc..f97b8b26 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.S7/S7Driver.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.S7/S7Driver.cs @@ -53,11 +53,21 @@ public sealed class S7Driver _resolver = new EquipmentTagRefResolver( r => _tagsByName.TryGetValue(r, out var t) ? t : null, r => S7EquipmentTagParser.TryParse(r, out var d) ? d : null); + // CONV-1: the polling overlay runs on the shared PollGroupEngine (structural array diff + + // drain-before-dispose + capped-exponential failure backoff + caller-token-filtered OCE), + // retiring the bespoke S7 poll fork. minInterval 100 ms matches the S7 scan-mailbox floor; + // backoffCap 30 s matches the retired fork's PollBackoffCap. + _poll = new PollGroupEngine( + reader: ReadAsync, + onChange: (handle, tagRef, snapshot) => + OnDataChange?.Invoke(this, new DataChangeEventArgs(handle, tagRef, snapshot)), + minInterval: TimeSpan.FromMilliseconds(100), + onError: HandlePollError, + backoffCap: TimeSpan.FromSeconds(30)); } // ---- ISubscribable + IHostConnectivityProbe state ---- - private readonly ConcurrentDictionary _subscriptions = new(); - private long _nextSubscriptionId; + private readonly PollGroupEngine _poll; private readonly object _probeLock = new(); private HostState _hostState = HostState.Unknown; private DateTime _hostStateChangedUtc = DateTime.UtcNow; @@ -269,38 +279,28 @@ public sealed class S7Driver // then await all of them with a bounded timeout BEFORE disposing the shared semaphore // and CTS objects. Without the drain, a loop iteration mid-_gate would call Release() // on (or WaitAsync against) a disposed semaphore. - var drain = new List(); + // Drain the polling overlay first — the engine cancels each loop and awaits it before + // disposing its CTS (drain-before-dispose, STAB-6-S7), so no poll iteration touches _gate + // after this returns. + await _poll.DisposeAsync().ConfigureAwait(false); var probeCts = _probeCts; var probeTask = _probeTask; try { probeCts?.Cancel(); } catch { } - if (probeTask is not null) drain.Add(probeTask); - - var subscriptions = _subscriptions.Values.ToArray(); - _subscriptions.Clear(); - foreach (var state in subscriptions) - { - try { state.Cts.Cancel(); } catch { } - drain.Add(state.PollTask); - } - - if (drain.Count > 0) + if (probeTask is not null) { try { - await Task.WhenAll(drain).WaitAsync(DrainTimeout, CancellationToken.None) - .ConfigureAwait(false); + await probeTask.WaitAsync(DrainTimeout, CancellationToken.None).ConfigureAwait(false); } - catch (TimeoutException) { /* a wedged loop — proceed; better than leaking the teardown */ } + catch (TimeoutException) { /* a wedged probe loop — proceed; better than leaking the teardown */ } catch { /* loop faults are already surfaced via health; teardown continues */ } } - // Loops have now observed cancellation and released _gate — safe to dispose the CTSs. + // The probe loop has now observed cancellation and released _gate — safe to dispose the CTS. probeCts?.Dispose(); _probeCts = null; _probeTask = null; - foreach (var state in subscriptions) - state.Cts.Dispose(); _initialized = false; _plcDead = false; @@ -1397,180 +1397,29 @@ public sealed class S7Driver /// public Task SubscribeAsync( IReadOnlyList fullReferences, TimeSpan publishingInterval, CancellationToken cancellationToken) - { - var id = Interlocked.Increment(ref _nextSubscriptionId); - var cts = new CancellationTokenSource(); - // Floor at 100 ms — S7 CPUs scan 2-10 ms but the comms mailbox is processed at most - // once per scan; sub-100 ms polling just queues wire-side with worse latency. - var interval = publishingInterval < TimeSpan.FromMilliseconds(100) - ? TimeSpan.FromMilliseconds(100) - : publishingInterval; - var handle = new S7SubscriptionHandle(id); - var state = new SubscriptionState(handle, [.. fullReferences], interval, cts); - _subscriptions[id] = state; - // Track the poll Task so ShutdownAsync can await it after cancelling — a poll - // iteration mid-_gate would otherwise race the semaphore's disposal. - state.PollTask = Task.Run(() => PollLoopAsync(state, cts.Token), CancellationToken.None); - return Task.FromResult(handle); - } + => Task.FromResult(_poll.Subscribe(fullReferences, publishingInterval)); /// public Task UnsubscribeAsync(ISubscriptionHandle handle, CancellationToken cancellationToken) { - if (handle is S7SubscriptionHandle h && _subscriptions.TryRemove(h.Id, out var state)) - { - state.Cts.Cancel(); - state.Cts.Dispose(); - } + _poll.Unsubscribe(handle); return Task.CompletedTask; } /// - /// Upper bound on the poll-loop backoff window. After enough consecutive failures the - /// loop waits this long between retries instead of , - /// so a subscription against a dropped / uninitialised driver doesn't spin. + /// 05/STAB-9 — routes a poll-loop reader failure (surfaced by the shared + /// 's onError sink) to the driver health surface: logs it and + /// degrades to preserving LastSuccessfulRead. Never + /// downgrades a state (e.g. PUT/GET-denied set by ReadAsync) + /// — Faulted is the stronger, permanent-config signal. Replaces the retired fork's + /// HandlePollFailure; the engine now owns the consecutive-failure count + backoff. /// - private static readonly TimeSpan PollBackoffCap = TimeSpan.FromSeconds(30); - - /// - /// Number of consecutive poll failures before the loop transitions the driver's - /// health to . One stray failure can be transient; - /// a sustained run indicates the operator should see it. Threshold of 1 because the - /// first failure already lives in the LastError surface. - /// - private const int PollFailureHealthThreshold = 1; - - private async Task PollLoopAsync(SubscriptionState state, CancellationToken ct) + /// The exception caught by the poll engine. + private void HandlePollError(Exception ex) { - var consecutiveFailures = 0; - - // Initial-data push per OPC UA Part 4 convention. - try - { - await PollOnceAsync(state, forceRaise: true, ct).ConfigureAwait(false); - consecutiveFailures = 0; - } - // Only a teardown OCE (the loop's own token) exits the loop; any other OCE (e.g. a - // connect-timeout that slipped past the wrapper filter) falls to the backoff path below - // so the loop lives (STAB-14 defense-in-depth). - catch (OperationCanceledException) when (ct.IsCancellationRequested) { return; } - catch (Exception ex) - { - // First-read error — polling continues; log so the operator has an event trail. - consecutiveFailures++; - HandlePollFailure(ex, consecutiveFailures, initial: true); - } - - while (!ct.IsCancellationRequested) - { - // Capped exponential backoff: Interval, 2×, 4×, ... up to PollBackoffCap. Healthy - // ticks reset consecutiveFailures back to 0 so the cadence snaps back to Interval. - var delay = ComputeBackoffDelay(state.Interval, consecutiveFailures); - try { await Task.Delay(delay, ct).ConfigureAwait(false); } - // Task.Delay can only observe ct-triggered OCE, so this filter is purely for uniformity. - catch (OperationCanceledException) when (ct.IsCancellationRequested) { return; } - - try - { - await PollOnceAsync(state, forceRaise: false, ct).ConfigureAwait(false); - consecutiveFailures = 0; - } - // Only teardown exits; any non-teardown OCE (e.g. a slipped-through connect timeout) - // backs off instead of killing the loop (STAB-14 defense-in-depth). - catch (OperationCanceledException) when (ct.IsCancellationRequested) { return; } - catch (Exception ex) - { - // Sustained polling error — loop continues with backoff; log + update health. - consecutiveFailures++; - HandlePollFailure(ex, consecutiveFailures, initial: false); - } - } - } - - /// - /// Logs the swallowed poll exception and, once - /// consecutive failures have accumulated, degrades the driver health so the failure - /// surfaces on the dashboard. The probe loop owns Running/Stopped - /// transitions for the host-connectivity surface, so we touch - /// rather than the probe state. - /// - private void HandlePollFailure(Exception ex, int consecutiveFailures, bool initial) - { - if (initial) - _logger.LogWarning(ex, "S7 poll initial-read failed. Driver={DriverInstanceId} ConsecutiveFailures={Count}", - _driverInstanceId, consecutiveFailures); - else - _logger.LogWarning(ex, "S7 poll tick failed. Driver={DriverInstanceId} ConsecutiveFailures={Count}", - _driverInstanceId, consecutiveFailures); - - if (consecutiveFailures >= PollFailureHealthThreshold) - { - // Don't downgrade a Faulted state (e.g. PUT/GET-denied set by ReadAsync) — Faulted - // is a stronger signal than Degraded and is reserved for permanent config faults. - if (_health.State != DriverState.Faulted) - _health = new DriverHealth(DriverState.Degraded, _health.LastSuccessfulRead, ex.Message); - } - } - - /// - /// Capped exponential backoff. consecutiveFailures == 0 returns the configured - /// ; each subsequent failure doubles the wait up to - /// . Computed in ticks to avoid overflow at large counts. - /// - /// Base polling interval. - /// Number of consecutive failures. - /// The computed backoff delay. - internal static TimeSpan ComputeBackoffDelay(TimeSpan interval, int consecutiveFailures) - { - if (consecutiveFailures <= 0) return interval; - // Cap the shift to avoid overflow — at 30 the result already saturates PollBackoffCap - // for any reasonable Interval. - var shift = Math.Min(consecutiveFailures - 1, 30); - var ticks = interval.Ticks << shift; - if (ticks <= 0 || ticks > PollBackoffCap.Ticks) return PollBackoffCap; - return TimeSpan.FromTicks(ticks); - } - - private async Task PollOnceAsync(SubscriptionState state, bool forceRaise, CancellationToken ct) - { -#pragma warning disable OTOPCUA0001 // Driver-INTERNAL self-call: S7Driver's own poll loop reading via its own ReadAsync; CapabilityInvoker wraps the driver from the dispatch layer OUTWARD, so the driver's internal poll must NOT re-wrap. Intentional, not a dispatch gap. - var snapshots = await ReadAsync(state.TagReferences, ct).ConfigureAwait(false); -#pragma warning restore OTOPCUA0001 - 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 || !Equals(lastSeen?.Value, current.Value) || lastSeen?.StatusCode != current.StatusCode) - { - state.LastValues[tagRef] = current; - OnDataChange?.Invoke(this, new DataChangeEventArgs(state.Handle, tagRef, current)); - } - } - } - - private sealed record SubscriptionState( - S7SubscriptionHandle Handle, - IReadOnlyList TagReferences, - TimeSpan Interval, - CancellationTokenSource Cts) - { - /// Gets the last known values for subscribed tags. - public ConcurrentDictionary LastValues { get; } - = new(StringComparer.OrdinalIgnoreCase); - - /// - /// Handle to this subscription's poll loop. Tracked so - /// can await it after cancelling. - /// - public Task PollTask { get; set; } = Task.CompletedTask; - } - - private sealed record S7SubscriptionHandle(long Id) : ISubscriptionHandle - { - /// - public string DiagnosticId => $"s7-sub-{Id}"; + _logger.LogWarning(ex, "S7 poll reader failed. Driver={DriverInstanceId}", _driverInstanceId); + if (_health.State != DriverState.Faulted) + _health = new DriverHealth(DriverState.Degraded, _health.LastSuccessfulRead, ex.Message); } // ---- IHostConnectivityProbe ---- @@ -1693,34 +1542,22 @@ public sealed class S7Driver /// private void SynchronousTeardown() { - var drain = new List(); + // Drain the polling overlay (drain-before-dispose, STAB-6-S7). + try { _poll.DisposeAsync().AsTask().Wait(DrainTimeout); } + catch { /* timeouts/loop faults are tolerated — teardown continues */ } var probeCts = _probeCts; var probeTask = _probeTask; try { probeCts?.Cancel(); } catch { } - if (probeTask is not null) drain.Add(probeTask); - - var subscriptions = _subscriptions.Values.ToArray(); - _subscriptions.Clear(); - foreach (var state in subscriptions) + if (probeTask is not null) { - try { state.Cts.Cancel(); } catch { } - drain.Add(state.PollTask); - } - - if (drain.Count > 0) - { - try { Task.WhenAll(drain).Wait(DrainTimeout); } + try { probeTask.Wait(DrainTimeout); } catch { /* timeouts/loop faults are tolerated — teardown continues */ } } probeCts?.Dispose(); _probeCts = null; _probeTask = null; - foreach (var state in subscriptions) - { - try { state.Cts.Dispose(); } catch { } - } _initialized = false; _plcDead = false; diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.S7.Tests/S7DiscoveryAndSubscribeTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.S7.Tests/S7DiscoveryAndSubscribeTests.cs index 76bfe731..c3195036 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.S7.Tests/S7DiscoveryAndSubscribeTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.S7.Tests/S7DiscoveryAndSubscribeTests.cs @@ -91,8 +91,10 @@ public sealed class S7DiscoveryAndSubscribeTests var h1 = await drv.SubscribeAsync(["T1"], TimeSpan.FromMilliseconds(200), TestContext.Current.CancellationToken); var h2 = await drv.SubscribeAsync(["T2"], TimeSpan.FromMilliseconds(200), TestContext.Current.CancellationToken); - h1.DiagnosticId.ShouldStartWith("s7-sub-"); - h2.DiagnosticId.ShouldStartWith("s7-sub-"); + // Post poll-fork retirement (CONV-1) the subscription handle is the shared engine's + // PollSubscriptionHandle (`poll-sub-`), not the retired S7SubscriptionHandle (`s7-sub-`). + h1.DiagnosticId.ShouldStartWith("poll-sub-"); + h2.DiagnosticId.ShouldStartWith("poll-sub-"); h1.DiagnosticId.ShouldNotBe(h2.DiagnosticId); await drv.UnsubscribeAsync(h1, TestContext.Current.CancellationToken);