fix(#662 review): back off alarm retries, contain failed OpcUaClient alarm subscribes

- Reconcile retries of a failing alarm subscribe back off (skip 2^(n-1)-1 ticks, cap 7 => >= every 4 min at the 30 s tick) so a hanging subscribe cannot hold the mailbox every tick; the stack is logged once per streak. - The unreachable handler re-attach logs at Warning (it only fires on a regression). - OpcUaClient: a failed SubscribeAlarmsAsync detaches its notification handler and bounds the session cleanup to 5 s under _gate; OnEventNotification drops events from subscriptions not registered in _alarmSubscriptions, so a stray that survives cleanup cannot deliver. - The retry test also covers 3 failures (crosses the backoff); its tick positive control documents its dependence on the stub's changing health. Claude-Session: https://claude.ai/code/session_01Ef2ozxS7QPanGT8uoATbTe
2026-10-09 07:35:50 -04:00
parent 5067509dd4
commit 28ca04e466
3 changed files with 75 additions and 17 deletions
@@ -1701,6 +1701,10 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
            AttributeId = Attributes.Value,
        });

+        // Set by CreateAlarmEventItemAsync so a failure after the handler is attached can detach it (#662).
+        MonitoredItem? eventItem = null;
+        MonitoredItemNotificationEventHandler? notifHandler = null;
+
        await _gate.WaitAsync(cancellationToken).ConfigureAwait(false);
        try
        {
@@ -1716,9 +1720,20 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
            }
            catch
            {
-                // #662: the host now retries a failed alarm subscribe on every reconcile tick, so a half-built
-                // subscription left on the session would accumulate one per retry for the length of an outage.
-                try { await session.RemoveSubscriptionAsync(subscription, CancellationToken.None).ConfigureAwait(false); }
+                // #662: the host retries a failed alarm subscribe from its reconcile, so a half-built subscription
+                // left on the session would accumulate one per retry for the length of an outage. Detach the
+                // handler first — the remove is best-effort and can itself fail against a broken server — and
+                // bound it, since _gate is held. OnEventNotification also drops anything from an unregistered id.
+                if (eventItem is not null && notifHandler is not null)
+                {
+                    try { eventItem.Notification -= notifHandler; }
+                    catch { /* best-effort */ }
+                }
+                try
+                {
+                    using var removeCts = new CancellationTokenSource(TimeSpan.FromSeconds(5));
+                    await session.RemoveSubscriptionAsync(subscription, removeCts.Token).ConfigureAwait(false);
+                }
                catch { /* best-effort — the session may be the thing that failed */ }
                throw;
            }
