fix(runtime): keep native-alarm feed across data re-subscribes; reconcile retries alarms (#662)

A data re-subscribe (every deploy's bulk SetDesiredSubscriptions push to a kept driver) dropped the prior data subscription through the full DetachSubscription, which also unhooked OnAlarmEvent and forgot the alarm handle. On any deploy that did not respawn the driver, the alarm subscription kept logging "established" while every event reached nobody until a reconnect or respawn. The data path now uses DetachDataSubscription; only a lost connection and PostStop detach both. The #488 health-poll reconcile only looked at data, so an alarm subscribe that failed on (re)connect gated the feed for the life of the connection. ReconcileAlarmSubscription now re-attaches a missing handler and re-subscribes a stale/missing alarm handle every tick. OpcUaClient removes a half-built alarm subscription from its session when SubscribeAlarmsAsync throws, so the new retries cannot accumulate. Tests: actor-level data-resubscribe + reconcile-retry tests and an end-to-end NativeAlarmFeedSurvivesDeployTests through DriverHostActor; all three fail against the unfixed actor. Claude-Session: https://claude.ai/code/session_01Ef2ozxS7QPanGT8uoATbTe
2026-10-09 07:30:25 -04:00
parent ea220363f3
commit 5067509dd4
7 changed files with 295 additions and 21 deletions
@@ -218,6 +218,82 @@ public sealed class DriverInstanceActorNativeAlarmTests : RuntimeActorTestBase
        parent.ExpectNoMsg(TimeSpan.FromMilliseconds(100));
    }

+    /// <summary>
+    ///     Gitea #662 case 1 — re-subscribing a Connected driver's DATA (what every deploy's
+    ///     <c>SetDesiredSubscriptions</c> push does) must leave the native-alarm feed alone. The data
+    ///     re-subscribe drops the prior data subscription first, and that teardown used to run the full
+    ///     detach — unhooking <c>OnAlarmEvent</c> and forgetting the alarm handle — so the alarm subscription
+    ///     stayed "established" while every event it carried reached nobody until a reconnect.
+    /// </summary>
+    [Fact]
+    public async Task A_data_re_subscribe_keeps_the_native_alarm_feed_attached()
+    {
+        var driver = new AlarmSourceStubDriver();
+        var parent = CreateTestProbe();
+        parent.IgnoreMessages(m => m is DriverInstanceActor.ConnectivityChanged);
+        var actor = parent.ChildActorOf(DriverInstanceActor.Props(driver));
+
+        actor.Tell(new DriverInstanceActor.InitializeRequested("{}"));
+        AwaitCondition(() => driver.AlarmSubscriberCount == 1, TimeSpan.FromSeconds(2));
+        actor.Tell(new DriverInstanceActor.SetDesiredSubscriptions(
+            new[] { "tag-a" }, TimeSpan.FromMilliseconds(100), new[] { "Temp.HiHi" }));
+        AwaitCondition(() => driver.SubscribeAlarmsCallCount == 1, TimeSpan.FromSeconds(2));
+
+        // A second data subscribe while the first handle is live — the drop-then-resubscribe path.
+        await actor.Ask<DriverInstanceActor.SubscriptionEstablished>(
+            new DriverInstanceActor.Subscribe(new[] { "tag-a", "tag-b" }, TimeSpan.FromMilliseconds(100)),
+            TimeSpan.FromSeconds(3));
+
+        driver.AlarmSubscriberCount.ShouldBe(1, "the data re-subscribe must not unhook OnAlarmEvent");
+        driver.SubscribeAlarmsCallCount.ShouldBe(1,
+            "the alarm handle is still live, so nothing should re-issue (and duplicate) the driver-side subscription");
+
+        driver.RaiseAlarm(new AlarmEventArgs(
+            new StubAlarmHandle(), "Temp.HiHi", "Temp.HiHi", "T", "after re-subscribe", AlarmSeverity.High,
+            DateTime.UtcNow, Kind: AlarmTransitionKind.Raise));
+        parent.FishForMessage<DriverInstanceActor.AttributeAlarmPublished>(_ => true, TimeSpan.FromSeconds(2))
+            .Args.ConditionId.ShouldBe("Temp.HiHi");
+    }
+
+    /// <summary>
+    ///     Gitea #662 case 2 — an alarm subscribe that fails on (re)connect (a session not yet ready after an
+    ///     outage) is retried by the periodic reconcile while the driver stays Connected. Before #662 the
+    ///     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()
+    {
+        var driver = new AlarmSourceStubDriver { AlarmSubscribeFailuresBeforeSuccess = 1 };
+        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(
+            Array.Empty<string>(), TimeSpan.FromMilliseconds(100), new[] { "Temp.HiHi" }));
+        actor.Tell(new DriverInstanceActor.InitializeRequested("{}"));
+
+        AwaitCondition(() => driver.SubscribeAlarmsCallCount >= 1, PresenceBudget);
+        AwaitCondition(() => driver.SubscribeAlarmsCallCount >= 2, PresenceBudget);
+        driver.LastAlarmRefs.ShouldBe(new[] { "Temp.HiHi" });
+
+        driver.RaiseAlarm(new AlarmEventArgs(
+            new StubAlarmHandle(), "Temp.HiHi", "Temp.HiHi", "T", "after retry", AlarmSeverity.High,
+            DateTime.UtcNow, Kind: AlarmTransitionKind.Raise));
+        parent.FishForMessage<DriverInstanceActor.AttributeAlarmPublished>(_ => true, TimeSpan.FromSeconds(2))
+            .Args.ConditionId.ShouldBe("Temp.HiHi");
+
+        // 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".
+        var settled = driver.SubscribeAlarmsCallCount;
+        var before = publisher.Count;
+        AwaitCondition(() => publisher.Count >= before + 4, PresenceBudget);
+        driver.SubscribeAlarmsCallCount.ShouldBe(settled, "a live, matching alarm handle must not be re-subscribed every tick");
+    }
+
    /// <summary>
    ///     After a full reconnect cycle (Connected → Reconnecting → Connected), a single raised alarm still
    ///     yields EXACTLY ONE <see cref="DriverInstanceActor.AttributeAlarmPublished"/> — the
