fix(runtime): in-place ApplyDelta releases and re-subscribes data + alarms (#663)

A Connected ApplyDelta called ReinitializeAsync and kept the actor's data and native-alarm handles, though every driver's reinit drops the driver-side subscriptions behind them (#630). The reconcile only re-subscribes a missing handle, so the driver went silent until a reconnect. Release both at the driver before the reinit and re-subscribe both in a finally. Unreachable from deploys since #516; pinned by a two-case test, each case failing against a different wrong fix. Claude-Session: https://claude.ai/code/session_01Ef2ozxS7QPanGT8uoATbTe
2026-10-09 07:49:56 -04:00
parent 83d129d1bd
commit 39c5af3d7f
4 changed files with 158 additions and 8 deletions
@@ -369,6 +369,56 @@ public sealed class DriverInstanceActorNativeAlarmTests : RuntimeActorTestBase
            .Connected.ShouldBeTrue();
    }

+    /// <summary>
+    ///     Gitea #663 — an in-place config change on a Connected driver (<see cref="DriverInstanceActor.ApplyDelta"/>
+    ///     → <see cref="IDriver.ReinitializeAsync"/>) must end with exactly one live data subscription and one live
+    ///     native-alarm subscription at the driver, and the alarm feed still forwarding. The handler used to keep
+    ///     the actor's handles across the reinit while the driver had dropped the subscriptions behind them, so
+    ///     the reconcile (which only re-subscribes a MISSING handle) never restored them and the driver went silent.
+    /// </summary>
+    /// <param name="reinitKeepsSubscriptions">false = the production shape (reinit drops everything, #630) —
+    /// the regression; true = a driver that keeps them, which pins that the fix releases before re-subscribing
+    /// instead of piling a second subscription on top.</param>
+    [Theory]
+    [InlineData(false)]
+    [InlineData(true)]
+    public void An_in_place_ApplyDelta_leaves_one_live_data_and_one_live_alarm_subscription(bool reinitKeepsSubscriptions)
+    {
+        var driver = new AlarmSourceStubDriver { ReinitKeepsSubscriptions = reinitKeepsSubscriptions };
+        var publisher = new CountingHealthPublisher();
+        var parent = CreateTestProbe();
+        parent.IgnoreMessages(m => m is DriverInstanceActor.ConnectivityChanged);
+        var actor = parent.ChildActorOf(DriverInstanceActor.Props(
+            driver, healthPublisher: publisher, healthPollInterval: TimeSpan.FromMilliseconds(100)));
+
+        actor.Tell(new DriverInstanceActor.SetDesiredSubscriptions(
+            new[] { "tag-a" }, TimeSpan.FromMilliseconds(100), new[] { "Temp.HiHi" }));
+        actor.Tell(new DriverInstanceActor.InitializeRequested("{}"));
+        AwaitCondition(() => driver.LiveDataSubscriptions == 1 && driver.LiveAlarmSubscriptions == 1, PresenceBudget);
+
+        var correlation = CorrelationId.NewId();
+        actor.Tell(new DriverInstanceActor.ApplyDelta("{\"changed\":true}", correlation), parent.Ref);
+        parent.FishForMessage<DriverInstanceActor.ApplyResult>(r => r.Correlation == correlation, PresenceBudget)
+            .Success.ShouldBeTrue();
+        driver.ReinitializeCount.ShouldBe(1);
+        driver.InitializeCount.ShouldBe(1, "an in-place delta must not go through the reconnect path");
+
+        AwaitCondition(() => driver.LiveDataSubscriptions == 1 && driver.LiveAlarmSubscriptions == 1, PresenceBudget);
+        // The "not two" half is an absence, so wait for three more health-poll ticks — each runs the reconcile
+        // that would issue a duplicate — before re-checking.
+        var ticks = publisher.Count;
+        AwaitCondition(() => publisher.Count >= ticks + 3, PresenceBudget);
+        driver.LiveDataSubscriptions.ShouldBe(1);
+        driver.LiveAlarmSubscriptions.ShouldBe(1);
+        driver.AlarmSubscriberCount.ShouldBe(1);
+
+        driver.RaiseAlarm(new AlarmEventArgs(
+            new StubAlarmHandle(), "Temp.HiHi", "Temp.HiHi", "T", "after reinit", AlarmSeverity.High,
+            DateTime.UtcNow, Kind: AlarmTransitionKind.Raise));
+        parent.FishForMessage<DriverInstanceActor.AttributeAlarmPublished>(_ => true, PresenceBudget)
+            .Args.ConditionId.ShouldBe("Temp.HiHi");
+    }
+
    // --- stub drivers ----------------------------------------------------------------------------

    private class StubDriver : IDriver
@@ -389,7 +439,7 @@ public sealed class DriverInstanceActorNativeAlarmTests : RuntimeActorTestBase
        }

        /// <summary>Reinitializes the driver.</summary>
