From a32894e93b5eaa1339be6afa9f4976b4bf072b8d Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Thu, 22 Oct 2026 14:03:08 -0400 Subject: [PATCH] fix(focas): build-and-swap the device table; make the lifecycle overlap-safe (R07:H13) _devices was a plain Dictionary cleared and refilled in place by InitializeAsync / ShutdownAsync while the retained poll engine, the fixed-tree and alarm loops and the actor's health tick enumerated it ("Collection was modified"). It is now built whole and published by reference swap, and a release swaps it out to empty before tearing it down. Lane A's runtime overlaps lifecycle calls, so the driver now tolerates both shapes: - a lifecycle generation is bumped by every Initialize/Shutdown; an initialise that finds itself superseded (abandoned by the runtime, then a shutdown ran) publishes nothing, starts no loops, writes no health and throws OperationCanceledException; - a shutdown overlapping another (dispose beside a still-running shutdown) waits for the first's teardown via a release tail instead of returning on an empty table; - released devices are Retired: EnsureConnectedAsync refuses to reconnect them, and a connect that completes after retirement disposes what it opened. Clients are disposed under the device ConnectGate (bounded). Refs #740 Claude-Session: https://claude.ai/code/session_01Ub9hGmYLQvn9XETzw7CMsT --- .../FocasDriver.cs | 334 ++++++++++++------ .../FakeFocasClient.cs | 2 +- .../FocasDeviceTableLifecycleTests.cs | 200 +++++++++++ 3 files changed, 424 insertions(+), 112 deletions(-) create mode 100644 tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.FOCAS.Tests/FocasDeviceTableLifecycleTests.cs 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 cecc6299..f0c0226f 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.FOCAS/FocasDriver.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.FOCAS/FocasDriver.cs @@ -43,7 +43,40 @@ public sealed class FocasDriver : IDriver, IReadable, IWritable, ITagDiscovery, private readonly Func? _rebind; private readonly PollGroupEngine _poll; private readonly ILogger _logger; - private readonly Dictionary _devices = new(StringComparer.OrdinalIgnoreCase); + /// + /// Host address → device state. Built whole in and SWAPPED in (and swapped + /// out whole by a release), never cleared or refilled in place: the retained poll engine, the fixed-tree + /// and alarm loops and the actor's health tick all read it concurrently with a re-initialise, and a + /// mutated under an enumerator throws "Collection was modified". + /// Readers take one Volatile.Read snapshot and use it for the whole operation. + /// + private IReadOnlyDictionary _devices = EmptyDevices; + + private static readonly IReadOnlyDictionary EmptyDevices = + new Dictionary(StringComparer.OrdinalIgnoreCase); + + /// The live device table (one consistent snapshot). + private IReadOnlyDictionary Devices => Volatile.Read(ref _devices); + + /// + /// Guards the publish/retire of the device table against an overlapping lifecycle call. Held only for + /// synchronous swaps — never across an await, and nothing a loop runs ever takes it. + /// + private readonly object _lifecycleLock = new(); + + /// + /// Bumped by every and . An initialise that finds + /// the counter moved on by the time it would publish has been superseded (the runtime abandoned it and + /// tore the driver down) and must start nothing. + /// + private int _lifecycleGeneration; + + /// + /// Completes when every device table released so far has finished its teardown. A shutdown that overlaps + /// another (Lane A's dispose beside a still-running shutdown) awaits it, so it never returns while the + /// first one is still joining loops and disposing clients. + /// + private Task _releaseTail = Task.CompletedTask; // v3: RawPath → definition, case-sensitive ordinal. Built in InitializeAsync from _options.RawTags // via the pure FocasTagDefinitionFactory mapper (the reference is always a RawPath). // Built whole in InitializeAsync and SWAPPED in: the retained poll engine keeps calling @@ -196,9 +229,14 @@ public sealed class FocasDriver : IDriver, IReadable, IWritable, ITagDiscovery, // release the previous device states' loops + clients BEFORE they are overwritten (a live driver can be // re-initialised 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). + // + // The generation is taken FIRST: an initialise the runtime abandons (it timed out and moved on to a + // shutdown) is superseded by that shutdown's bump, and checks for it before publishing anything. + var generation = Interlocked.Increment(ref _lifecycleGeneration); await ReleaseDeviceStatesAsync().ConfigureAwait(false); ResetAlarmHealth(); - Volatile.Write(ref _health, new DriverHealth(DriverState.Initializing, null, null)); + WriteHealthIfCurrent(generation, new DriverHealth(DriverState.Initializing, null, null)); + var devices = new Dictionary(StringComparer.OrdinalIgnoreCase); try { // adopt a changed config. Options and the backend client factory are re-derived by ONE @@ -222,7 +260,7 @@ public sealed class FocasDriver : IDriver, IReadable, IWritable, ITagDiscovery, var addr = FocasHostAddress.TryParse(device.HostAddress) ?? throw new InvalidOperationException( $"FOCAS device has invalid HostAddress '{device.HostAddress}' — expected 'focas://{{ip}}[:{{port}}]'."); - _devices[device.HostAddress] = new DeviceState(addr, device); + devices[device.HostAddress] = new DeviceState(addr, device); } // Build the RawPath → definition table from the authored raw tags. Each entry's TagConfig // blob is mapped by the pure factory; the entry's WriteIdempotent flag + DeviceName are @@ -256,7 +294,7 @@ public sealed class FocasDriver : IDriver, IReadable, IWritable, ITagDiscovery, throw new InvalidOperationException( $"FOCAS tag '{entry.RawPath}' references device '{entry.DeviceName}' " + $"which is not in the Devices list. Check for a typo (e.g. wrong port or hostname)."); - if (!_devices.TryGetValue(deviceHost, out var device)) + if (!devices.TryGetValue(deviceHost, out var device)) throw new InvalidOperationException( $"FOCAS tag '{entry.RawPath}' references device '{entry.DeviceName}' " + $"which is not in the Devices list. Check for a typo (e.g. wrong port or hostname)."); @@ -309,75 +347,135 @@ public sealed class FocasDriver : IDriver, IReadable, IWritable, ITagDiscovery, // above; reusing the parsed record avoids per-tick allocs on subscription pollers. parsedAddresses[entry.RawPath] = parsed; } - // Publish the skip tally only once the whole table is built. An unknown-device throw above - // leaves the previous snapshot standing rather than a truncated one that would read as - // "we now skip fewer tags". - _skippedTags.Commit(); - Volatile.Write(ref _parsedAddressesByTagName, parsedAddresses); - Volatile.Write(ref _tagsByRawPath, tagsByRawPath); - Volatile.Write(ref _alarmTags, alarmTags); - - if (_options.Probe.Enabled) + // Publish everything at once, and only if no later lifecycle call has superseded this one. The check + // and the publish share the lock with the shutdown's swap-out, so either the shutdown sees (and + // releases) this table, or this initialise sees the shutdown and publishes nothing. + lock (_lifecycleLock) { - foreach (var state in _devices.Values) - { - state.ProbeCts = new CancellationTokenSource(); - var ct = state.ProbeCts.Token; - state.ProbeTask = Task.Run(() => ProbeLoopAsync(state, ct), ct); - } - } + if (generation != Volatile.Read(ref _lifecycleGeneration)) + throw new OperationCanceledException( + $"FOCAS initialise of {_driverInstanceId} was superseded by a later shutdown/initialise; nothing was started."); - if (_options.HandleRecycle.Enabled) - { - foreach (var state in _devices.Values) - { - state.RecycleCts = new CancellationTokenSource(); - var ct = state.RecycleCts.Token; - state.RecycleTask = Task.Run(() => RecycleLoopAsync(state, ct), ct); - } - } - - // See _fixedTreeActive: an authored FIXED: tag arms the loop on its own, because the - // config flag defaults off and the operator who committed those tags has already said - // what they want. - _fixedTreeActive = _options.FixedTree.Enabled - || _parsedAddressesByTagName.Values.Any(a => a.Kind == FocasAreaKind.Fixed); - if (_fixedTreeActive) - { - foreach (var state in _devices.Values) - { - state.FixedTreeCts = new CancellationTokenSource(); - var ct = state.FixedTreeCts.Token; - state.FixedTreeTask = Task.Run(() => FixedTreeLoopAsync(state, ct), ct); - } + // Publish the skip tally only once the whole table is built. An unknown-device throw above + // leaves the previous snapshot standing rather than a truncated one that would read as + // "we now skip fewer tags". + _skippedTags.Commit(); + Volatile.Write(ref _parsedAddressesByTagName, parsedAddresses); + Volatile.Write(ref _tagsByRawPath, tagsByRawPath); + Volatile.Write(ref _alarmTags, alarmTags); + // See _fixedTreeActive: an authored FIXED: tag arms the loop on its own, because the + // config flag defaults off and the operator who committed those tags has already said + // what they want. + _fixedTreeActive = _options.FixedTree.Enabled + || parsedAddresses.Values.Any(a => a.Kind == FocasAreaKind.Fixed); + Volatile.Write(ref _devices, devices); + StartDeviceLoops(devices.Values); } // Restart the alarm poll a re-initialise paused, against the tables just rebuilt. A // driver with no subscription yet starts nothing — the actor subscribes once it is Connected. await _alarmProjection.ResumeAsync().ConfigureAwait(false); + // A shutdown that landed while the resume ran owns the projection now: put it back the way it left it. + if (generation != Volatile.Read(ref _lifecycleGeneration)) + { + await _alarmProjection.PauseAsync().ConfigureAwait(false); + throw new OperationCanceledException( + $"FOCAS initialise of {_driverInstanceId} was superseded while resuming the alarm poll."); + } - Volatile.Write(ref _health, new DriverHealth(DriverState.Healthy, DateTime.UtcNow, null)); + WriteHealthIfCurrent(generation, new DriverHealth(DriverState.Healthy, DateTime.UtcNow, null)); + } + catch (OperationCanceledException) when (generation != Volatile.Read(ref _lifecycleGeneration)) + { + // Superseded: whatever this attempt built but never published is released here (a published table + // is the superseding call's to release); health belongs to that call too. + if (!ReferenceEquals(Devices, devices)) await ReleaseTableAsync(devices.Values).ConfigureAwait(false); + throw; } catch (Exception ex) { - Volatile.Write(ref _health, new DriverHealth(DriverState.Faulted, null, ex.Message)); + WriteHealthIfCurrent(generation, new DriverHealth(DriverState.Faulted, null, ex.Message)); throw; } } + /// Writes unless a later lifecycle call has taken over. + private void WriteHealthIfCurrent(int generation, DriverHealth health) + { + lock (_lifecycleLock) + { + if (generation == Volatile.Read(ref _lifecycleGeneration)) Volatile.Write(ref _health, health); + } + } + + /// Starts the probe / recycle / fixed-tree loops for a freshly built table. Caller holds the lifecycle lock. + private void StartDeviceLoops(IEnumerable states) + { + foreach (var state in states) + { + if (_options.Probe.Enabled) + { + state.ProbeCts = new CancellationTokenSource(); + var ct = state.ProbeCts.Token; + state.ProbeTask = Task.Run(() => ProbeLoopAsync(state, ct), ct); + } + if (_options.HandleRecycle.Enabled) + { + state.RecycleCts = new CancellationTokenSource(); + var ct = state.RecycleCts.Token; + state.RecycleTask = Task.Run(() => RecycleLoopAsync(state, ct), ct); + } + if (_fixedTreeActive) + { + state.FixedTreeCts = new CancellationTokenSource(); + var ct = state.FixedTreeCts.Token; + state.FixedTreeTask = Task.Run(() => FixedTreeLoopAsync(state, ct), ct); + } + } + } + /// - /// 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). + /// Swaps the live device table out (to empty) and releases it: cancels, bounded-joins and disposes every + /// device state's loops and client, after pausing the alarm projection. Device part of + /// only (no poll / subscription teardown). Also waits for any release an + /// overlapping lifecycle call started earlier, so it never returns while that one is still running. /// private async Task ReleaseDeviceStatesAsync() { // Pause, not dispose: the subscription and what the server was told survive the rebuild. await _alarmProjection.PauseAsync().ConfigureAwait(false); - if (_devices.Count == 0) return; - var loops = new List(); - foreach (var state in _devices.Values) + IReadOnlyDictionary old; + Task prior; + var done = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + lock (_lifecycleLock) { + old = _devices; + Volatile.Write(ref _devices, EmptyDevices); + prior = _releaseTail; + _releaseTail = done.Task; + } + try + { + if (old.Count > 0) await ReleaseTableAsync(old.Values).ConfigureAwait(false); + await prior.ConfigureAwait(false); + } + finally { done.TrySetResult(); } + } + + /// + /// Retires and tears down one swapped-out table: retire every device first (so a connect still in flight + /// disposes what it opens), cancel every loop, join them (bounded), then dispose the CTSs and — under each + /// device's connect gate — its client. + /// + private async Task ReleaseTableAsync(IEnumerable table) + { + var states = table.ToList(); + var loops = new List(); + foreach (var state in states) + { + state.Retire(); + // Cancel-then-dispose can race in tight shutdown loops; swallowing is intentional + // but we log the cause so a noisy shutdown leaves a Debug trace. try { state.ProbeCts?.Cancel(); } catch (Exception ex) { _logger.LogDebug(ex, "Cancelling probe CTS for {Host} failed", state.Options.HostAddress); } try { state.RecycleCts?.Cancel(); } @@ -388,15 +486,18 @@ public sealed class FocasDriver : IDriver, IReadable, IWritable, ITagDiscovery, if (state.RecycleTask is { } rt) loops.Add(rt); if (state.FixedTreeTask is { } ft) loops.Add(ft); } + // Cancel every loop first, JOIN them (bounded), and only then dispose the CTSs and the + // client. Fire-and-forget loops used to outlive shutdown: one parked in EnsureConnectedAsync could + // finish its connect AFTER DisposeClient and leave a live socket on a shut-down driver. 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", + _logger.LogWarning("FOCAS teardown: {Count} background loop(s) did not stop within {Timeout}; proceeding", loops.Count, LoopJoinTimeout); } - foreach (var state in _devices.Values) + foreach (var state in states) { state.ProbeCts?.Dispose(); state.ProbeCts = null; @@ -407,9 +508,22 @@ public sealed class FocasDriver : IDriver, IReadable, IWritable, ITagDiscovery, state.FixedTreeCts?.Dispose(); state.FixedTreeCts = null; state.FixedTreeTask = null; - state.DisposeClient(); + await DisposeClientGatedAsync(state).ConfigureAwait(false); } - _devices.Clear(); + } + + /// + /// Disposes a device's client under its , so it can neither null a + /// client a concurrent just installed nor race that connect. The wait + /// is bounded: a connect wedged past may not wedge teardown too (the device + /// is retired, so that connect disposes its own client when it finally returns). + /// + private async Task DisposeClientGatedAsync(DeviceState state) + { + var entered = await state.ConnectGate.WaitAsync(LoopJoinTimeout).ConfigureAwait(false); + try { state.DisposeClient(); } + catch (Exception ex) { _logger.LogDebug(ex, "FOCAS client dispose failed for {Host}", state.Options.HostAddress); } + finally { if (entered) state.ConnectGate.Release(); } } /// @@ -423,58 +537,28 @@ public sealed class FocasDriver : IDriver, IReadable, IWritable, ITagDiscovery, internal static readonly TimeSpan LoopJoinTimeout = TimeSpan.FromSeconds(5); /// + // Safe to overlap with itself (Lane A disposes beside a still-running shutdown — the second call waits for + // the first's teardown via _releaseTail) and with an abandoned InitializeAsync (the generation bump below + // supersedes it, so it publishes nothing). public async Task ShutdownAsync(CancellationToken cancellationToken) { + var generation = Interlocked.Increment(ref _lifecycleGeneration); await _poll.DisposeAsync().ConfigureAwait(false); - // Pause, not dispose — see _alarmProjection. A later InitializeAsync resumes it. - await _alarmProjection.PauseAsync().ConfigureAwait(false); - // cancel every loop first, JOIN them (bounded), and only then dispose the CTSs and the - // client. Fire-and-forget loops used to outlive shutdown: one parked in EnsureConnectedAsync could - // finish its connect AFTER DisposeClient and leave a live socket on a shut-down driver. - var loops = new List(); - foreach (var state in _devices.Values) + // Pauses (not disposes) the alarm projection — see _alarmProjection; a later InitializeAsync resumes it. + await ReleaseDeviceStatesAsync().ConfigureAwait(false); + lock (_lifecycleLock) { - // Cancel-then-dispose can race in tight shutdown loops; swallowing is intentional - // but we now log the cause so a noisy shutdown leaves a Debug trace. - 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); + // A later initialise owns the tables (and health) now — leave them alone. + if (generation != Volatile.Read(ref _lifecycleGeneration)) return; + Volatile.Write(ref _tagsByRawPath, new Dictionary(StringComparer.Ordinal)); + Volatile.Write(ref _alarmTags, new Dictionary(StringComparer.Ordinal)); + Volatile.Write(ref _parsedAddressesByTagName, + new ConcurrentDictionary(StringComparer.OrdinalIgnoreCase)); + _resolver.Clear(); // no-op in v3 (the resolver reads the live table by closure); kept for symmetry + _fixedTreeActive = false; + ResetAlarmHealth(); + Volatile.Write(ref _health, new DriverHealth(DriverState.Unknown, Volatile.Read(ref _health).LastSuccessfulRead, null)); } - 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 shutdown: {Count} background loop(s) did not stop within {Timeout}; proceeding with teardown", - 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(); - Volatile.Write(ref _tagsByRawPath, new Dictionary(StringComparer.Ordinal)); - Volatile.Write(ref _alarmTags, new Dictionary(StringComparer.Ordinal)); - _resolver.Clear(); // no-op in v3 (the resolver reads the live table by closure); kept for symmetry - _parsedAddressesByTagName.Clear(); - _fixedTreeActive = false; - ResetAlarmHealth(); - Volatile.Write(ref _health, new DriverHealth(DriverState.Unknown, Volatile.Read(ref _health).LastSuccessfulRead, null)); } /// @@ -499,12 +583,14 @@ public sealed class FocasDriver : IDriver, IReadable, IWritable, ITagDiscovery, } /// Gets the number of configured devices. - internal int DeviceCount => _devices.Count; + internal int DeviceCount => Devices.Count; + /// Test seam — the live device table snapshot (swapped whole, never mutated in place). + internal IReadOnlyDictionary DeviceTable => Devices; /// Gets the state of a device by host address. /// The host address of the device. /// The device state if found; otherwise null. internal DeviceState? GetDeviceState(string hostAddress) => - _devices.TryGetValue(hostAddress, out var s) ? s : null; + Devices.TryGetValue(hostAddress, out var s) ? s : null; /// Test seam — the resolved options the factory built this driver from. internal FocasDriverOptions Options => _options; @@ -592,7 +678,7 @@ public sealed class FocasDriver : IDriver, IReadable, IWritable, ITagDiscovery, results[i] = new DataValueSnapshot(null, FocasStatusMapper.BadNodeIdUnknown, null, now); continue; } - if (!_devices.TryGetValue(def.DeviceHostAddress, out var device)) + if (!Devices.TryGetValue(def.DeviceHostAddress, out var device)) { results[i] = new DataValueSnapshot(null, FocasStatusMapper.BadNodeIdUnknown, null, now); continue; @@ -680,7 +766,7 @@ public sealed class FocasDriver : IDriver, IReadable, IWritable, ITagDiscovery, var reference = references[i]; if (_fixedTreeActive && TryReadFixedTree(reference, now) is not null) continue; if (!_resolver.TryResolve(reference, out var def)) continue; - if (!_devices.TryGetValue(def.DeviceHostAddress, out var device)) continue; + if (!Devices.TryGetValue(def.DeviceHostAddress, out var device)) continue; var parsed = ResolveParsedAddress(def); if (parsed.Kind != FocasAreaKind.Pmc || parsed.PmcLetter is null) continue; (byDevice ??= []).TryAdd(device, []); @@ -802,7 +888,7 @@ public sealed class FocasDriver : IDriver, IReadable, IWritable, ITagDiscovery, results[i] = new WriteResult(FocasStatusMapper.BadNotWritable); continue; } - if (!_devices.TryGetValue(def.DeviceHostAddress, out var device)) + if (!Devices.TryGetValue(def.DeviceHostAddress, out var device)) { results[i] = new WriteResult(FocasStatusMapper.BadNodeIdUnknown); continue; @@ -890,7 +976,7 @@ public sealed class FocasDriver : IDriver, IReadable, IWritable, ITagDiscovery, // Fixed-tree T1 — Identity + Axes subtrees, populated once per session // from cnc_sysinfo + cnc_rdaxisname at init time and kept in DeviceState. if (_fixedTreeActive - && _devices.TryGetValue(device.HostAddress, out var state) + && Devices.TryGetValue(device.HostAddress, out var state) && state.FixedTreeCache is { } cache) { var identity = deviceFolder.Folder("Identity", "Identity"); @@ -1142,7 +1228,7 @@ public sealed class FocasDriver : IDriver, IReadable, IWritable, ITagDiscovery, /// public IReadOnlyList GetHostStatuses() => - [.. _devices.Values.Select(s => new HostConnectivityStatus(s.Options.HostAddress, s.HostState, s.HostStateChangedUtc))]; + [.. Devices.Values.Select(s => new HostConnectivityStatus(s.Options.HostAddress, s.HostState, s.HostStateChangedUtc))]; private async Task ProbeLoopAsync(DeviceState state, CancellationToken ct) { @@ -1510,7 +1596,7 @@ public sealed class FocasDriver : IDriver, IReadable, IWritable, ITagDiscovery, /// private DataValueSnapshot? TryReadFixedTree(string reference, DateTime now) { - foreach (var state in _devices.Values) + foreach (var state in Devices.Values) { if (!reference.StartsWith(state.Options.HostAddress + "/", StringComparison.OrdinalIgnoreCase)) continue; if (TryReadFixedTreeOnDevice(state, reference[(state.Options.HostAddress.Length + 1)..], now) is { } hit) @@ -1776,7 +1862,7 @@ public sealed class FocasDriver : IDriver, IReadable, IWritable, ITagDiscovery, /// The device's active alarms. internal async Task> ReadDeviceAlarmsAsync(string hostAddress, CancellationToken ct) { - if (!_devices.TryGetValue(hostAddress, out var state)) + if (!Devices.TryGetValue(hostAddress, out var state)) throw new InvalidOperationException($"FOCAS device '{hostAddress}' is not configured."); var client = await EnsureConnectedAsync(state, ct).ConfigureAwait(false); return await client.ReadAlarmsAsync(ct).ConfigureAwait(false); @@ -1822,6 +1908,12 @@ public sealed class FocasDriver : IDriver, IReadable, IWritable, ITagDiscovery, await device.ConnectGate.WaitAsync(ct).ConfigureAwait(false); try { + // A released device (re-initialise / shutdown) is never reconnected: whoever still holds it read + // the table before the swap, and a client opened now would be orphaned on a state nobody tears down. + if (device.IsRetired) + throw new ObjectDisposedException(nameof(DeviceState), + $"FOCAS device {device.Options.HostAddress} was released by a re-initialise or shutdown."); + // Another caller may have connected while this one waited. if (device.Client is { IsConnected: true } connected) return connected; @@ -1864,6 +1956,15 @@ public sealed class FocasDriver : IDriver, IReadable, IWritable, ITagDiscovery, throw; } device.Backoff.RecordSuccess(); // recovery resets the window immediately + // Retired WHILE this connect ran (the release waits on this gate only up to its bound): the connect + // is ours to undo, or it leaves a live session on a shut-down driver. + if (device.IsRetired) + { + client.Dispose(); + if (ReferenceEquals(device.Client, client)) device.Client = null; + throw new ObjectDisposedException(nameof(DeviceState), + $"FOCAS device {device.Options.HostAddress} was released while connecting."); + } return client; } finally { device.ConnectGate.Release(); } @@ -2020,6 +2121,17 @@ public sealed class FocasDriver : IDriver, IReadable, IWritable, ITagDiscovery, /// public ConcurrentDictionary LastSpindleMotorSpeeds { get; } = new(); + private int _retired; + + /// + /// True once a re-initialise / shutdown has released this device. A retired device never connects + /// again, and a connect that was already running when it retired disposes what it opened. + /// + public bool IsRetired => Volatile.Read(ref _retired) != 0; + + /// Marks the device released. Idempotent. + public void Retire() => Volatile.Write(ref _retired, 1); + /// Disposes the FOCAS client instance. public void DisposeClient() { diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.FOCAS.Tests/FakeFocasClient.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.FOCAS.Tests/FakeFocasClient.cs index 00cf61d5..370bff15 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.FOCAS.Tests/FakeFocasClient.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.FOCAS.Tests/FakeFocasClient.cs @@ -259,7 +259,7 @@ internal sealed class FakeFocasClientFactory : IFocasClientFactory public IFocasClient Create() { var c = Customise?.Invoke() ?? new FakeFocasClient(); - Clients.Add(c); + lock (Clients) Clients.Add(c); return c; } } diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.FOCAS.Tests/FocasDeviceTableLifecycleTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.FOCAS.Tests/FocasDeviceTableLifecycleTests.cs new file mode 100644 index 00000000..056b7ca8 --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.FOCAS.Tests/FocasDeviceTableLifecycleTests.cs @@ -0,0 +1,200 @@ +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; +using ZB.MOM.WW.OtOpcUa.Driver.FOCAS; + +namespace ZB.MOM.WW.OtOpcUa.Driver.FOCAS.Tests; + +/// +/// #740 / R07:H13 — the device table is read by the retained poll engine, the fixed-tree loop, the alarm +/// loop and the actor's health tick, so a re-initialise must build a fresh table and swap it in rather than +/// clearing and refilling the live one. Also pins the lifecycle overlaps the runtime produces since Lane A: +/// a dispose running beside a still-running shutdown, and an abandoned initialise overlapping the next +/// attempt's shutdown. +/// +[Trait("Category", "Unit")] +public sealed class FocasDeviceTableLifecycleTests +{ + private const string Host = "focas://10.0.0.5:8193"; + + private static FocasDriverOptions Options(bool probe = false, bool fixedTree = false) => new() + { + Devices = [new FocasDeviceOptions(Host)], + RawTags = [FocasRawTags.Entry(new FocasTagDefinition("raw/R", Host, "R100", FocasDataType.Byte))], + Probe = new FocasProbeOptions { Enabled = probe, Interval = TimeSpan.FromMilliseconds(20) }, + FixedTree = new FocasFixedTreeOptions { Enabled = fixedTree, PollInterval = TimeSpan.FromMilliseconds(10) }, + }; + + /// A re-initialise publishes a NEW table; a reader holding the old one never sees it emptied. + [Fact] + public async Task Reinitialize_swaps_in_a_new_table_and_never_mutates_the_one_a_reader_holds() + { + var drv = new FocasDriver(Options(), "focas-swap", new FakeFocasClientFactory()); + await drv.InitializeAsync("{}", CancellationToken.None); + var before = drv.DeviceTable; + before.Count.ShouldBe(1); + + await drv.InitializeAsync("{}", CancellationToken.None); + + drv.DeviceTable.ShouldNotBeSameAs(before); + drv.DeviceTable.Count.ShouldBe(1); + before.Count.ShouldBe(1, "the table a concurrent reader enumerates must not be cleared under it"); + await drv.ShutdownAsync(CancellationToken.None); + before.Count.ShouldBe(1); + drv.DeviceTable.Count.ShouldBe(0); + } + + /// Hammering every table reader during repeated re-initialises throws nothing — and reads still succeed. + [Fact] + public async Task Readers_survive_repeated_reinitialise() + { + var factory = new FakeFocasClientFactory(); + var drv = new FocasDriver(Options(fixedTree: true), "focas-hammer", factory); + await drv.InitializeAsync("{}", CancellationToken.None); + + using var stop = new CancellationTokenSource(); + var good = 0; + var errors = new List(); + var reader = Task.Run(async () => + { + while (!stop.IsCancellationRequested) + { + try + { + _ = drv.GetHostStatuses(); + var snaps = await drv.ReadAsync(["raw/R", $"{Host}/Identity/SeriesNumber"], CancellationToken.None); + if (snaps[0].StatusCode == FocasStatusMapper.Good) Interlocked.Increment(ref good); + } + catch (Exception ex) { lock (errors) errors.Add(ex); } + } + }); + + for (var i = 0; i < 100; i++) + await drv.InitializeAsync("{}", CancellationToken.None); + await Task.Delay(100); + stop.Cancel(); + await reader; + await drv.ShutdownAsync(CancellationToken.None); + + errors.ShouldBeEmpty(); + good.ShouldBeGreaterThan(0, "positive control: the reader really did read through the table"); + } + + /// A client whose connect parks until released and ignores cancellation. + private sealed class ParkedConnectClient(TaskCompletionSource entered, Task release) : FakeFocasClient + { + public override async Task ConnectAsync(FocasHostAddress address, TimeSpan timeout, CancellationToken ct) + { + entered.TrySetResult(); + await release.ConfigureAwait(false); + await base.ConnectAsync(address, timeout, CancellationToken.None).ConfigureAwait(false); + } + } + + /// + /// A data-path read parked in connect when shutdown runs must not leave a live client on the released + /// device: the device is retired, so the connect that completes afterwards disposes what it opened. + /// + [Fact] + public async Task Connect_completing_after_shutdown_does_not_leave_a_live_client() + { + var entered = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var release = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var factory = new FakeFocasClientFactory { Customise = () => new ParkedConnectClient(entered, release.Task) }; + var drv = new FocasDriver(Options(), "focas-retire", factory); + await drv.InitializeAsync("{}", CancellationToken.None); + + var read = drv.ReadAsync(["raw/R"], CancellationToken.None); + await entered.Task.WaitAsync(TimeSpan.FromSeconds(5)); + + var shutdown = drv.ShutdownAsync(CancellationToken.None); + await Task.Delay(50); + release.SetResult(); + await shutdown.WaitAsync(TimeSpan.FromSeconds(10)); + var snaps = await read.WaitAsync(TimeSpan.FromSeconds(5)); + + factory.Clients.Count.ShouldBe(1); + factory.Clients[0].IsConnected.ShouldBeFalse("a connect finishing on a retired device must dispose its client"); + factory.Clients[0].DisposeCount.ShouldBeGreaterThanOrEqualTo(1); + snaps[0].StatusCode.ShouldBe(FocasStatusMapper.BadCommunicationError); + } + + /// + /// An initialise abandoned mid-way (the runtime timed it out and moved on to a shutdown) must not + /// resurrect the driver when it finally completes: no device table, no loops, no connection. + /// + [Fact] + public async Task Abandoned_initialise_finishing_after_shutdown_starts_nothing() + { + var factory = new FakeFocasClientFactory(); + var options = Options(probe: true); + using var gate = new ManualResetEventSlim(false); + var inRebind = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var drv = new FocasDriver(options, "focas-abandoned", factory, rebind: _ => + { + inRebind.TrySetResult(); + gate.Wait(TimeSpan.FromSeconds(10)); + return new FocasDriverBinding(options, factory); + }); + + var init = Task.Run(() => drv.InitializeAsync("{\"devices\":[]}", CancellationToken.None)); + await inRebind.Task.WaitAsync(TimeSpan.FromSeconds(5)); + await drv.ShutdownAsync(CancellationToken.None); + gate.Set(); + await Should.ThrowAsync(() => init.WaitAsync(TimeSpan.FromSeconds(10))); + + await Task.Delay(200); // several probe intervals — a started loop would have connected by now + drv.DeviceCount.ShouldBe(0); + factory.Clients.ShouldBeEmpty(); + drv.GetHealth().State.ShouldNotBe(DriverState.Healthy); + + // Positive control: the same driver, initialised without being superseded, does start its probe. + await drv.InitializeAsync("{\"devices\":[]}", CancellationToken.None); + await WaitUntil(() => factory.Clients.Count > 0); + drv.DeviceCount.ShouldBe(1); + drv.GetHealth().State.ShouldBe(DriverState.Healthy); + await drv.DisposeAsync(); + } + + /// A loop wedged in a read that ignores cancellation for a while. + private sealed class SlowProbeClient : FakeFocasClient + { + public override async Task ProbeAsync(CancellationToken ct) + { + await Task.Delay(400, CancellationToken.None).ConfigureAwait(false); + return true; + } + } + + /// + /// Lane A disposes while its own shutdown may still be running. The dispose must not return before the + /// teardown it overlaps has released the client — it waits for it rather than finding an empty table. + /// + [Fact] + public async Task Dispose_concurrent_with_shutdown_waits_for_the_teardown() + { + var factory = new FakeFocasClientFactory { Customise = () => new SlowProbeClient() }; + var drv = new FocasDriver(Options(probe: true), "focas-dispose", factory); + await drv.InitializeAsync("{}", CancellationToken.None); + await WaitUntil(() => factory.Clients.Count > 0 && factory.Clients[0].IsConnected); + + var shutdown = drv.ShutdownAsync(CancellationToken.None); + var dispose = drv.DisposeAsync().AsTask(); + await dispose.WaitAsync(TimeSpan.FromSeconds(10)); + + factory.Clients.ShouldAllBe(c => !c.IsConnected); + factory.Clients.ShouldAllBe(c => c.DisposeCount >= 1); + await shutdown.WaitAsync(TimeSpan.FromSeconds(10)); + drv.DeviceCount.ShouldBe(0); + } + + private static async Task WaitUntil(Func condition) + { + var deadline = DateTime.UtcNow + TimeSpan.FromSeconds(5); + while (!condition()) + { + if (DateTime.UtcNow > deadline) throw new TimeoutException("condition not reached"); + await Task.Delay(10); + } + } +}