@@ -354,10 +430,16 @@ public sealed class DriverInstanceActorNativeAlarmTests : RuntimeActorTestBase
            IReadOnlyList<string> sourceNodeIds, CancellationToken cancellationToken)
        {
            LastAlarmRefs = sourceNodeIds;
-            Interlocked.Increment(ref _subscribeAlarmsCallCount);
+            var call = Interlocked.Increment(ref _subscribeAlarmsCallCount);
+            if (call <= AlarmSubscribeFailuresBeforeSuccess)
+                throw new InvalidOperationException($"alarm subscribe failure {call} (test-injected)");
            return Task.FromResult<IAlarmSubscriptionHandle>(new StubAlarmHandle());
        }

+        /// <summary>How many leading <see cref="SubscribeAlarmsAsync"/> calls throw — the "session not ready
+        /// after an outage" shape from #662.</summary>
+        public int AlarmSubscribeFailuresBeforeSuccess { get; init; }
+
        /// <summary>Cancels an alarm subscription.</summary>
        public Task UnsubscribeAlarmsAsync(IAlarmSubscriptionHandle handle, CancellationToken cancellationToken) =>
            Task.CompletedTask;
@@ -416,6 +498,28 @@ public sealed class DriverInstanceActorNativeAlarmTests : RuntimeActorTestBase
        }
    }