-        public Task ReinitializeAsync(string driverConfigJson, CancellationToken cancellationToken) =>
+        public virtual Task ReinitializeAsync(string driverConfigJson, CancellationToken cancellationToken) =>
            Task.CompletedTask;

        /// <summary>Shuts down the driver.</summary>
@@ -413,14 +463,55 @@ public sealed class DriverInstanceActorNativeAlarmTests : RuntimeActorTestBase
        /// <summary>Number of live subscribers on <see cref="OnAlarmEvent"/> — proves attach/detach.</summary>
        public int AlarmSubscriberCount => OnAlarmEvent?.GetInvocationList().Length ?? 0;

+        private readonly object _liveLock = new();
+        private readonly HashSet<object> _liveData = new(ReferenceEqualityComparer.Instance);
+        private readonly HashSet<object> _liveAlarms = new(ReferenceEqualityComparer.Instance);
+        private int _reinitializeCount;
+
+        /// <summary>Driver-side data subscriptions currently open (subscribed and not yet released).</summary>
+        public int LiveDataSubscriptions { get { lock (_liveLock) return _liveData.Count; } }
+
+        /// <summary>Driver-side native-alarm subscriptions currently open.</summary>
+        public int LiveAlarmSubscriptions { get { lock (_liveLock) return _liveAlarms.Count; } }
+
+        /// <summary>Number of <see cref="ReinitializeAsync"/> calls.</summary>
+        public int ReinitializeCount => Volatile.Read(ref _reinitializeCount);
+
+        /// <summary>When false (the default, and every production driver since #630), a reinit releases the
+        /// device/session state and with it every driver-side subscription; when true it keeps them — the
+        /// shape that would double-subscribe if the actor re-subscribed without releasing first.</summary>
+        public bool ReinitKeepsSubscriptions { get; init; }
+
+        /// <inheritdoc />
+        public override Task ReinitializeAsync(string driverConfigJson, CancellationToken cancellationToken)
+        {
+            Interlocked.Increment(ref _reinitializeCount);
+            if (!ReinitKeepsSubscriptions)
+            {
+                lock (_liveLock)
+                {
+                    _liveData.Clear();
+                    _liveAlarms.Clear();
+                }
+            }
+            return Task.CompletedTask;
+        }
+
        /// <summary>Subscribes to the specified full references.</summary>
        public Task<ISubscriptionHandle> SubscribeAsync(
-            IReadOnlyList<string> fullReferences, TimeSpan publishingInterval, CancellationToken cancellationToken) =>
-            Task.FromResult<ISubscriptionHandle>(_subHandle);
+            IReadOnlyList<string> fullReferences, TimeSpan publishingInterval, CancellationToken cancellationToken)
+        {
+            var handle = new StubHandle();
+            lock (_liveLock) _liveData.Add(handle);
+            return Task.FromResult<ISubscriptionHandle>(handle);
+        }

        /// <summary>Unsubscribes from the specified subscription handle.</summary>
-        public Task UnsubscribeAsync(ISubscriptionHandle handle, CancellationToken cancellationToken) =>
-            Task.CompletedTask;
+        public Task UnsubscribeAsync(ISubscriptionHandle handle, CancellationToken cancellationToken)
+        {
+            lock (_liveLock) _liveData.Remove(handle);
+            return Task.CompletedTask;
+        }

        private int _subscribeAlarmsCallCount;

@@ -439,7 +530,9 @@ public sealed class DriverInstanceActorNativeAlarmTests : RuntimeActorTestBase
            var call = Interlocked.Increment(ref _subscribeAlarmsCallCount);
            if (call <= AlarmSubscribeFailuresBeforeSuccess)
                throw new InvalidOperationException($"alarm subscribe failure {call} (test-injected)");
-            return Task.FromResult<IAlarmSubscriptionHandle>(new StubAlarmHandle());
+            var handle = new StubAlarmHandle();
+            lock (_liveLock) _liveAlarms.Add(handle);
+            return Task.FromResult<IAlarmSubscriptionHandle>(handle);
        }

        /// <summary>How many leading <see cref="SubscribeAlarmsAsync"/> calls throw — the "session not ready
@@ -447,8 +540,11 @@ public sealed class DriverInstanceActorNativeAlarmTests : RuntimeActorTestBase
        public int AlarmSubscribeFailuresBeforeSuccess { get; init; }

        /// <summary>Cancels an alarm subscription.</summary>
-        public Task UnsubscribeAlarmsAsync(IAlarmSubscriptionHandle handle, CancellationToken cancellationToken) =>
-            Task.CompletedTask;
+        public Task UnsubscribeAlarmsAsync(IAlarmSubscriptionHandle handle, CancellationToken cancellationToken)
+        {
+            lock (_liveLock) _liveAlarms.Remove(handle);
+            return Task.CompletedTask;
+        }

        /// <summary>Acknowledges a batch of alarms.</summary>
        public Task AcknowledgeAsync(