From 39c5af3d7f7c0613ae0055e78d03f0951141aeba Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Fri, 9 Oct 2026 07:49:56 -0400 Subject: [PATCH] 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 --- CLAUDE.md | 2 + docs/DriverLifecycle.md | 9 ++ .../Drivers/DriverInstanceActor.cs | 43 +++++++ .../DriverInstanceActorNativeAlarmTests.cs | 112 ++++++++++++++++-- 4 files changed, 158 insertions(+), 8 deletions(-) diff --git a/CLAUDE.md b/CLAUDE.md index 70fd354b..48c557c9 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -1166,6 +1166,8 @@ host logger factory (wired in `DriverFactoryBootstrap`; the FOCAS and TwinCAT fa 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. + A Connected in-place `ApplyDelta` (unreachable from deploys since #516) releases both subscriptions at the + driver, reinitialises, then re-subscribes both (#663) — it used to keep handles the reinit had invalidated. `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. diff --git a/docs/DriverLifecycle.md b/docs/DriverLifecycle.md index 0838cd34..dda21ba4 100644 --- a/docs/DriverLifecycle.md +++ b/docs/DriverLifecycle.md @@ -255,6 +255,15 @@ Things to know before touching any of this: 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`. +- **An in-place `ApplyDelta` on a Connected driver releases and re-establishes BOTH subscriptions (#663).** + `ReinitializeAsync` drops every driver-side subscription (#630), but the handler used to keep the actor's + handles, and the reconcile only re-subscribes a *missing* handle — so the driver went silent until a + reconnect. It now releases data and alarms at the driver first (explicitly, so a driver whose reinit keeps + its subscriptions is not left holding two), reinitialises, and re-subscribes both in a `finally` (a failed + reinit has lost them too). It stays on `ReinitializeAsync` rather than shutdown + initialise because + Calculation and Galaxy take their config at construction and ignore the JSON passed to `InitializeAsync`. + Not reachable from a deploy since #516 (a DriverConfig change respawns). Pin: + `An_in_place_ApplyDelta_leaves_one_live_data_and_one_live_alarm_subscription` (one case per wrong fix). - **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/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverInstanceActor.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverInstanceActor.cs index 4c8912cf..dd669688 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverInstanceActor.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverInstanceActor.cs @@ -881,9 +881,21 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers Sender.Tell(new ApplyResult(true, "config adopted; reinitializing", msg.Correlation)); } + /// Adopt a new config while Connected, in place, via . + /// #663: every driver's reinit releases its device/session state (#630), so the driver-side data and + /// native-alarm subscriptions do not survive it — but this handler used to keep both handles, and the + /// health-poll reconcile only re-subscribes a MISSING handle, so the driver went silent until a + /// reconnect or respawn. Both subscriptions are now released explicitly BEFORE the reinit (explicitly, + /// not just by nulling the handles, so a driver whose reinit keeps its subscriptions is not left + /// holding two) and re-established after it — in a finally, because a failed reinit has already + /// lost them too. Reinit stays the adoption path rather than shutdown + initialise: Calculation and + /// Galaxy take their config at construction and ignore 's JSON. + /// Not reachable from a deploy since #516 (a DriverConfig change respawns the driver). private async Task HandleApplyDeltaAsync(ApplyDelta msg) { var replyTo = Sender; + await UnsubscribeAsync(); + await ReleaseAlarmSubscriptionAsync(); try { await _driver.ReinitializeAsync(msg.DriverConfigJson, CancellationToken.None); @@ -900,6 +912,37 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers { replyTo.Tell(new ApplyResult(false, ex.Message, msg.Correlation)); } + finally + { + ResubscribeDesired(); + AttachAlarmSource(); + SubscribeDesiredAlarms(); + } + } + + /// Release the native-alarm subscription at the driver (best-effort, bounded) and drop the + /// actor's handler + handle. The in-place reinit counterpart of the data + /// (#663). + private async Task ReleaseAlarmSubscriptionAsync() + { + if (_driver is IAlarmSource src && _alarmSubscriptionHandle is { } handle) + { + try + { + using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(5)); + await _invoker.ExecuteAsync( + DriverCapability.AlarmSubscribe, + _driverInstanceId, + async ct => await src.UnsubscribeAlarmsAsync(handle, ct), + cts.Token); + } + catch (Exception ex) + { + _log.Warning(ex, "DriverInstance {Id}: releasing the native-alarm subscription ({Diag}) threw (continuing)", + _driverInstanceId, handle.DiagnosticId); + } + } + DetachAlarmSource(); } private async Task HandleWriteAsync(WriteAttribute msg) 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 c3895b53..7663c411 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 @@ -369,6 +369,56 @@ public sealed class DriverInstanceActorNativeAlarmTests : RuntimeActorTestBase .Connected.ShouldBeTrue(); } + /// + /// Gitea #663 — an in-place config change on a Connected driver ( + /// → ) 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. + /// + /// 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. + [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(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(_ => true, PresenceBudget) + .Args.ConditionId.ShouldBe("Temp.HiHi"); + } + // --- stub drivers ---------------------------------------------------------------------------- private class StubDriver : IDriver @@ -389,7 +439,7 @@ public sealed class DriverInstanceActorNativeAlarmTests : RuntimeActorTestBase } /// Reinitializes the driver. - public Task ReinitializeAsync(string driverConfigJson, CancellationToken cancellationToken) => + public virtual Task ReinitializeAsync(string driverConfigJson, CancellationToken cancellationToken) => Task.CompletedTask; /// Shuts down the driver. @@ -413,14 +463,55 @@ public sealed class DriverInstanceActorNativeAlarmTests : RuntimeActorTestBase /// Number of live subscribers on — proves attach/detach. public int AlarmSubscriberCount => OnAlarmEvent?.GetInvocationList().Length ?? 0; + private readonly object _liveLock = new(); + private readonly HashSet _liveData = new(ReferenceEqualityComparer.Instance); + private readonly HashSet _liveAlarms = new(ReferenceEqualityComparer.Instance); + private int _reinitializeCount; + + /// Driver-side data subscriptions currently open (subscribed and not yet released). + public int LiveDataSubscriptions { get { lock (_liveLock) return _liveData.Count; } } + + /// Driver-side native-alarm subscriptions currently open. + public int LiveAlarmSubscriptions { get { lock (_liveLock) return _liveAlarms.Count; } } + + /// Number of calls. + public int ReinitializeCount => Volatile.Read(ref _reinitializeCount); + + /// 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. + public bool ReinitKeepsSubscriptions { get; init; } + + /// + public override Task ReinitializeAsync(string driverConfigJson, CancellationToken cancellationToken) + { + Interlocked.Increment(ref _reinitializeCount); + if (!ReinitKeepsSubscriptions) + { + lock (_liveLock) + { + _liveData.Clear(); + _liveAlarms.Clear(); + } + } + return Task.CompletedTask; + } + /// Subscribes to the specified full references. public Task SubscribeAsync( - IReadOnlyList fullReferences, TimeSpan publishingInterval, CancellationToken cancellationToken) => - Task.FromResult(_subHandle); + IReadOnlyList fullReferences, TimeSpan publishingInterval, CancellationToken cancellationToken) + { + var handle = new StubHandle(); + lock (_liveLock) _liveData.Add(handle); + return Task.FromResult(handle); + } /// Unsubscribes from the specified subscription handle. - 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(new StubAlarmHandle()); + var handle = new StubAlarmHandle(); + lock (_liveLock) _liveAlarms.Add(handle); + return Task.FromResult(handle); } /// How many leading calls throw — the "session not ready @@ -447,8 +540,11 @@ public sealed class DriverInstanceActorNativeAlarmTests : RuntimeActorTestBase public int AlarmSubscribeFailuresBeforeSuccess { get; init; } /// Cancels an alarm subscription. - public Task UnsubscribeAlarmsAsync(IAlarmSubscriptionHandle handle, CancellationToken cancellationToken) => - Task.CompletedTask; + public Task UnsubscribeAlarmsAsync(IAlarmSubscriptionHandle handle, CancellationToken cancellationToken) + { + lock (_liveLock) _liveAlarms.Remove(handle); + return Task.CompletedTask; + } /// Acknowledges a batch of alarms. public Task AcknowledgeAsync(