@@ -1736,7 +1751,7 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
        {
            await subscription.CreateAsync(cancellationToken).ConfigureAwait(false);

-            var eventItem = new MonitoredItem(telemetry: null!, new MonitoredItemOptions
+            eventItem = new MonitoredItem(telemetry: null!, new MonitoredItemOptions
            {
                DisplayName = "Server/Events",
                StartNodeId = ObjectIds.Server,
@@ -1753,7 +1768,7 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
            // can `-=` it later. The lambda captures `handle` and `sourceFilter`, so without
            // the explicit detach the SDK's invocation list keeps the driver instance alive
            // until the session itself is disposed.
-            MonitoredItemNotificationEventHandler notifHandler = (mi, args) =>
+            notifHandler = (mi, args) =>
                OnEventNotification(handle, sourceFilter, mi, args);
            eventItem.Notification += notifHandler;
            subscription.AddItem(eventItem);
@@ -1898,6 +1913,10 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
        MonitoredItemNotificationEventArgs args)
    {
        if (args.NotificationValue is not EventFieldList efl) return;
+        // #662 review: only a subscription SubscribeAlarmsAsync completed (and nothing has since released) may
+        // deliver. A half-built one whose cleanup failed against a broken server can survive on the session and
+        // be transferred by an SDK reconnect; without this it would feed RouteAlarmEvent alongside the real one.
+        if (!_alarmSubscriptions.ContainsKey(handle.Id)) return;
        RouteAlarmEvent(efl.EventFields, sourceFilter, handle);
    }

@@ -281,6 +281,17 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
    /// newly-added alarm for the life of the connection. Empty whenever no handle is held.</para></summary>
    private IReadOnlyList<string> _subscribedAlarmRefs = Array.Empty<string>();

+    /// <summary>#662 — consecutive <see cref="HandleSubscribeAlarmsAsync"/> failures; reset on success and on
+    /// <see cref="DetachAlarmSource"/> (a new connection starts a new streak).</summary>
+    private int _alarmSubscribeFailures;
+
+    /// <summary>#662 — health-poll ticks the alarm reconcile still skips before its next retry (backoff).</summary>
+    private int _alarmRetrySkipTicks;
+
+    /// <summary>#662 — backoff cap: at the production 30 s tick a failing alarm subscribe is retried at least
+    /// every 4 minutes.</summary>
+    private const int MaxAlarmRetrySkipTicks = 7;
+
    /// <summary>
    /// Gets or sets the timer scheduler for scheduling reconnection attempts.
    /// </summary>
@@ -707,18 +718,27 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
    ///     its data streaming. Re-asserts both halves the feed needs: the <c>OnAlarmEvent</c> handler
    ///     (idempotent attach) and a subscription matching the desired set (<see cref="SubscribeDesiredAlarms"/>
    ///     is a no-op while the live handle already matches). Gated on <see cref="IAlarmSource"/> + a
-    ///     non-empty desired set, so it costs nothing on a driver with no alarm tags; a persistently failing
-    ///     subscribe retries once per tick, the same cadence as the data reconcile.
+    ///     non-empty desired set, so it costs nothing on a driver with no alarm tags.
+    ///     <para>A persistently failing subscribe backs off: after the n-th consecutive failure the reconcile
+    ///     skips 2^(n-1) − 1 ticks, capped at <see cref="MaxAlarmRetrySkipTicks"/>, because a HANGING subscribe
+    ///     holds the mailbox for its full 10 s bound (writes queued behind it can miss the host's 8 s Ask). A
+    ///     Connected entry or a deploy's desired-set push still retries immediately — only the timer backs off.</para>
    /// </summary>
    private void ReconcileAlarmSubscription()
    {
        if (_driver is not IAlarmSource || _desiredAlarmRefs.Count == 0) return;
        if (_alarmEventHandler is null)
        {
-            _log.Info("DriverInstance {Id}: connected with desired alarm refs but no native-alarm handler — re-attaching (periodic reconcile)",
+            // Unreachable while every Connected entry attaches — so if this ever logs, something regressed #662.
+            _log.Warning("DriverInstance {Id}: connected with desired alarm refs but no native-alarm handler — re-attaching (periodic reconcile)",
                _driverInstanceId);
            AttachAlarmSource();
        }
+        if (_alarmRetrySkipTicks > 0)
+        {
+            _alarmRetrySkipTicks--;
+            return;
+        }
        if (AlarmSubscriptionIsStale())
        {
            _log.Info(
@@ -1249,6 +1269,8 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
        // across reconnects. The gate (Count > 0) and the one-shot fan-out are unchanged.
        _alarmSubscriptionHandle = null;
        _subscribedAlarmRefs = Array.Empty<string>();
+        _alarmSubscribeFailures = 0;
+        _alarmRetrySkipTicks = 0;
    }

    /// <summary>Establish the native-alarm subscription that un-gates an <see cref="IAlarmSource"/> driver's
@@ -1314,13 +1336,24 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
                async ct => await src.SubscribeAlarmsAsync(refs, ct),
                cts.Token);
            _subscribedAlarmRefs = refs;
-            _log.Info("DriverInstance {Id}: native-alarm subscription established for {Count} alarm ref(s) ({Diag})",
-                _driverInstanceId, refs.Count, _alarmSubscriptionHandle.DiagnosticId);
+            _log.Info("DriverInstance {Id}: native-alarm subscription established for {Count} alarm ref(s) ({Diag}){After}",
+                _driverInstanceId, refs.Count, _alarmSubscriptionHandle.DiagnosticId,
+                _alarmSubscribeFailures > 0 ? $" after {_alarmSubscribeFailures} failed attempt(s)" : "");
+            _alarmSubscribeFailures = 0;
+            _alarmRetrySkipTicks = 0;
        }
        catch (Exception ex)
        {
-            _log.Warning(ex, "DriverInstance {Id}: native-alarm subscription failed — feed stays gated until the next reconcile tick retries it",
-                _driverInstanceId);
+            _alarmSubscribeFailures++;
+            _alarmRetrySkipTicks = Math.Min((1 << Math.Min(_alarmSubscribeFailures - 1, 30)) - 1, MaxAlarmRetrySkipTicks);
+            // The stack once per failure streak; the retries that follow repeat the same cause, so they log
+            // just the message rather than a full trace every tick for the length of an outage.
+            if (_alarmSubscribeFailures == 1)
+                _log.Warning(ex, "DriverInstance {Id}: native-alarm subscription failed — feed stays gated until the periodic reconcile retries it",
+                    _driverInstanceId);
+            else
+                _log.Warning("DriverInstance {Id}: native-alarm subscription retry {Attempt} failed ({Reason}) — next retry in {Ticks} health-poll tick(s)",
+                    _driverInstanceId, _alarmSubscribeFailures, ex.Message, _alarmRetrySkipTicks + 1);
        }
    }

@@ -261,10 +261,14 @@ public sealed class DriverInstanceActorNativeAlarmTests : RuntimeActorTestBase
    ///     reconcile looked at data only, so the feed stayed gated for the life of the connection. Nothing
    ///     else here can produce the second call: no reconnect, no deploy.
    /// </summary>
-    [Fact]
-    public void A_failed_alarm_subscribe_is_retried_by_the_periodic_reconcile()
+    /// <param name="failures">Leading failures; 3 crosses the backoff (skips 0, 1, then 3 ticks), proving the
+    /// retries keep coming rather than giving up.</param>
+    [Theory]
+    [InlineData(1)]
+    [InlineData(3)]
+    public void A_failed_alarm_subscribe_is_retried_by_the_periodic_reconcile(int failures)
    {
-        var driver = new AlarmSourceStubDriver { AlarmSubscribeFailuresBeforeSuccess = 1 };
+        var driver = new AlarmSourceStubDriver { AlarmSubscribeFailuresBeforeSuccess = failures };
        var publisher = new CountingHealthPublisher();
        var parent = CreateTestProbe();
        parent.IgnoreMessages(m => m is DriverInstanceActor.ConnectivityChanged);
@@ -276,7 +280,7 @@ public sealed class DriverInstanceActorNativeAlarmTests : RuntimeActorTestBase
        actor.Tell(new DriverInstanceActor.InitializeRequested("{}"));

        AwaitCondition(() => driver.SubscribeAlarmsCallCount >= 1, PresenceBudget);
-        AwaitCondition(() => driver.SubscribeAlarmsCallCount >= 2, PresenceBudget);
+        AwaitCondition(() => driver.SubscribeAlarmsCallCount >= failures + 1, PresenceBudget);
        driver.LastAlarmRefs.ShouldBe(new[] { "Temp.HiHi" });

        driver.RaiseAlarm(new AlarmEventArgs(
@@ -287,7 +291,9 @@ public sealed class DriverInstanceActorNativeAlarmTests : RuntimeActorTestBase

        // Converged: a live, matching handle is left alone on later ticks. POSITIVE CONTROL: each tick
        // publishes one health snapshot, so waiting for four more makes the unchanged count mean "the
-        // reconcile ran and declined", not "no tick had fired yet".
+        // reconcile ran and declined", not "no tick had fired yet". It relies on the base StubDriver's
+        // GetHealth() stamping UtcNow, which defeats the publish dedup so EVERY tick publishes — a stub with
+        // stable health would publish once and this wait would time out (loudly, not vacuously).
        var settled = driver.SubscribeAlarmsCallCount;
        var before = publisher.Count;
        AwaitCondition(() => publisher.Count >= before + 4, PresenceBudget);