+    /// <summary>Counts health-snapshot publishes — one per processed <c>health-poll</c> tick.</summary>
+    private sealed class CountingHealthPublisher : IDriverHealthPublisher
+    {
+        private int _count;
+
+        /// <summary>Number of snapshots published so far.</summary>
+        public int Count => Volatile.Read(ref _count);
+
+        /// <inheritdoc />
+        public void Publish(
+            string clusterId,
+            string driverInstanceId,
+            DriverHealth health,
+            int errorCount5Min,
+            DateTime? rediscoveryNeededUtc = null,
+            string? rediscoveryReason = null,
+            IReadOnlyList<HostConnectivityStatus>? hostStatuses = null,
+            SkippedTagReport? skippedTags = null,
+            IReadOnlyDictionary<string, string>? diagnostics = null)
+            => Interlocked.Increment(ref _count);
+    }
+
    /// <summary>Minimal <see cref="IAlarmSubscriptionHandle"/> for building <see cref="AlarmEventArgs"/>.</summary>
    private sealed class StubAlarmHandle : IAlarmSubscriptionHandle
    {
@@ -0,0 +1,87 @@
+using Akka.Actor;
+using Shouldly;
+using Xunit;
+using ZB.MOM.WW.OtOpcUa.Commons.Messages.Deploy;
+using ZB.MOM.WW.OtOpcUa.Commons.Types;
+using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
+using ZB.MOM.WW.OtOpcUa.Runtime.OpcUa;
+using ZB.MOM.WW.OtOpcUa.Runtime.Tests.Harness;
+
+namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.Drivers;
+
+/// <summary>
+///     Gitea #662 — a native-alarm feed must survive a deploy that does NOT respawn its driver, end to end
+///     through the real <see cref="Runtime.Drivers.DriverHostActor"/> and its real
+///     <see cref="Runtime.Drivers.DriverInstanceActor"/> child.
+/// </summary>
+/// <remarks>
+///     Found live on the docker-dev rig: a deploy that only disabled ANOTHER driver re-pushed the desired set to
+///     an unchanged OpcUaClient driver, whose data re-subscribe tore down the alarm event handler too. The
+///     condition then froze while upstream kept emitting, and only a respawn brought it back. Every existing
+///     end-to-end alarm test drives the driver directly (the host's driver is <c>Enabled=false</c>), so none of
+///     them could see the instance actor drop the feed.
+/// </remarks>
+public sealed class NativeAlarmFeedSurvivesDeployTests : RuntimeActorTestBase
+{
+    private static readonly RevisionHash RevA = RevisionHash.Parse(new string('a', 64));
+    private static readonly RevisionHash RevB = RevisionHash.Parse(new string('b', 64));
+    private static readonly NodeId TestNode = NodeId.Parse("node-1:4053");
+
+    private const string DriverId = "DI-1";
+    private const string AlarmXPath = "Plant/Modbus/dev1/alarmX";
+    private const string AlarmConfig = """{"alarm":{"alarmType":"AlarmCondition","severity":500}}""";
+
+    private static readonly (string Driver, string DriverName, string Device, string Tag, string DataType, bool Writable)[]
+        RawTags =
+        [
+            (DriverId, "Modbus", "dev1", "speed", "Float", false),
+            (DriverId, "Modbus", "dev1", "alarmX", "Float", false),
+        ];
+
+    private static readonly Dictionary<string, string> TagConfigs = new() { ["alarmX"] = AlarmConfig };
+
+    [Fact]
+    public void An_upstream_alarm_still_reaches_its_condition_after_a_non_respawning_deploy()
+    {
+        var db = NewInMemoryDbFactory();
+        var factory = new RecordingSubscribableDriverFactory("Modbus");
+        var revA = V3ArtifactSeeder.SeedV3Deployment(
+            db, RevA, RawTags, Array.Empty<(string, string, string, string, string?)>(),
+            enabled: true, tagConfigByTag: TagConfigs);
+        var (host, publish, _) = V3ArtifactSeeder.SpawnHostAndApply(this, db, TestNode, revA, RevA, factory);
+
+        RecordingSubscribableDriver driver = null!;
+        AwaitAssert(() =>
+        {
+            factory.Created.Count.ShouldBe(1);
+            driver = factory.Created[0];
+            driver.SubscribeCount.ShouldBeGreaterThanOrEqualTo(1);
+            driver.LastSubscribedAlarmRefs.ShouldBe(new[] { AlarmXPath });
+            driver.AlarmSubscriberCount.ShouldBe(1);
+        }, duration: PresenceBudget);
+        var dataSubscribesBefore = driver.SubscribeCount;
+
+        // Revision B only ADDS another driver: DI-1's merged DriverConfig is byte-identical, so it is kept,
+        // and the bulk push re-sends its (unchanged) desired set — which re-subscribes its data.
+        var revB = V3ArtifactSeeder.SeedV3Deployment(
+            db, RevB, RawTags, Array.Empty<(string, string, string, string, string?)>(),
+            enabled: true, tagConfigByTag: TagConfigs, taglessDrivers: ["DI-2"]);
+        host.Tell(new DispatchDeployment(revB, RevB, CorrelationId.NewId()));
+
+        AwaitAssert(() =>
+        {
+            driver.SubscribeCount.ShouldBeGreaterThan(dataSubscribesBefore,
+                "precondition: the deploy must re-subscribe DI-1's data, or this test exercises nothing");
+        }, duration: PresenceBudget);
+        driver.InitializeCount.ShouldBe(1, "precondition: DI-1 must NOT have been respawned or re-initialised");
+        driver.ShutdownCount.ShouldBe(0);
+        factory.Created[0].ShouldBeSameAs(driver);
+
+        driver.AlarmSubscriberCount.ShouldBe(1, "the alarm event handler must survive the data re-subscribe");
+        driver.FireAlarm(AlarmXPath, AlarmTransitionKind.Raise);
+
+        var update = publish.FishForMessage<OpcUaPublishActor.AlarmStateUpdate>(
+            m => m.AlarmNodeId == AlarmXPath && m.State.Active, PresenceBudget);
+        update.State.Acknowledged.ShouldBeFalse();
+    }
+}
@@ -343,9 +343,15 @@ internal sealed class RecordingSubscribableDriver : IDriver, ISubscribable, IAla

    /// <summary>Raises <see cref="OnAlarmEvent"/>; present so the event is not merely written-to-never-read.</summary>
    /// <param name="conditionId">The condition id to raise.</param>
-    public void FireAlarm(string conditionId) =>
+    /// <param name="kind">The transition kind to report.</param>
+    public void FireAlarm(string conditionId, AlarmTransitionKind kind = AlarmTransitionKind.Unspecified) =>
        OnAlarmEvent?.Invoke(this, new AlarmEventArgs(
-            _alarmHandle, conditionId, conditionId, "AlarmCondition", "test", AlarmSeverity.High, DateTime.UtcNow));
+            _alarmHandle, conditionId, conditionId, "AlarmCondition", "test", AlarmSeverity.High, DateTime.UtcNow,
+            Kind: kind));
+
+    /// <summary>Live subscribers on <see cref="OnAlarmEvent"/> — an alarm subscription with no handler
+    /// attached delivers nothing (#662), so this is the other half of "the feed is up".</summary>
+    public int AlarmSubscriberCount => OnAlarmEvent?.GetInvocationList().Length ?? 0;

    private sealed class RecordingAlarmHandle : IAlarmSubscriptionHandle
    {