From 5067509dd4b80037a51b7273e9747f4ba96434f1 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Fri, 9 Oct 2026 07:30:25 -0400 Subject: [PATCH 1/2] 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 --- CLAUDE.md | 5 + docs/DriverLifecycle.md | 13 +++ .../OpcUaClientDriver.cs | 32 ++++-- .../Drivers/DriverInstanceActor.cs | 63 +++++++++-- .../DriverInstanceActorNativeAlarmTests.cs | 106 +++++++++++++++++- .../NativeAlarmFeedSurvivesDeployTests.cs | 87 ++++++++++++++ .../Drivers/StubDrivers.cs | 10 +- 7 files changed, 295 insertions(+), 21 deletions(-) create mode 100644 tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/NativeAlarmFeedSurvivesDeployTests.cs diff --git a/CLAUDE.md b/CLAUDE.md index 3691fd00..70fd354b 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -1162,6 +1162,11 @@ host logger factory (wired in `DriverFactoryBootstrap`; the FOCAS and TwinCAT fa FOCAS's is PAUSED, its told record kept, #647) — the device-state half of `ShutdownAsync` only; the poll engine is kept and the actor re-subscribes after every successful init. AbCip publishes fresh tag tables by reference swap (a second init used to throw "tag name collision" on `AdoptConfigDuringInit`); S7 swaps `Plc` under `_gate`. +- **Data and alarm subscriptions are independent in `DriverInstanceActor` (#662):** a data re-subscribe (every + deploy's bulk push to a kept driver) detaches DATA only (`DetachDataSubscription`) — it used to unhook + `OnAlarmEvent` too, silently killing the native-alarm feed until a respawn. The health-poll reconcile + re-attaches/re-subscribes alarms as well as data, so a failed alarm subscribe is retried every tick. + `docs/DriverLifecycle.md`. - **S7 (#632/#638):** STRING/WSTRING/CHAR/WCHAR and DTL/DATE/TOD/TIME/S5TIME codecs (`S7ValueCodecs`; TOD/TIME/S5TIME surface as Int32 ms); an undecodable buffer or unrepresentable write is `BadOutOfRange` with no health impact. Optional `localTsap`/`remoteTsap`/`connectionType` override the slot TSAP (`remoteTsap` ⊕ `connectionType`). Test diff --git a/docs/DriverLifecycle.md b/docs/DriverLifecycle.md index f1ea741a..0838cd34 100644 --- a/docs/DriverLifecycle.md +++ b/docs/DriverLifecycle.md @@ -242,6 +242,19 @@ Things to know before touching any of this: re-subscribing** so a filtering driver does not accumulate two overlapping ones. There is still deliberately no unsubscribe-on-**empty**: a removed alarm tag's condition node is torn down on the address-space rebuild and `ForwardNativeAlarm` drops any transition whose `ConditionId` no longer maps. +- ⚠️ **A data re-subscribe must not touch the alarm feed (#662).** Every deploy's bulk push re-sends + `SetDesiredSubscriptions` to every kept driver, and `HandleSubscribeAsync` drops the prior DATA + subscription first. That teardown used to be the full `DetachSubscription`, which also unhooked + `OnAlarmEvent` and forgot the alarm handle — so on any deploy that did **not** respawn a driver (e.g. one + that only disabled another driver) its 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 (`DisconnectObserved` / `ForceReconnect`) and `PostStop` detach both. The **#488 periodic + reconcile covers alarms too** (`ReconcileAlarmSubscription`): it re-attaches a missing `OnAlarmEvent` + handler and re-subscribes when the live alarm handle does not match the desired set, so an alarm subscribe + that failed on (re)connect — a session not yet ready after an outage — is retried every health-poll tick + instead of gating the feed for the life of the connection. OpcUaClient removes a half-built alarm + subscription from its session when the subscribe throws, so those retries do not accumulate. + End-to-end pin: `NativeAlarmFeedSurvivesDeployTests`. - **Restart tests must assert a FUNCTION, never existence.** `RestartDriver_respawns_the_child_without_an_actor_name_collision` asserts only that a child exists after a restart, and it passed green against the dead driver 03/S16 describes. It is kept — its own subject, the `InvalidActorNameException`, is real — but the load-bearing diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient/OpcUaClientDriver.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient/OpcUaClientDriver.cs index 53144baf..a5d6cad6 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient/OpcUaClientDriver.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient/OpcUaClientDriver.cs @@ -1710,6 +1710,30 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit ?? throw new InvalidOperationException( "OpcUaClient session was lost before the alarm subscription could be created."); session.AddSubscription(subscription); + try + { + await CreateAlarmEventItemAsync().ConfigureAwait(false); + } + 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); } + catch { /* best-effort — the session may be the thing that failed */ } + throw; + } + + // #619: a new event item only reports FUTURE transitions — a condition already active upstream would + // stay invisible until it next changed. ConditionRefresh re-delivers every retained condition's + // current state through the normal path (deduplicated against the last-told state). + await TryConditionRefreshAsync(subscription, cancellationToken).ConfigureAwait(false); + } + finally { _gate.Release(); } + + return handle; + + async Task CreateAlarmEventItemAsync() + { await subscription.CreateAsync(cancellationToken).ConfigureAwait(false); var eventItem = new MonitoredItem(telemetry: null!, new MonitoredItemOptions @@ -1736,15 +1760,7 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit await subscription.CreateItemsAsync(cancellationToken).ConfigureAwait(false); _alarmSubscriptions[id] = new RemoteAlarmSubscription(subscription, handle, eventItem, notifHandler); - - // #619: a new event item only reports FUTURE transitions — a condition already active upstream would - // stay invisible until it next changed. ConditionRefresh re-delivers every retained condition's - // current state through the normal path (deduplicated against the last-told state). - await TryConditionRefreshAsync(subscription, cancellationToken).ConfigureAwait(false); } - finally { _gate.Release(); } - - return handle; } /// diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverInstanceActor.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverInstanceActor.cs index ceeaab39..3bcf103a 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverInstanceActor.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverInstanceActor.cs @@ -690,6 +690,7 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers /// private void ReconcileSubscription() { + ReconcileAlarmSubscription(); if (_driver is not ISubscribable) return; if (_desiredRefs.Count == 0 || _subscriptionHandle is not null) return; @@ -699,6 +700,34 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers Self.Tell(new Subscribe(_desiredRefs, _desiredInterval)); } + /// + /// #662 — the native-alarm half of . The data reconcile above never + /// looked at alarms, so an alarm subscribe that failed on (re)connect — e.g. a session not yet ready + /// after an outage — left the feed gated for the life of the connection, with the driver Healthy and + /// its data streaming. Re-asserts both halves the feed needs: the OnAlarmEvent handler + /// (idempotent attach) and a subscription matching the desired set ( + /// is a no-op while the live handle already matches). Gated on + 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. + /// + 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)", + _driverInstanceId); + AttachAlarmSource(); + } + if (AlarmSubscriptionIsStale()) + { + _log.Info( + "DriverInstance {Id}: connected with {Count} desired alarm ref(s) but no matching native-alarm subscription — re-subscribing (periodic reconcile)", + _driverInstanceId, _desiredAlarmRefs.Count); + Self.Tell(new SubscribeAlarms()); + } + } + private void Reconnecting() { // #477 — announce comms loss to the host so it annotates this driver's native alarm conditions Bad @@ -1002,7 +1031,7 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers } catch (Exception ex) { - DetachSubscription(); + DetachDataSubscription(); _log.Warning(ex, "DriverInstance {Id}: subscribe failed", _driverInstanceId); replyTo.Tell(new SubscriptionFailed(ex.Message)); } @@ -1012,7 +1041,7 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers { if (_driver is not ISubscribable subscribable || _subscriptionHandle is null) { - DetachSubscription(); + DetachDataSubscription(); return; } try @@ -1031,17 +1060,32 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers } finally { - DetachSubscription(); + DetachDataSubscription(); } } - /// Tear down the data-change (per-tag AND batch) + native-alarm event handlers + null the handle. - /// Called from the Unsubscribe path, on PostStop, and on Connected → Reconnecting transitions so a stale - /// handler doesn't push data-change / alarm events to an actor that has lost its driver connection. + /// Tear down the data-change (per-tag AND batch) + native-alarm event handlers + null both handles. + /// Called on PostStop and on Connected → Reconnecting transitions so a stale handler doesn't push + /// data-change / alarm events to an actor that has lost its driver connection. + /// Not for the data Unsubscribe / re-subscribe path — that uses + /// (#662). + private void DetachSubscription() + { + DetachDataSubscription(); + DetachAlarmSource(); + } + + /// Tear down the data-change (per-tag AND batch) handlers + null the data handle, leaving the + /// native-alarm handler and handle alone. + /// #662: the data Unsubscribe / re-subscribe path used to run the full , + /// so every deploy that re-subscribed a Connected driver's DATA also unhooked OnAlarmEvent and + /// forgot the alarm handle — the alarm subscription still logged "established" but its events reached + /// nobody until a reconnect or respawn. The two subscriptions are independent at the driver; only a lost + /// connection ends both. /// Both data-change fields are cleared unconditionally even though only one is ever attached — the /// IDriver object outlives the actor when the host respawns a child around it, so a handler left on the /// driver's event would keep Telling a dead actor. - private void DetachSubscription() + private void DetachDataSubscription() { if (_driver is ISubscribable subscribable && _dataChangeHandler is not null) { @@ -1054,7 +1098,6 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers } _batchChangeHandler = null; _subscriptionHandle = null; - DetachAlarmSource(); } /// Subscribe the driver's native alarm event (if it is an ), @@ -1229,7 +1272,7 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers /// Calls the driver's (bounded) to register the /// alarm subscription that un-gates its native feed, caching the returned handle. Re-checks the guard /// (the desired set may have cleared, or another SubscribeAlarms may have already established a handle, - /// while this was queued). Failures are logged and retried on the next Connected entry — the feed simply + /// while this was queued). Failures are logged and retried on the next Connected entry or health-poll reconcile tick (#662) — the feed /// stays gated until then. private async Task HandleSubscribeAlarmsAsync(SubscribeAlarms _) { @@ -1276,7 +1319,7 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers } catch (Exception ex) { - _log.Warning(ex, "DriverInstance {Id}: native-alarm subscription failed — feed stays gated until reconnect", + _log.Warning(ex, "DriverInstance {Id}: native-alarm subscription failed — feed stays gated until the next reconcile tick retries it", _driverInstanceId); } } diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverInstanceActorNativeAlarmTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverInstanceActorNativeAlarmTests.cs index 1e1ce406..dbfcc356 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverInstanceActorNativeAlarmTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverInstanceActorNativeAlarmTests.cs @@ -218,6 +218,82 @@ public sealed class DriverInstanceActorNativeAlarmTests : RuntimeActorTestBase parent.ExpectNoMsg(TimeSpan.FromMilliseconds(100)); } + /// + /// Gitea #662 case 1 — re-subscribing a Connected driver's DATA (what every deploy's + /// SetDesiredSubscriptions 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 OnAlarmEvent and forgetting the alarm handle — so the alarm subscription + /// stayed "established" while every event it carried reached nobody until a reconnect. + /// + [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( + 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(_ => true, TimeSpan.FromSeconds(2)) + .Args.ConditionId.ShouldBe("Temp.HiHi"); + } + + /// + /// 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. + /// + [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(), 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(_ => 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"); + } + /// /// After a full reconnect cycle (Connected → Reconnecting → Connected), a single raised alarm still /// yields EXACTLY ONE — the @@ -354,10 +430,16 @@ public sealed class DriverInstanceActorNativeAlarmTests : RuntimeActorTestBase IReadOnlyList 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(new StubAlarmHandle()); } + /// How many leading calls throw — the "session not ready + /// after an outage" shape from #662. + public int AlarmSubscribeFailuresBeforeSuccess { get; init; } + /// Cancels an alarm subscription. public Task UnsubscribeAlarmsAsync(IAlarmSubscriptionHandle handle, CancellationToken cancellationToken) => Task.CompletedTask; @@ -416,6 +498,28 @@ public sealed class DriverInstanceActorNativeAlarmTests : RuntimeActorTestBase } } + /// Counts health-snapshot publishes — one per processed health-poll tick. + private sealed class CountingHealthPublisher : IDriverHealthPublisher + { + private int _count; + + /// Number of snapshots published so far. + public int Count => Volatile.Read(ref _count); + + /// + public void Publish( + string clusterId, + string driverInstanceId, + DriverHealth health, + int errorCount5Min, + DateTime? rediscoveryNeededUtc = null, + string? rediscoveryReason = null, + IReadOnlyList? hostStatuses = null, + SkippedTagReport? skippedTags = null, + IReadOnlyDictionary? diagnostics = null) + => Interlocked.Increment(ref _count); + } + /// Minimal for building . private sealed class StubAlarmHandle : IAlarmSubscriptionHandle { diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/NativeAlarmFeedSurvivesDeployTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/NativeAlarmFeedSurvivesDeployTests.cs new file mode 100644 index 00000000..d2b447f3 --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/NativeAlarmFeedSurvivesDeployTests.cs @@ -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; + +/// +/// Gitea #662 — a native-alarm feed must survive a deploy that does NOT respawn its driver, end to end +/// through the real and its real +/// child. +/// +/// +/// 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 Enabled=false), so none of +/// them could see the instance actor drop the feed. +/// +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 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( + m => m.AlarmNodeId == AlarmXPath && m.State.Active, PresenceBudget); + update.State.Acknowledged.ShouldBeFalse(); + } +} diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/StubDrivers.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/StubDrivers.cs index 3c4783d0..627a1c38 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/StubDrivers.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/StubDrivers.cs @@ -343,9 +343,15 @@ internal sealed class RecordingSubscribableDriver : IDriver, ISubscribable, IAla /// Raises ; present so the event is not merely written-to-never-read. /// The condition id to raise. - public void FireAlarm(string conditionId) => + /// The transition kind to report. + 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)); + + /// Live subscribers on — an alarm subscription with no handler + /// attached delivers nothing (#662), so this is the other half of "the feed is up". + public int AlarmSubscriberCount => OnAlarmEvent?.GetInvocationList().Length ?? 0; private sealed class RecordingAlarmHandle : IAlarmSubscriptionHandle { From 28ca04e4668108a1d1e31abd141e11be960ee011 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Fri, 9 Oct 2026 07:35:50 -0400 Subject: [PATCH 2/2] 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 --- .../OpcUaClientDriver.cs | 29 ++++++++++-- .../Drivers/DriverInstanceActor.cs | 47 ++++++++++++++++--- .../DriverInstanceActorNativeAlarmTests.cs | 16 +++++-- 3 files changed, 75 insertions(+), 17 deletions(-) diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient/OpcUaClientDriver.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient/OpcUaClientDriver.cs index a5d6cad6..e851bc9c 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient/OpcUaClientDriver.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient/OpcUaClientDriver.cs @@ -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); } diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverInstanceActor.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverInstanceActor.cs index 3bcf103a..4c8912cf 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverInstanceActor.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverInstanceActor.cs @@ -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. private IReadOnlyList _subscribedAlarmRefs = Array.Empty(); + /// #662 — consecutive failures; reset on success and on + /// (a new connection starts a new streak). + private int _alarmSubscribeFailures; + + /// #662 — health-poll ticks the alarm reconcile still skips before its next retry (backoff). + private int _alarmRetrySkipTicks; + + /// #662 — backoff cap: at the production 30 s tick a failing alarm subscribe is retried at least + /// every 4 minutes. + private const int MaxAlarmRetrySkipTicks = 7; + /// /// Gets or sets the timer scheduler for scheduling reconnection attempts. /// @@ -707,18 +718,27 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers /// its data streaming. Re-asserts both halves the feed needs: the OnAlarmEvent handler /// (idempotent attach) and a subscription matching the desired set ( /// is a no-op while the live handle already matches). Gated on + 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. + /// A persistently failing subscribe backs off: after the n-th consecutive failure the reconcile + /// skips 2^(n-1) − 1 ticks, capped at , 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. /// 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(); + _alarmSubscribeFailures = 0; + _alarmRetrySkipTicks = 0; } /// Establish the native-alarm subscription that un-gates an 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); } } diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverInstanceActorNativeAlarmTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverInstanceActorNativeAlarmTests.cs index dbfcc356..c3895b53 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverInstanceActorNativeAlarmTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverInstanceActorNativeAlarmTests.cs @@ -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. /// - [Fact] - public void A_failed_alarm_subscribe_is_retried_by_the_periodic_reconcile() + /// Leading failures; 3 crosses the backoff (skips 0, 1, then 3 ticks), proving the + /// retries keep coming rather than giving up. + [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);