From 18a20981a593c4f6e66d1fba21a8e9fd4cad3bc3 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Thu, 8 Oct 2026 14:39:05 -0400 Subject: [PATCH] fix(drivers): #630 review - AbCip disposes alarm projection + clears tag tables, join not abortable by init token; S7 swaps Plc under _gate; doc rationale Claude-Session: https://claude.ai/code/session_01Ef2ozxS7QPanGT8uoATbTe --- docs/DriverLifecycle.md | 2 +- .../AbCipDriver.cs | 11 ++++++-- .../AbLegacyDriver.cs | 2 +- .../FocasDriver.cs | 2 +- .../ZB.MOM.WW.OtOpcUa.Driver.S7/S7Driver.cs | 22 +++++++++++---- .../AbCipAlarmProjectionTests.cs | 27 +++++++++++++++++++ 6 files changed, 56 insertions(+), 10 deletions(-) diff --git a/docs/DriverLifecycle.md b/docs/DriverLifecycle.md index 8382f48d..41cc3b71 100644 --- a/docs/DriverLifecycle.md +++ b/docs/DriverLifecycle.md @@ -432,7 +432,7 @@ the previous state's probe / recycle / fixed-tree loops and its client. FOCAS, A OpcUaClient now cancel, bounded-join (5 s) and dispose that state at the top of `InitializeAsync` (`ReleaseDeviceStatesAsync` / `ReleaseConnectionAsync` / `ReleaseSessionAsync`), alongside Modbus and MQTT. It is deliberately the device-state half of `ShutdownAsync` only: the poll engine and subscriptions are *not* -dropped, because nothing would re-drive them. A new driver must do the same. +dropped: the release is scoped to device state so it cannot take state the shutdown path owns. The actor re-subscribes values and alarms after every successful init, so driver-side subscriptions kept across a no-teardown re-init are retained leftovers, not needed for data flow. A new driver must do the same. #### The fan-out that watchdog triggers: indexed per driver, and one batch diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip/AbCipDriver.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip/AbCipDriver.cs index e55a3576..f621240d 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip/AbCipDriver.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip/AbCipDriver.cs @@ -471,7 +471,9 @@ public sealed class AbCipDriver : IDriver, IReadable, IWritable, ITagDiscovery, /// private async Task ReleaseDeviceStatesAsync(CancellationToken cancellationToken) { - if (_devices.Count == 0) return; + // The projection is replaced (not reused) when a config body is adopted; dispose the old one so its + // poll loop does not outlive it and raise duplicate alarm events through this driver. + await _alarmProjection.DisposeAsync().ConfigureAwait(false); foreach (var state in _devices.Values) { try { state.ProbeCts?.Cancel(); } catch (ObjectDisposedException) { } @@ -480,7 +482,9 @@ public sealed class AbCipDriver : IDriver, IReadable, IWritable, ITagDiscovery, { var probeTask = state.ProbeTask; if (probeTask is null) continue; - try { await probeTask.WaitAsync(TimeSpan.FromSeconds(5), cancellationToken).ConfigureAwait(false); } + // Own bounded timeout, independent of the caller token: cancelling init must not abort the join + // and dispose handles under a probe loop that is still running. + try { await probeTask.WaitAsync(TimeSpan.FromSeconds(5)).ConfigureAwait(false); } catch (TimeoutException) { } catch (OperationCanceledException) { } } @@ -493,6 +497,9 @@ public sealed class AbCipDriver : IDriver, IReadable, IWritable, ITagDiscovery, } _devices.Clear(); _devicesByRoutingKey.Clear(); + // Rebuilt wholesale by Initialize; stale entries would trip its member fan-out collision check. + _tagsByRawPath.Clear(); + _declaredTags.Clear(); } /// diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbLegacy/AbLegacyDriver.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbLegacy/AbLegacyDriver.cs index 0ef75617..e23eb5d1 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbLegacy/AbLegacyDriver.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbLegacy/AbLegacyDriver.cs @@ -148,7 +148,7 @@ public sealed class AbLegacyDriver : IDriver, IReadable, IWritable, ITagDiscover // #630: InitializeAsync can run on a live driver (DriverInstanceActor.AdoptConfigDuringInit calls it // without a teardown first). Release the previous device states' probe loops + runtimes BEFORE the // overwrite below, or they are orphaned beyond the reach of ShutdownAsync. Deliberately NOT - // ShutdownAsync: that also drops _poll and every subscription, which nothing would re-drive. + // ShutdownAsync: that also drops _poll and every subscription, which are owned by the shutdown path (the actor re-subscribes after every successful init). await ReleaseDeviceStatesAsync().ConfigureAwait(false); _health = new DriverHealth(DriverState.Initializing, null, null); _logger.LogInformation( diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.FOCAS/FocasDriver.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.FOCAS/FocasDriver.cs index bf35b356..7152be71 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.FOCAS/FocasDriver.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.FOCAS/FocasDriver.cs @@ -182,7 +182,7 @@ public sealed class FocasDriver : IDriver, IReadable, IWritable, ITagDiscovery, { // #630: release the previous device states' loops + clients BEFORE they are overwritten (the actor's // AdoptConfigDuringInit calls this without a teardown first). Not ShutdownAsync: that also drops _poll - // and subscriptions, which nothing would re-drive. + // and subscriptions, which are owned by the shutdown path (the actor re-subscribes after every successful init). await ReleaseDeviceStatesAsync().ConfigureAwait(false); Volatile.Write(ref _health, new DriverHealth(DriverState.Initializing, null, null)); try diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.S7/S7Driver.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.S7/S7Driver.cs index 8dec593b..adda1429 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.S7/S7Driver.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.S7/S7Driver.cs @@ -226,7 +226,7 @@ public sealed class S7Driver // #630: InitializeAsync is reachable on a live driver (DriverInstanceActor.AdoptConfigDuringInit // runs it with no teardown first). Stop the previous probe loop and drop the previous connection // before they are overwritten below. Deliberately NOT ShutdownAsync: that also drops _poll and the - // subscriptions, which nothing would re-drive. + // subscriptions, which are owned by the shutdown path (the actor re-subscribes after every successful init). await ReleaseConnectionAsync().ConfigureAwait(false); Health = new DriverHealth(DriverState.Initializing, null, null); try @@ -395,10 +395,22 @@ public sealed class S7Driver _probeCts = null; _probeTask = null; - _initialized = false; - _plcDead = false; - try { Plc?.Dispose(); } catch { /* best-effort — tearing down anyway */ } - Plc = null; + // Swap the connection out under _gate so a Read/Write/poll pass that already holds the gate (they + // re-read Plc inside it) can never see a disposed Plc mid-swap. Bounded: a wedged holder must not hang + // teardown, in which case we proceed exactly as the un-gated path always did. + var gated = false; + try { gated = await _gate.WaitAsync(DrainTimeout).ConfigureAwait(false); } catch { } + try + { + _initialized = false; + _plcDead = false; + try { Plc?.Dispose(); } catch { /* best-effort — tearing down anyway */ } + Plc = null; + } + finally + { + if (gated) { try { _gate.Release(); } catch (ObjectDisposedException) { } } + } } /// diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.Tests/AbCipAlarmProjectionTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.Tests/AbCipAlarmProjectionTests.cs index 8542a977..808b3043 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.Tests/AbCipAlarmProjectionTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.Tests/AbCipAlarmProjectionTests.cs @@ -229,6 +229,33 @@ public sealed class AbCipAlarmProjectionTests } /// Waits for further reads of the fake runtime (a processed-count wait). + /// + /// #630 — a re-Initialize on a live driver must dispose the alarm projection's poll loop rather + /// than leave it running (it would keep raising duplicate alarm events through this driver). + /// + [Fact] + public async Task Reinitialize_Stops_The_Previous_Projection_Poll_Loop() + { + var factory = new FakeAbCipTagFactory(); + var drv = new AbCipDriver(Options(enabled: true), "drv-1", factory); + await drv.InitializeAsync("{}", CancellationToken.None); + await drv.SubscribeAlarmsAsync(["HighTemp"], CancellationToken.None); + + await WaitForTagCreation(factory, "HighTemp"); + var tag = factory.Tags["HighTemp"]; + // Positive control: the loop is demonstrably polling before the re-init. + await WaitForMoreReadsAsync(factory, "HighTemp", 3); + + await drv.InitializeAsync("{}", CancellationToken.None); + + int Reads() => factory.Tags.Values.Sum(t => t.ReadCount); // the loop may re-create its runtime on the new device state + var frozen = Reads(); + await Task.Delay(300); // ~15 poll intervals; a live loop would advance the count + Reads().ShouldBe(frozen); + + await drv.ShutdownAsync(CancellationToken.None); + } + private static async Task WaitForMoreReadsAsync(FakeAbCipTagFactory factory, string tagName, int more) { var tag = factory.Tags[tagName];