diff --git a/docs/DriverLifecycle.md b/docs/DriverLifecycle.md index c1651178..41cc3b71 100644 --- a/docs/DriverLifecycle.md +++ b/docs/DriverLifecycle.md @@ -425,6 +425,15 @@ re-entered this path; the watchdog makes it automatic and unbounded, so a device have leaked ~360 transports at the 10 s retry interval. The shutdown is best-effort: a driver that throws on it still gets its initialise attempt. +**`InitializeAsync` itself releases the previous device/connection state first (#630).** The retry above is not +the only way in: `DriverInstanceActor.AdoptConfigDuringInit` calls `InitializeAsync` on a live driver with *no* +teardown, so a driver that does `_devices[host] = new DeviceState(...)` (or `Session = ...`, `Plc = ...`) orphans +the previous state's probe / recycle / fixed-tree loops and its client. FOCAS, AbLegacy, AbCip, TwinCAT, S7 and +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: 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 `DisconnectObserved` raises `ConnectivityChanged`, and `DriverHostActor` answers it by pushing 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 9157f8ba..a80a86e0 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip/AbCipDriver.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip/AbCipDriver.cs @@ -106,10 +106,10 @@ public sealed class AbCipDriver : IDriver, IReadable, IWritable, ITagDiscovery, private readonly Dictionary _devicesByRoutingKey = new(StringComparer.OrdinalIgnoreCase); // v3 authored RawPath → definition table (parents + fanned-out UDT members), keyed case-sensitive // ordinal. Built in InitializeAsync from _options.RawTags via the pure mapper. - private readonly Dictionary _tagsByRawPath = new(StringComparer.Ordinal); + private volatile Dictionary _tagsByRawPath = new(StringComparer.Ordinal); // Top-level authored definitions (parents + flat tags, NOT synthesised members), in author order — // the source DiscoverAsync groups by device. Members are synthesised inside the discovery/read paths. - private readonly List _declaredTags = new(); + private volatile List _declaredTags = new(); // Resolves a read/write/subscribe fullReference (always a RawPath in v3) to a tag definition by a // single hit on the authored _tagsByRawPath table; a miss is a miss (the pre-v3 blob-parse fallback @@ -335,8 +335,12 @@ public sealed class AbCipDriver : IDriver, IReadable, IWritable, ITagDiscovery, public string DriverType => "AbCip"; /// - public Task InitializeAsync(string driverConfigJson, CancellationToken cancellationToken) + public async Task InitializeAsync(string driverConfigJson, CancellationToken cancellationToken) { + // #630: InitializeAsync is reachable on a live driver (DriverInstanceActor.AdoptConfigDuringInit + // runs it with no teardown first). Release the previous device states' probe loops + handles before + // the overwrite below. Deliberately NOT ShutdownAsync: that also drops _poll and the subscriptions. + await ReleaseDeviceStatesAsync(cancellationToken).ConfigureAwait(false); _health = new DriverHealth(DriverState.Initializing, null, null); _logger.LogInformation( "AbCip driver {DriverInstanceId} initializing: transport={Transport} tagFactory={TagFactory}", @@ -373,6 +377,8 @@ public sealed class AbCipDriver : IDriver, IReadable, IWritable, ITagDiscovery, // threaded onto the def (they live on the RawTagEntry, not inside the blob). A mapper miss is // skipped (logged), never thrown — a bad tag must not fail the whole driver init. _skippedTags.BeginRebuild(_options.RawTags.Count); + var builtTagsByRawPath = new Dictionary(StringComparer.Ordinal); + var builtDeclaredTags = new List(); foreach (var entry in _options.RawTags) { if (!AbCipTagDefinitionFactory.FromTagConfig(entry.TagConfig, entry.RawPath, out var mapped)) @@ -387,13 +393,13 @@ public sealed class AbCipDriver : IDriver, IReadable, IWritable, ITagDiscovery, // Duplicate-key check: a collision means two authored tags share the same RawPath. // Fail fast at init time with a diagnostic rather than silently clobbering. - if (_tagsByRawPath.TryGetValue(tag.Name, out var existingTag)) + if (builtTagsByRawPath.TryGetValue(tag.Name, out var existingTag)) throw new InvalidOperationException( $"AbCip tag name collision: '{tag.Name}' is declared more than once. " + $"Existing entry DeviceHostAddress='{existingTag.DeviceHostAddress}', " + $"TagPath='{existingTag.TagPath}'. Rename or remove the duplicate."); - _tagsByRawPath[tag.Name] = tag; - _declaredTags.Add(tag); + builtTagsByRawPath[tag.Name] = tag; + builtDeclaredTags.Add(tag); if (tag.DataType == AbCipDataType.Structure && tag.Members is { Count: > 0 }) { @@ -416,15 +422,18 @@ public sealed class AbCipDriver : IDriver, IReadable, IWritable, ITagDiscovery, // Member fan-out duplicate check: a member-path collision means two // configured structure tags produce the same member path, or a member // name collides with an independently-declared tag. - if (_tagsByRawPath.TryGetValue(memberTag.Name, out var existingMember)) + if (builtTagsByRawPath.TryGetValue(memberTag.Name, out var existingMember)) throw new InvalidOperationException( $"AbCip tag name collision: '{memberTag.Name}' is produced by both " + $"'{tag.Name}.{member.Name}' (member fan-out) and an existing tag " + $"'{existingMember.Name}'. Rename one of the configured tags to resolve."); - _tagsByRawPath[memberTag.Name] = memberTag; + builtTagsByRawPath[memberTag.Name] = memberTag; } } } + // Publish the finished tables by reference swap — readers see the old or the new, never a half-built one. + _tagsByRawPath = builtTagsByRawPath; + _declaredTags = builtDeclaredTags; _skippedTags.Commit(); // Probe loops — one per device when enabled + a ProbeTagPath is configured. @@ -459,7 +468,43 @@ public sealed class AbCipDriver : IDriver, IReadable, IWritable, ITagDiscovery, _logger.LogError(ex, "AbCip driver {DriverInstanceId} failed to initialize", _driverInstanceId); throw; } - return Task.CompletedTask; + } + + /// + /// Cancels, joins (bounded) and disposes every existing device state's probe loop + handles, then + /// clears the device/route tables. Device part of only. + /// + private async Task ReleaseDeviceStatesAsync(CancellationToken cancellationToken) + { + // 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) { } + } + foreach (var state in _devices.Values) + { + var probeTask = state.ProbeTask; + if (probeTask is null) continue; + // 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) { } + } + foreach (var state in _devices.Values) + { + state.ProbeCts?.Dispose(); + state.ProbeCts = null; + state.ProbeTask = null; + state.DisposeHandles(); + } + _devices.Clear(); + _devicesByRoutingKey.Clear(); + // Rebuilt wholesale by Initialize; stale entries would trip its member fan-out collision check. + _tagsByRawPath = new(StringComparer.Ordinal); // swap, never Clear(): the retained poll engine / discovery may be enumerating + _declaredTags = new(); } /// @@ -506,8 +551,8 @@ public sealed class AbCipDriver : IDriver, IReadable, IWritable, ITagDiscovery, } _devices.Clear(); _devicesByRoutingKey.Clear(); - _tagsByRawPath.Clear(); - _declaredTags.Clear(); + _tagsByRawPath = new(StringComparer.Ordinal); + _declaredTags = new(); _resolver.Clear(); // no-op in v3 (the resolver reads the live table by closure); kept for symmetry _health = new DriverHealth(DriverState.Unknown, _health.LastSuccessfulRead, null); } 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 982df163..e23eb5d1 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbLegacy/AbLegacyDriver.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbLegacy/AbLegacyDriver.cs @@ -143,8 +143,13 @@ public sealed class AbLegacyDriver : IDriver, IReadable, IWritable, ITagDiscover } /// - public Task InitializeAsync(string driverConfigJson, CancellationToken cancellationToken) + public async Task InitializeAsync(string driverConfigJson, CancellationToken cancellationToken) { + // #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 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( "AbLegacy driver {DriverInstanceId} initializing: transport={Transport} tagFactory={TagFactory}", @@ -255,7 +260,35 @@ public sealed class AbLegacyDriver : IDriver, IReadable, IWritable, ITagDiscover _resolver.Clear(); // drop transient equipment-tag parses so a config change re-parses throw; } - return Task.CompletedTask; + } + + /// + /// Cancels, bounded-joins and disposes every existing device state, then clears the table. + /// Device-state part of only (no poll / subscription teardown). + /// + private async Task ReleaseDeviceStatesAsync() + { + if (_devices.Count == 0) return; + foreach (var state in _devices.Values) + try { state.ProbeCts?.Cancel(); } catch { } + var loops = _devices.Values.Select(d => d.ProbeTask).OfType().ToList(); + if (loops.Count > 0) + { + var all = Task.WhenAll(loops); + var finished = await Task.WhenAny(all, Task.Delay(LoopJoinTimeout, CancellationToken.None)).ConfigureAwait(false); + if (finished != all) + _logger.LogWarning( + "AbLegacy re-initialise: {Count} probe loop(s) did not stop within {Timeout}; proceeding. Driver={DriverInstanceId}", + loops.Count, LoopJoinTimeout, _driverInstanceId); + } + foreach (var state in _devices.Values) + { + state.ProbeCts?.Dispose(); + state.ProbeCts = null; + state.ProbeTask = null; + state.DisposeRuntimes(); + } + _devices.Clear(); } /// 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 07fc913c..7152be71 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.FOCAS/FocasDriver.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.FOCAS/FocasDriver.cs @@ -178,8 +178,12 @@ public sealed class FocasDriver : IDriver, IReadable, IWritable, ITagDiscovery, /// The driver configuration JSON (see remarks — not re-parsed). /// Cancellation token for the operation. /// A completed task. - public Task InitializeAsync(string driverConfigJson, CancellationToken cancellationToken) + public async Task InitializeAsync(string driverConfigJson, CancellationToken cancellationToken) { + // #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 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 { @@ -316,7 +320,56 @@ public sealed class FocasDriver : IDriver, IReadable, IWritable, ITagDiscovery, Volatile.Write(ref _health, new DriverHealth(DriverState.Faulted, null, ex.Message)); throw; } - return Task.CompletedTask; + } + + /// + /// Cancels, bounded-joins and disposes every existing device state (probe / recycle / fixed-tree + /// loops + client) and the alarm projection, then clears the table. Device part of + /// only (no poll / subscription teardown). + /// + private async Task ReleaseDeviceStatesAsync() + { + if (_alarmProjection is { } proj) + { + await proj.DisposeAsync().ConfigureAwait(false); + _alarmProjection = null; + } + if (_devices.Count == 0) return; + var loops = new List(); + foreach (var state in _devices.Values) + { + try { state.ProbeCts?.Cancel(); } + catch (Exception ex) { _logger.LogDebug(ex, "Cancelling probe CTS for {Host} failed", state.Options.HostAddress); } + try { state.RecycleCts?.Cancel(); } + catch (Exception ex) { _logger.LogDebug(ex, "Cancelling recycle CTS for {Host} failed", state.Options.HostAddress); } + try { state.FixedTreeCts?.Cancel(); } + catch (Exception ex) { _logger.LogDebug(ex, "Cancelling fixed-tree CTS for {Host} failed", state.Options.HostAddress); } + if (state.ProbeTask is { } pt) loops.Add(pt); + if (state.RecycleTask is { } rt) loops.Add(rt); + if (state.FixedTreeTask is { } ft) loops.Add(ft); + } + if (loops.Count > 0) + { + var all = Task.WhenAll(loops); + var finished = await Task.WhenAny(all, Task.Delay(LoopJoinTimeout, CancellationToken.None)).ConfigureAwait(false); + if (finished != all) + _logger.LogWarning("FOCAS re-initialise: {Count} background loop(s) did not stop within {Timeout}; proceeding", + loops.Count, LoopJoinTimeout); + } + foreach (var state in _devices.Values) + { + state.ProbeCts?.Dispose(); + state.ProbeCts = null; + state.ProbeTask = null; + state.RecycleCts?.Dispose(); + state.RecycleCts = null; + state.RecycleTask = null; + state.FixedTreeCts?.Dispose(); + state.FixedTreeCts = null; + state.FixedTreeTask = null; + state.DisposeClient(); + } + _devices.Clear(); } /// 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 967111cf..a5a3c25f 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient/OpcUaClientDriver.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient/OpcUaClientDriver.cs @@ -194,6 +194,10 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit /// public async Task InitializeAsync(string driverConfigJson, CancellationToken cancellationToken) { + // #630: InitializeAsync is reachable on a live driver (DriverInstanceActor.AdoptConfigDuringInit runs + // it with no teardown first); `Session = session` below would otherwise orphan the previous session, + // its keep-alive handler and any reconnect handler. + await ReleaseSessionAsync(cancellationToken).ConfigureAwait(false); _health = new DriverHealth(DriverState.Initializing, null, null); try { @@ -587,32 +591,14 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit await InitializeAsync(driverConfigJson, cancellationToken).ConfigureAwait(false); } - /// - public async Task ShutdownAsync(CancellationToken cancellationToken) + /// + /// Cancels any in-flight reconnect, detaches the keep-alive handler and closes + disposes the current + /// session. The session part of ; also run at the top of + /// so a re-initialise on a live driver (#630) does not orphan the + /// previous session and its reconnect handler. Subscriptions are NOT touched here. + /// + private async Task ReleaseSessionAsync(CancellationToken cancellationToken) { - // Tear down remote subscriptions first — otherwise Session.Close will try and may fail - // with BadSubscriptionIdInvalid noise in the upstream log. _subscriptions is cleared - // whether or not the wire-side delete succeeds since the local handles are useless - // after close anyway. Before deleting each subscription we detach the Notification - // handlers we attached at subscribe time so the SDK's invocation list no longer - // holds the driver instance through the closure. - foreach (var rs in _subscriptions.Values) - { - DetachNotificationHandlers(rs.ItemHandlers); - try { await rs.Subscription.DeleteAsync(silent: true, cancellationToken).ConfigureAwait(false); } - catch { /* best-effort */ } - } - _subscriptions.Clear(); - - foreach (var ras in _alarmSubscriptions.Values) - { - try { ras.EventItem.Notification -= ras.Handler; } - catch { /* best-effort */ } - try { await ras.Subscription.DeleteAsync(silent: true, cancellationToken).ConfigureAwait(false); } - catch { /* best-effort */ } - } - _alarmSubscriptions.Clear(); - // Abort any in-flight reconnect attempts before touching the session — BeginReconnect's // retry loop holds a reference to the current session and would fight Session.CloseAsync // if left spinning. Take the handler under _probeLock so a keep-alive callback racing @@ -645,6 +631,35 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit try { if (sessionToClose is Session s) await s.CloseAsync(cancellationToken).ConfigureAwait(false); } catch { /* best-effort */ } try { sessionToClose?.Dispose(); } catch { } + } + + /// + public async Task ShutdownAsync(CancellationToken cancellationToken) + { + // Tear down remote subscriptions first — otherwise Session.Close will try and may fail + // with BadSubscriptionIdInvalid noise in the upstream log. _subscriptions is cleared + // whether or not the wire-side delete succeeds since the local handles are useless + // after close anyway. Before deleting each subscription we detach the Notification + // handlers we attached at subscribe time so the SDK's invocation list no longer + // holds the driver instance through the closure. + foreach (var rs in _subscriptions.Values) + { + DetachNotificationHandlers(rs.ItemHandlers); + try { await rs.Subscription.DeleteAsync(silent: true, cancellationToken).ConfigureAwait(false); } + catch { /* best-effort */ } + } + _subscriptions.Clear(); + + foreach (var ras in _alarmSubscriptions.Values) + { + try { ras.EventItem.Notification -= ras.Handler; } + catch { /* best-effort */ } + try { await ras.Subscription.DeleteAsync(silent: true, cancellationToken).ConfigureAwait(false); } + catch { /* best-effort */ } + } + _alarmSubscriptions.Clear(); + + await ReleaseSessionAsync(cancellationToken).ConfigureAwait(false); _namespaceMap = null; _connectedEndpointUrl = null; 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 7c2a0ef1..43a5dc08 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.S7/S7Driver.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.S7/S7Driver.cs @@ -225,6 +225,11 @@ public sealed class S7Driver /// public async Task InitializeAsync(string driverConfigJson, CancellationToken cancellationToken) { + // #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 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 { @@ -363,7 +368,18 @@ public sealed class S7Driver // disposing its CTS (drain-before-dispose, STAB-6-S7), so no poll iteration touches _gate // after this returns. await _poll.DisposeAsync().ConfigureAwait(false); + await ReleaseConnectionAsync().ConfigureAwait(false); + _writeFault = null; + _connectBackoff.RecordSuccess(); + Health = new DriverHealth(DriverState.Unknown, Health.LastSuccessfulRead, null); + } + /// + /// Cancels + bounded-joins the probe loop, then disposes its CTS and the PLC connection. The + /// connection part of (everything but the poll engine and health). + /// + private async Task ReleaseConnectionAsync() + { var probeCts = _probeCts; var probeTask = _probeTask; try { probeCts?.Cancel(); } catch { } @@ -382,14 +398,27 @@ public sealed class S7Driver _probeCts = null; _probeTask = null; - _initialized = false; - _plcDead = false; - _writeFault = null; - _connectBackoff.RecordSuccess(); - try { Plc?.Dispose(); } catch { /* best-effort — tearing down anyway */ } - Plc = null; - Volatile.Write(ref _negotiatedPduSize, 0); // a stale post-teardown reading would read as "online" - Health = new DriverHealth(DriverState.Unknown, Health.LastSuccessfulRead, 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 { } + if (!gated) + _logger.LogWarning( + "S7 driver {DriverInstanceId}: disposing the S7 connection while an operation still holds the gate (waited {Timeout})", + _driverInstanceId, DrainTimeout); + try + { + _initialized = false; + _plcDead = false; + try { Plc?.Dispose(); } catch { /* best-effort — tearing down anyway */ } + Plc = null; + Volatile.Write(ref _negotiatedPduSize, 0); // a stale post-teardown reading would read as "online" + } + finally + { + if (gated) { try { _gate.Release(); } catch (ObjectDisposedException) { } } + } } /// diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.TwinCAT/TwinCATDriver.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.TwinCAT/TwinCATDriver.cs index 86d7831e..5782300a 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.TwinCAT/TwinCATDriver.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.TwinCAT/TwinCATDriver.cs @@ -110,8 +110,12 @@ public sealed class TwinCATDriver : IDriver, IReadable, IWritable, ITagDiscovery public string DriverType => "TwinCAT"; /// - public Task InitializeAsync(string driverConfigJson, CancellationToken cancellationToken) + public async Task InitializeAsync(string driverConfigJson, CancellationToken cancellationToken) { + // #630: InitializeAsync is reachable on a live driver (DriverInstanceActor.AdoptConfigDuringInit + // runs it with no teardown first). Release the previous device states' probe loops + clients before + // the overwrite below. Deliberately NOT ShutdownAsync: that also drops _poll and the subscriptions. + await ReleaseDeviceStatesAsync().ConfigureAwait(false); _health = new DriverHealth(DriverState.Initializing, null, null); try { @@ -175,7 +179,30 @@ public sealed class TwinCATDriver : IDriver, IReadable, IWritable, ITagDiscovery _health = new DriverHealth(DriverState.Faulted, null, ex.Message); throw; } - return Task.CompletedTask; + } + + /// + /// Cancels, awaits and disposes every existing device state's probe loop, client and gate, then + /// clears the device table. Device part of only. + /// + private async Task ReleaseDeviceStatesAsync() + { + if (_devices.IsEmpty) return; + foreach (var state in _devices.Values) + { + try { state.ProbeCts?.Cancel(); } catch { } + if (state.ProbeTask is Task pt) + { + try { await pt.WaitAsync(TimeSpan.FromSeconds(5)).ConfigureAwait(false); } + catch { /* cancelled, faulted or wedged — teardown continues */ } + } + state.ProbeCts?.Dispose(); + state.ProbeCts = null; + state.ProbeTask = null; + state.DisposeClient(); + state.DisposeGate(); + } + _devices.Clear(); } /// 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]; diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.Tests/AbCipReinitializeReleasesStateTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.Tests/AbCipReinitializeReleasesStateTests.cs new file mode 100644 index 00000000..ebee2056 --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.Tests/AbCipReinitializeReleasesStateTests.cs @@ -0,0 +1,54 @@ +using Shouldly; +using Xunit; + +namespace ZB.MOM.WW.OtOpcUa.Driver.AbCip.Tests; + +/// +/// #630 (same pattern) — InitializeAsync on a live driver overwrote _devices[host] and +/// orphaned the previous state's probe loop + handles. It now releases them first. +/// +[Trait("Category", "Unit")] +public sealed class AbCipReinitializeReleasesStateTests +{ + /// Initialize twice: the first state's probe loop has stopped and its probe handle is disposed. + [Fact] + public async Task SecondInitialize_StopsFirstStatesLoop_AndDisposesItsHandles() + { + const string host = "ab://10.0.0.5/1,0"; + var factory = new FakeAbCipTagFactory(); + var drv = new AbCipDriver(new AbCipDriverOptions + { + Devices = [new AbCipDeviceOptions(host)], + Probe = new AbCipProbeOptions + { + Enabled = true, + Interval = TimeSpan.FromMilliseconds(20), + Timeout = TimeSpan.FromMilliseconds(200), + ProbeTagPath = "@raw_cpu_type", + }, + }, "drv-reinit", factory); + await drv.InitializeAsync("{}", CancellationToken.None); + + var first = drv.GetDeviceState(host)!; + // Positive control: the first probe loop is running and has initialised its runtime. + var deadline = DateTime.UtcNow.AddSeconds(5); + while (!(factory.Tags.TryGetValue("@raw_cpu_type", out var t) && t.InitializeCount > 0)) + { + if (DateTime.UtcNow > deadline) throw new TimeoutException("probe never started"); + await Task.Delay(10); + } + var firstTag = factory.Tags["@raw_cpu_type"]; + var firstLoop = first.ProbeTask; + firstLoop.ShouldNotBeNull(); + firstLoop!.IsCompleted.ShouldBeFalse(); + firstTag.Disposed.ShouldBeFalse(); + + await drv.InitializeAsync("{}", CancellationToken.None); + + firstLoop.IsCompleted.ShouldBeTrue(); // joined + firstTag.Disposed.ShouldBeTrue(); + drv.GetDeviceState(host).ShouldNotBeSameAs(first); + + await drv.ShutdownAsync(CancellationToken.None); + } +} diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbLegacy.Tests/AbLegacyReinitializeReleasesStateTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbLegacy.Tests/AbLegacyReinitializeReleasesStateTests.cs new file mode 100644 index 00000000..ffcce445 --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbLegacy.Tests/AbLegacyReinitializeReleasesStateTests.cs @@ -0,0 +1,50 @@ +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Driver.AbLegacy.PlcFamilies; + +namespace ZB.MOM.WW.OtOpcUa.Driver.AbLegacy.Tests; + +/// +/// #630 — InitializeAsync on a live driver (DriverInstanceActor.AdoptConfigDuringInit runs +/// it with no teardown first) overwrote _devices[host] and orphaned the previous state's +/// probe loop + runtime. It now releases them first. +/// +[Trait("Category", "Unit")] +public sealed class AbLegacyReinitializeReleasesStateTests +{ + /// Initialize twice: the first state's probe loop has stopped and its probe runtime is disposed. + [Fact] + public async Task SecondInitialize_StopsFirstStatesLoop_AndDisposesItsRuntime() + { + const string host = "ab://10.0.0.5/1,0"; + var factory = new FakeAbLegacyTagFactory(); + var drv = new AbLegacyDriver(new AbLegacyDriverOptions + { + Devices = [new AbLegacyDeviceOptions(host, AbLegacyPlcFamily.Slc500)], + Probe = new AbLegacyProbeOptions { Enabled = true, Interval = TimeSpan.FromMilliseconds(20), ProbeAddress = "S:0" }, + }, "drv-reinit", factory); + await drv.InitializeAsync("{}", CancellationToken.None); + + var first = drv.GetDeviceState(host)!; + // Positive control: the first state's probe loop is running and has created + initialised its runtime. + var deadline = DateTime.UtcNow.AddSeconds(5); + while (!(factory.Tags.TryGetValue("S:0", out var t) && t.InitializeCount > 0)) + { + if (DateTime.UtcNow > deadline) throw new TimeoutException("probe never started"); + await Task.Delay(10); + } + var firstTag = factory.Tags["S:0"]; + var firstLoop = first.ProbeTask; + firstLoop.ShouldNotBeNull(); + firstLoop!.IsCompleted.ShouldBeFalse(); + firstTag.Disposed.ShouldBeFalse(); + + await drv.InitializeAsync("{}", CancellationToken.None); + + firstLoop.IsCompleted.ShouldBeTrue(); // joined + firstTag.Disposed.ShouldBeTrue(); + drv.GetDeviceState(host).ShouldNotBeSameAs(first); + + await drv.ShutdownAsync(CancellationToken.None); + } +} diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.FOCAS.Tests/FocasReinitializeReleasesStateTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.FOCAS.Tests/FocasReinitializeReleasesStateTests.cs new file mode 100644 index 00000000..89b3cd5f --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.FOCAS.Tests/FocasReinitializeReleasesStateTests.cs @@ -0,0 +1,58 @@ +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Driver.FOCAS; + +namespace ZB.MOM.WW.OtOpcUa.Driver.FOCAS.Tests; + +/// +/// #630 — InitializeAsync on a live driver (DriverInstanceActor.AdoptConfigDuringInit runs +/// it with no teardown first) overwrote _devices[host] and orphaned the previous state's +/// probe loop + client. It now releases them first. +/// +[Trait("Category", "Unit")] +public sealed class FocasReinitializeReleasesStateTests +{ + /// Initialize twice: the first state's probe loop has finished and its client is disposed. + [Fact] + public async Task SecondInitialize_StopsFirstStatesLoop_AndDisposesItsClient() + { + const string host = "focas://10.0.0.5:8193"; + var factory = new FakeFocasClientFactory(); + var options = new FocasDriverOptions + { + Devices = [new FocasDeviceOptions(host)], + RawTags = [FocasRawTags.Entry(new FocasTagDefinition("R", host, "R100", FocasDataType.Byte))], + Probe = new FocasProbeOptions { Enabled = true, Interval = TimeSpan.FromMilliseconds(20), Timeout = TimeSpan.FromSeconds(2) }, + }; + var drv = new FocasDriver(options, "focas-reinit", factory); + await drv.InitializeAsync("{}", CancellationToken.None); + + var first = drv.GetDeviceState(host)!; + // Positive control: the first state's probe loop is demonstrably running and holds a live client. + await WaitUntilAsync(() => factory.Clients.Count > 0 && factory.Clients[0].IsConnected); + var firstLoop = first.ProbeTask; + firstLoop.ShouldNotBeNull(); + firstLoop!.IsCompleted.ShouldBeFalse(); + var firstClient = factory.Clients[0]; + + await drv.InitializeAsync("{}", CancellationToken.None); + + firstLoop.IsCompleted.ShouldBeTrue(); // joined + firstClient.DisposeCount.ShouldBeGreaterThanOrEqualTo(1); + firstClient.IsConnected.ShouldBeFalse(); + drv.GetDeviceState(host).ShouldNotBeSameAs(first); + drv.GetDeviceState(host)!.ProbeTask.ShouldNotBeNull(); // the new state is running + + await drv.ShutdownAsync(CancellationToken.None); + } + + private static async Task WaitUntilAsync(Func cond) + { + var deadline = DateTime.UtcNow.AddSeconds(5); + while (!cond()) + { + if (DateTime.UtcNow > deadline) throw new TimeoutException("condition not met"); + await Task.Delay(10); + } + } +} diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Tests/OpcUaClientReinitializeReleasesSessionTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Tests/OpcUaClientReinitializeReleasesSessionTests.cs new file mode 100644 index 00000000..6f260f00 --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Tests/OpcUaClientReinitializeReleasesSessionTests.cs @@ -0,0 +1,30 @@ +using Moq; +using Opc.Ua.Client; +using Shouldly; +using Xunit; + +namespace ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Tests; + +/// +/// #630 (same pattern) — InitializeAsync on a live driver (no ShutdownAsync first) used to +/// overwrite Session and orphan the previous session + its keep-alive handler. +/// +[Trait("Category", "Unit")] +public sealed class OpcUaClientReinitializeReleasesSessionTests +{ + /// A second Initialize disposes the live session before attempting to build a new one. + [Fact] + public async Task Initialize_OnLiveDriver_DisposesThePreviousSession() + { + var options = new OpcUaClientDriverOptions { EndpointUrl = "opc.tcp://127.0.0.1:1" }; + using var drv = new OpcUaClientDriver(options, "opcua-reinit"); + var old = new Mock(); + drv.SetSessionForTest(old.Object); + + // The new connect is refused (nothing listens on port 1) — what matters is what happened to the old one. + await Should.ThrowAsync(() => drv.InitializeAsync("{}", TestContext.Current.CancellationToken)); + + old.Verify(s => s.Dispose(), Times.AtLeastOnce); + drv.Session.ShouldBeNull(); + } +} diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.S7.Tests/S7DriverReconnectTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.S7.Tests/S7DriverReconnectTests.cs index 66417dcf..6f278194 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.S7.Tests/S7DriverReconnectTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.S7.Tests/S7DriverReconnectTests.cs @@ -482,6 +482,37 @@ public sealed class S7DriverReconnectTests (factory.Created.Count - atFirstAttempt).ShouldBeLessThanOrEqualTo(1); } + /// + /// #630 — a second InitializeAsync on a live driver (no ShutdownAsync between, as + /// DriverInstanceActor.AdoptConfigDuringInit does) must stop the first probe loop and + /// dispose the first connection rather than orphan them. + /// + [Fact] + public async Task Second_Initialize_stops_first_probe_loop_and_disposes_first_connection() + { + var factory = new FakeS7PlcFactory(); + using var drv = new S7Driver(ProbeEnabledOptions(), "s7-reinit", factory); + await drv.InitializeAsync("{}", TestContext.Current.CancellationToken); + var first = factory.Created[0]; + + // Positive control: the first probe loop is demonstrably running on the first connection. + await WaitUntilAsync(() => Volatile.Read(ref first.ReadStatusCount) >= 2, TimeSpan.FromSeconds(5)); + Volatile.Read(ref first.ReadStatusCount).ShouldBeGreaterThanOrEqualTo(2); + first.Disposed.ShouldBeFalse(); + + await drv.InitializeAsync("{}", TestContext.Current.CancellationToken); + + first.Disposed.ShouldBeTrue(); + var frozen = Volatile.Read(ref first.ReadStatusCount); + var second = factory.Created[^1]; + second.ShouldNotBeSameAs(first); + // Positive control for the absence check: the NEW loop keeps ticking over the same window. + var secondBefore = Volatile.Read(ref second.ReadStatusCount); + await WaitUntilAsync(() => Volatile.Read(ref second.ReadStatusCount) >= secondBefore + 3, TimeSpan.FromSeconds(5)); + Volatile.Read(ref second.ReadStatusCount).ShouldBeGreaterThanOrEqualTo(secondBefore + 3); + Volatile.Read(ref first.ReadStatusCount).ShouldBe(frozen); + } + private static S7DriverOptions ProbeEnabledOptions() => new() { Host = "192.0.2.1", @@ -626,8 +657,12 @@ public sealed class S7DriverReconnectTests public Task WriteBytesAsync(S7NetDataType area, int db, int startByteAdr, byte[] value, CancellationToken cancellationToken) => Task.CompletedTask; + /// Number of probe calls made on this connection. + public int ReadStatusCount; + public async Task ReadStatusAsync(CancellationToken cancellationToken) { + Interlocked.Increment(ref ReadStatusCount); if (ReadStatusHangsUntilCancelled) await Task.Delay(System.Threading.Timeout.Infinite, cancellationToken).ConfigureAwait(false); if (ReadStatusThrows is { } ex) diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.TwinCAT.Tests/TwinCATReinitializeReleasesStateTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.TwinCAT.Tests/TwinCATReinitializeReleasesStateTests.cs new file mode 100644 index 00000000..8ef5f9eb --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.TwinCAT.Tests/TwinCATReinitializeReleasesStateTests.cs @@ -0,0 +1,48 @@ +using Shouldly; +using Xunit; + +namespace ZB.MOM.WW.OtOpcUa.Driver.TwinCAT.Tests; + +/// +/// #630 (same pattern) — InitializeAsync on a live driver overwrote _devices[host] and +/// orphaned the previous state's probe loop + client. It now releases them first. +/// +[Trait("Category", "Unit")] +public sealed class TwinCATReinitializeReleasesStateTests +{ + /// Initialize twice: the first state's probe loop has finished and its client is disposed. + [Fact] + public async Task SecondInitialize_StopsFirstStatesLoop_AndDisposesItsClient() + { + const string host = "ads://5.23.91.23.1.1:851"; + var factory = new FakeTwinCATClientFactory(); + var drv = new TwinCATDriver(new TwinCATDriverOptions + { + Devices = [new TwinCATDeviceOptions(host)], + Probe = new TwinCATProbeOptions { Enabled = true, Interval = TimeSpan.FromMilliseconds(20), Timeout = TimeSpan.FromSeconds(1) }, + }, "drv-reinit", factory); + await drv.InitializeAsync("{}", CancellationToken.None); + + var first = drv.GetDeviceState(host)!; + // Positive control: the first probe loop is running and holds a connected client. + var deadline = DateTime.UtcNow.AddSeconds(5); + while (!(factory.Clients.Count > 0 && factory.Clients[0].IsConnected)) + { + if (DateTime.UtcNow > deadline) throw new TimeoutException("probe never connected"); + await Task.Delay(10); + } + var firstClient = factory.Clients[0]; + var firstLoop = first.ProbeTask; + firstLoop.ShouldNotBeNull(); + firstLoop!.IsCompleted.ShouldBeFalse(); + + await drv.InitializeAsync("{}", CancellationToken.None); + + firstLoop.IsCompleted.ShouldBeTrue(); // joined + firstClient.DisposeCount.ShouldBeGreaterThanOrEqualTo(1); + firstClient.IsConnected.ShouldBeFalse(); + drv.GetDeviceState(host).ShouldNotBeSameAs(first); + + await drv.ShutdownAsync(CancellationToken.None); + } +}