From 20cb2a52d644322755d42b1726f533f9b803504e Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Thu, 22 Oct 2026 14:02:00 -0400 Subject: [PATCH] fix(abcip): build-and-swap device tables, volatile health, async-only dispose H13: _devices/_devicesByRoutingKey are published by reference swap under a lifecycle lock and never mutated after publish, so the retained poll engine, discovery and the actor's GetHostStatuses pull never read a table mid-Clear. Every retire (init's own release, shutdown, dispose) takes the table it replaces, so ShutdownAsync racing DisposeAsync tears each state down once, and a generation check makes an InitializeAsync overtaken by a shutdown publish nothing (it throws OperationCanceledException) instead of resurrecting probe loops behind the shutdown. Probe join is one bounded 5 s wait independent of the caller token. H21: _health is volatile and read-modify-write updates go through a CAS. F10: the sync Dispose() (sync-over-async) is removed; DisposeAsync is the complete, idempotent teardown the runtime calls after ShutdownAsync. Refs #737 Claude-Session: https://claude.ai/code/session_01Ub9hGmYLQvn9XETzw7CMsT --- .../AbCipDriver.cs | 260 +++++++++++------- .../AbCipLifecycleConcurrencyTests.cs | 172 ++++++++++++ 2 files changed, 333 insertions(+), 99 deletions(-) create mode 100644 tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.Tests/AbCipLifecycleConcurrencyTests.cs 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 9f1d4fec..02c5f055 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip/AbCipDriver.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip/AbCipDriver.cs @@ -27,7 +27,7 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.AbCip; /// handles GC-collected tags. /// public sealed class AbCipDriver : IDriver, IReadable, IWritable, ITagDiscovery, IBatchSubscribable, - IHostConnectivityProbe, IPerCallHostResolver, IAlarmSource, ISkippedTagReporter, IDisposable, IAsyncDisposable + IHostConnectivityProbe, IPerCallHostResolver, IAlarmSource, ISkippedTagReporter, IAsyncDisposable { /// Authored tags this driver could not build into its poll table. private readonly SkippedTagLedger _skippedTags = new(); @@ -98,12 +98,21 @@ public sealed class AbCipDriver : IDriver, IReadable, IWritable, ITagDiscovery, private readonly PollGroupEngine _poll; // Devices keyed by host address — the primary registry (GetDeviceState, discovery, probe, diagnostics). - private readonly Dictionary _devices = new(StringComparer.OrdinalIgnoreCase); + // BUILD-AND-SWAP, never mutated after publish: the retained poll engine, the alarm projection, discovery and + // the actor's GetHostStatuses pull all read it while a re-init / shutdown runs. Each teardown takes the table + // it replaces, so concurrent teardowns (ShutdownAsync racing DisposeAsync) own disjoint device states. + private volatile Dictionary _devices = new(StringComparer.OrdinalIgnoreCase); // Tag-routing index: a tag's device routing key (DeviceName ?? HostAddress, plus the HostAddress alias) // → its DeviceState. A tag resolves to its device by DeviceName-or-host so multi-device routing keys on // the RawPath's device segment. The HostAddress alias is deliberate (Client CLI + tests route by host); - // _options.Devices is already the merged per-device DeviceConfig. - private readonly Dictionary _devicesByRoutingKey = new(StringComparer.OrdinalIgnoreCase); + // _options.Devices is already the merged per-device DeviceConfig. Swapped with _devices, same rule. + private volatile Dictionary _devicesByRoutingKey = new(StringComparer.OrdinalIgnoreCase); + + // Serialises table publish/retire. _lifecycleGeneration is bumped by every retire (an init's own release, a + // shutdown, a dispose); an init publishes only if no retire happened after its own, so an abandoned init that a + // shutdown overtook can never resurrect devices or probe loops after that shutdown returned. + private readonly object _lifecycleLock = new(); + private int _lifecycleGeneration; // 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 volatile Dictionary _tagsByRawPath = new(StringComparer.Ordinal); @@ -133,7 +142,24 @@ public sealed class AbCipDriver : IDriver, IReadable, IWritable, ITagDiscovery, /// when a definition exists for . internal bool TryGetTagDefinition(string rawPath, out AbCipTagDefinition definition) => _tagsByRawPath.TryGetValue(rawPath, out definition!); - private DriverHealth _health = new(DriverState.Unknown, null, null); + // volatile: written from the read batch, the write path, the poll engine's error hook and the lifecycle, and + // read from the actor thread (GetHealth). Read-modify-write updates go through UpdateHealth (CAS) so a + // concurrent update is never lost. + private volatile DriverHealth _health = new(DriverState.Unknown, null, null); + + /// Atomically replaces with applied to its current value. + /// Computes the new health from the current one. + private void UpdateHealth(Func update) + { + while (true) + { + var current = _health; + var next = update(current); +#pragma warning disable CS0420 // a volatile field passed by ref to Interlocked keeps its full-fence semantics + if (ReferenceEquals(Interlocked.CompareExchange(ref _health, next, current), current)) return; +#pragma warning restore CS0420 + } + } /// Occurs when a subscribed tag's value changes. public event EventHandler? OnDataChange; @@ -270,9 +296,9 @@ public sealed class AbCipDriver : IDriver, IReadable, IWritable, ITagDiscovery, internal void HandlePollError(Exception ex) { _logger.LogWarning(ex, "AbCip poll reader failed. Driver={DriverInstanceId}", _driverInstanceId); - var current = _health; - if (current.State != DriverState.Faulted) - _health = new DriverHealth(DriverState.Degraded, current.LastSuccessfulRead, ex.Message); + UpdateHealth(current => current.State == DriverState.Faulted + ? current + : new DriverHealth(DriverState.Degraded, current.LastSuccessfulRead, ex.Message)); } /// @@ -364,48 +390,58 @@ public sealed class AbCipDriver : IDriver, IReadable, IWritable, ITagDiscovery, public async Task InitializeAsync(string driverConfigJson, CancellationToken cancellationToken) { // InitializeAsync is reachable on a live driver with no teardown first (the IDriver contract - // allows it, and a reconnect retry can overlap a running attempt). 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); + // allows it, and a reconnect retry can overlap a running attempt). Retire the previous device states' probe + // loops + handles before building new ones. Deliberately NOT ShutdownAsync: that also drops _poll and the + // subscriptions. The generation this release produced is what the publish below is checked against. + var generation = await ReleaseDeviceStatesAsync().ConfigureAwait(false); + lock (_lifecycleLock) + { + if (generation == _lifecycleGeneration) + _health = new DriverHealth(DriverState.Initializing, null, null); + } _logger.LogInformation( "AbCip driver {DriverInstanceId} initializing: transport={Transport} tagFactory={TagFactory}", _driverInstanceId, _transportDescription, _tagFactory.GetType().Name); + Dictionary? builtDevices = null; try { + var options = _options; + var adoptedConfig = false; if (!string.IsNullOrWhiteSpace(driverConfigJson)) { var parsed = AbCipDriverFactoryExtensions.ParseOptions(_driverInstanceId, driverConfigJson); if (parsed.Devices.Count > 0 || parsed.RawTags.Count > 0) { - _options = parsed; - _alarmProjection = new AbCipAlarmProjection(this, _options.AlarmPollInterval, _logger); + options = parsed; + adoptedConfig = true; } } - foreach (var device in _options.Devices) + builtDevices = new Dictionary(StringComparer.OrdinalIgnoreCase); + var builtRoutes = new Dictionary(StringComparer.OrdinalIgnoreCase); + foreach (var device in options.Devices) { var addr = AbCipHostAddress.TryParse(device.HostAddress) ?? throw new InvalidOperationException( $"AbCip device has invalid HostAddress '{device.HostAddress}' — expected 'ab://gateway[:port]/cip-path'."); var profile = AbCipPlcFamilyProfile.ForFamily(device.PlcFamily); var state = new DeviceState(addr, device, profile); - _devices[device.HostAddress] = state; + builtDevices[device.HostAddress] = state; // Route index: a tag reaches this device by matching its DeviceName routing key OR its // host address. Devices with no explicit DeviceName are identified purely by host today. - _devicesByRoutingKey[device.HostAddress] = state; + builtRoutes[device.HostAddress] = state; if (!string.IsNullOrEmpty(device.DeviceName)) - _devicesByRoutingKey[device.DeviceName] = state; + builtRoutes[device.DeviceName] = state; } // Build the RawPath → definition table from the authored raw tags. Each entry's TagConfig blob // is mapped by the pure factory; the entry's DeviceName (routing key) + WriteIdempotent flag are // 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); + _skippedTags.BeginRebuild(options.RawTags.Count); var builtTagsByRawPath = new Dictionary(StringComparer.Ordinal); var builtDeclaredTags = new List(); - foreach (var entry in _options.RawTags) + foreach (var entry in options.RawTags) { if (!AbCipTagDefinitionFactory.FromTagConfig(entry.TagConfig, entry.RawPath, out var mapped)) { @@ -457,24 +493,43 @@ public sealed class AbCipDriver : IDriver, IReadable, IWritable, ITagDiscovery, } } } - // 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. - if (_options.Probe.Enabled && !string.IsNullOrWhiteSpace(_options.Probe.ProbeTagPath)) + // Publish everything under the lifecycle lock, and only if no shutdown / other init retired the tables + // after this init's own release — readers see the old tables or the new ones, never a half-built one, + // and an overtaken init leaves nothing running behind the shutdown that overtook it. + lock (_lifecycleLock) { - foreach (var state in _devices.Values) + if (generation != _lifecycleGeneration) + throw new OperationCanceledException( + $"AbCip driver {_driverInstanceId} initialization was superseded by a shutdown or a newer initialization."); + + _options = options; + if (adoptedConfig) + _alarmProjection = new AbCipAlarmProjection(this, options.AlarmPollInterval, _logger); + _devices = builtDevices; + _devicesByRoutingKey = builtRoutes; + _tagsByRawPath = builtTagsByRawPath; + _declaredTags = builtDeclaredTags; + _skippedTags.Commit(); + + // Probe loops — one per device when enabled + a ProbeTagPath is configured. Started inside the lock so + // a retire that follows this publish always sees (and joins) them. + if (options.Probe.Enabled && !string.IsNullOrWhiteSpace(options.Probe.ProbeTagPath)) { - state.ProbeCts = new CancellationTokenSource(); - var ct = state.ProbeCts.Token; - // Keep the loop Task so ShutdownAsync can await its clean exit before - // disposing the CTS / handles the loop is still using. - state.ProbeTask = Task.Run(() => ProbeLoopAsync(state, ct), ct); + foreach (var state in builtDevices.Values) + { + state.ProbeCts = new CancellationTokenSource(); + var ct = state.ProbeCts.Token; + // Keep the loop Task so a teardown can await its clean exit before + // disposing the CTS / handles the loop is still using. + state.ProbeTask = Task.Run(() => ProbeLoopAsync(state, ct), ct); + } } + _health = new DriverHealth(DriverState.Healthy, DateTime.UtcNow, null); } - else if (_options.Probe.Enabled && _devices.Count > 0) + builtDevices = null; // published: owned by the next retire now + + if (options.Probe.Enabled && string.IsNullOrWhiteSpace(options.Probe.ProbeTagPath) && options.Devices.Count > 0) { // Probe is Enabled but no ProbeTagPath is configured. Without a // tag path the loop has nothing to read, so HostState would stay Unknown forever @@ -486,51 +541,84 @@ public sealed class AbCipDriver : IDriver, IReadable, IWritable, ITagDiscovery, "as Unknown until a ProbeTagPath is set or Probe.Enabled is set to false.", _driverInstanceId); } - _health = new DriverHealth(DriverState.Healthy, DateTime.UtcNow, null); } catch (Exception ex) { - _health = new DriverHealth(DriverState.Faulted, null, ex.Message); + // Never-published states hold no probe loop and no runtime yet; dispose for symmetry. + if (builtDevices is not null) + foreach (var state in builtDevices.Values) state.DisposeHandles(); + if (ex is OperationCanceledException) + { + _logger.LogInformation("AbCip driver {DriverInstanceId} initialization abandoned: {Reason}", + _driverInstanceId, ex.Message); + throw; + } + lock (_lifecycleLock) + { + // A superseded init must not overwrite the health of whatever superseded it. + if (generation == _lifecycleGeneration) + _health = new DriverHealth(DriverState.Faulted, null, ex.Message); + } _logger.LogError(ex, "AbCip driver {DriverInstanceId} failed to initialize", _driverInstanceId); throw; } } + /// Bound on joining the retired probe loops — matches the fleet's "≤ 5 s join" teardown rule. + private static readonly TimeSpan ProbeJoinTimeout = TimeSpan.FromSeconds(5); + /// - /// Cancels, joins (bounded) and disposes every existing device state's probe loop + handles, then - /// clears the device/route tables. Device part of only. + /// Retires the published device states: swaps empty device/route/tag tables in under the lifecycle lock + /// (bumping the generation), then cancels, joins (bounded, independent of any caller token — a cancelled + /// caller must not dispose handles under a probe loop that is still running) and disposes the states it took. + /// Concurrent callers each take a disjoint table, so every state is torn down exactly once. The device part of + /// and the first step of . /// - private async Task ReleaseDeviceStatesAsync(CancellationToken cancellationToken) + /// The lifecycle generation this retire produced. + private async Task ReleaseDeviceStatesAsync() { - // 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) + Dictionary retired; + AbCipAlarmProjection projection; + int generation; + lock (_lifecycleLock) + { + generation = ++_lifecycleGeneration; + retired = _devices; + _devices = new Dictionary(StringComparer.OrdinalIgnoreCase); + _devicesByRoutingKey = new Dictionary(StringComparer.OrdinalIgnoreCase); + // Rebuilt wholesale by Initialize; swap, never Clear(): the retained poll engine / discovery may be enumerating. + _tagsByRawPath = new(StringComparer.Ordinal); + _declaredTags = new(); + projection = _alarmProjection; + } + + // The projection is replaced (not reused) when a config body is adopted; dispose it so its poll loop does + // not outlive it and raise duplicate alarm events through this driver. Idempotent, and reusable afterwards. + await projection.DisposeAsync().ConfigureAwait(false); + + foreach (var state in retired.Values) { try { state.ProbeCts?.Cancel(); } catch (ObjectDisposedException) { } } - foreach (var state in _devices.Values) + var loops = retired.Values.Select(s => s.ProbeTask).OfType().ToArray(); + if (loops.Length > 0) { - 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) { } + try { await Task.WhenAll(loops).WaitAsync(ProbeJoinTimeout).ConfigureAwait(false); } + catch (TimeoutException) + { + _logger.LogWarning("AbCip driver {DriverInstanceId}: a probe loop did not exit within {Timeout}; disposing anyway", + _driverInstanceId, ProbeJoinTimeout); + } + catch (Exception) { /* a loop never faults by design; nothing to add here */ } } - foreach (var state in _devices.Values) + foreach (var state in retired.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(); + return generation; } /// @@ -541,46 +629,17 @@ public sealed class AbCipDriver : IDriver, IReadable, IWritable, ITagDiscovery, } /// + /// + /// A complete teardown: the poll engine (and every subscription), the alarm projection, every probe loop + /// (joined ≤ 5 s) and every runtime handle. Safe to run concurrently with and with + /// an it overtakes (that init then publishes nothing). + /// public async Task ShutdownAsync(CancellationToken cancellationToken) { - await _alarmProjection.DisposeAsync().ConfigureAwait(false); await _poll.DisposeAsync().ConfigureAwait(false); - - // Phase 1: signal every probe loop to stop. - foreach (var state in _devices.Values) - { - try { state.ProbeCts?.Cancel(); } catch (ObjectDisposedException) { } - } - - // Phase 2: wait for each probe loop to observe cancellation and exit. The loop never - // throws on cancellation (it catches OperationCanceledException internally), but guard - // anyway so one slow device can't wedge the whole shutdown. - foreach (var state in _devices.Values) - { - var probeTask = state.ProbeTask; - if (probeTask is null) continue; - try - { - await probeTask.WaitAsync(TimeSpan.FromSeconds(10), cancellationToken).ConfigureAwait(false); - } - catch (TimeoutException) { } - catch (OperationCanceledException) { } - } - - // Phase 3: now the loops are gone, dispose the CTS + native handles with no live reader. - foreach (var state in _devices.Values) - { - state.ProbeCts?.Dispose(); - state.ProbeCts = null; - state.ProbeTask = null; - state.DisposeHandles(); - } - _devices.Clear(); - _devicesByRoutingKey.Clear(); - _tagsByRawPath = new(StringComparer.Ordinal); - _declaredTags = new(); + await ReleaseDeviceStatesAsync().ConfigureAwait(false); _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); + UpdateHealth(h => new DriverHealth(DriverState.Unknown, h.LastSuccessfulRead, null)); } // ---- ISubscribable (polling overlay via shared engine) ---- @@ -875,10 +934,10 @@ public sealed class AbCipDriver : IDriver, IReadable, IWritable, ITagDiscovery, // exactly as the per-item writes did before. if (worst is not { } worstState) return; - _health = new DriverHealth( + UpdateHealth(current => new DriverHealth( worstState, - anySuccess ? now : _health.LastSuccessfulRead, - worstState == DriverState.Healthy ? null : worstError); + anySuccess ? now : current.LastSuccessfulRead, + worstState == DriverState.Healthy ? null : worstError)); } private async Task ReadSingleAsync( @@ -1896,10 +1955,13 @@ public sealed class AbCipDriver : IDriver, IReadable, IWritable, ITagDiscovery, internal DeviceState? GetDeviceState(string hostAddress) => _devices.TryGetValue(hostAddress, out var s) ? s : null; - /// Releases all resources used by the driver. - public void Dispose() => DisposeAsync().AsTask().GetAwaiter().GetResult(); - - /// Asynchronously releases all resources used by the driver. + /// + /// Asynchronously releases all resources used by the driver — the same complete teardown as + /// . Idempotent, and safe CONCURRENTLY with a still-running + /// (the runtime's final stop abandons a shutdown that overruns its budget and + /// disposes anyway): each teardown retires a disjoint device table. There is deliberately no synchronous + /// Dispose() — it could only block on this teardown (sync-over-async). + /// /// A value task representing the asynchronous disposal. public async ValueTask DisposeAsync() { diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.Tests/AbCipLifecycleConcurrencyTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.Tests/AbCipLifecycleConcurrencyTests.cs new file mode 100644 index 00000000..00cb921a --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.AbCip.Tests/AbCipLifecycleConcurrencyTests.cs @@ -0,0 +1,172 @@ +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; + +namespace ZB.MOM.WW.OtOpcUa.Driver.AbCip.Tests; + +/// +/// Lifecycle overlap the runtime can produce (review 07 H13/F10, Lane A): the final stop runs +/// ShutdownAsync then DisposeAsync under one bounded budget, so a slow shutdown is abandoned and +/// the dispose runs CONCURRENTLY with it; an abandoned InitializeAsync can overlap the next attempt's +/// shutdown. Device tables are build-and-swapped, so each teardown owns a disjoint set of device states. +/// +[Trait("Category", "Unit")] +public sealed class AbCipLifecycleConcurrencyTests +{ + private const string Host = "ab://10.0.0.5/1,0"; + + /// A fake whose reads ignore cancellation for a while, so a teardown waits on the probe loop. + private sealed class SlowTag(AbCipTagCreateParams p, TimeSpan readDelay) : FakeAbCipTag(p) + { + private int _disposeCount; + + public int DisposeCount => Volatile.Read(ref _disposeCount); + + public override async Task ReadAsync(CancellationToken cancellationToken) + { + await base.ReadAsync(cancellationToken); + if (readDelay > TimeSpan.Zero) await Task.Delay(readDelay, CancellationToken.None); + } + + public override void Dispose() + { + Interlocked.Increment(ref _disposeCount); + base.Dispose(); + } + } + + private static (AbCipDriver Driver, FakeAbCipTagFactory Factory) Build(TimeSpan probeReadDelay) + { + var factory = new FakeAbCipTagFactory + { + Customise = p => new SlowTag(p, p.TagName == "ProbeTag" ? probeReadDelay : TimeSpan.Zero) { Value = 1 }, + }; + var drv = new AbCipDriver(new AbCipDriverOptions + { + Devices = [new AbCipDeviceOptions(Host)], + RawTags = AbCipRawTags.From(new AbCipTagDefinition("Speed", Host, "Motor1.Speed", AbCipDataType.DInt)), + Probe = new AbCipProbeOptions + { + Enabled = true, + ProbeTagPath = "ProbeTag", + Interval = TimeSpan.FromMilliseconds(10), + Timeout = TimeSpan.FromMilliseconds(200), + }, + }, "abcip-lifecycle", factory); + return (drv, factory); + } + + private static async Task WaitUntilAsync(Func condition, string what) + { + var deadline = DateTime.UtcNow.AddSeconds(5); + while (!condition()) + { + if (DateTime.UtcNow > deadline) throw new TimeoutException(what); + await Task.Delay(5); + } + } + + /// + /// Shutdown and Dispose racing each other tear the device down exactly once: the data runtime is + /// disposed once (not once per teardown), neither call throws, and the probe loop has exited. + /// + [Fact] + public async Task Dispose_concurrent_with_a_running_Shutdown_tears_each_state_down_exactly_once() + { + var (drv, factory) = Build(probeReadDelay: TimeSpan.FromMilliseconds(300)); + await drv.InitializeAsync("{}", CancellationToken.None); + (await drv.ReadAsync(["Speed"], CancellationToken.None))[0].StatusCode.ShouldBe(AbCipStatusMapper.Good); + await WaitUntilAsync(() => factory.Tags.ContainsKey("ProbeTag"), "probe never started"); + var state = drv.GetDeviceState(Host)!; + var probeLoop = state.ProbeTask!; + var data = (SlowTag)factory.Tags["Motor1.Speed"]; + + var shutdown = drv.ShutdownAsync(CancellationToken.None); + var dispose = drv.DisposeAsync().AsTask(); + await Task.WhenAll(shutdown, dispose); + + data.DisposeCount.ShouldBe(1); + probeLoop.IsCompleted.ShouldBeTrue(); + drv.GetDeviceState(Host).ShouldBeNull(); + } + + /// + /// An InitializeAsync that a ShutdownAsync overtakes publishes nothing: no device state and no probe + /// loop survive the shutdown. Positive control: an init that nothing overtakes still publishes. + /// + [Fact] + public async Task An_init_overtaken_by_Shutdown_publishes_no_device_and_starts_no_probe() + { + var (drv, factory) = Build(probeReadDelay: TimeSpan.FromMilliseconds(300)); + await drv.InitializeAsync("{}", CancellationToken.None); + await WaitUntilAsync(() => factory.Tags.ContainsKey("ProbeTag"), "probe never started"); + var firstProbe = factory.Tags["ProbeTag"]; + + // The second init's own release waits on the slow first probe loop; the shutdown overtakes it there. + var init = drv.InitializeAsync("{}", CancellationToken.None); + await Task.Delay(50); + await drv.ShutdownAsync(CancellationToken.None); + await Should.ThrowAsync(() => init); + + drv.GetDeviceState(Host).ShouldBeNull(); + drv.GetHostStatuses().ShouldBeEmpty(); + await Task.Delay(100); + factory.Tags["ProbeTag"].ShouldBeSameAs(firstProbe); // no second probe loop ever created a runtime + drv.GetHealth().State.ShouldNotBe(DriverState.Healthy); + + // Positive control: a later init that nothing overtakes publishes the device again. + await drv.InitializeAsync("{}", CancellationToken.None); + drv.GetDeviceState(Host).ShouldNotBeNull(); + drv.GetHealth().State.ShouldBe(DriverState.Healthy); + await drv.DisposeAsync(); + } + + /// + /// Readers of the device/routing tables (the retained poll engine, the actor's host-status pull) never + /// see a table mid-mutation while the driver re-initialises underneath them. + /// + [Fact] + public async Task Readers_survive_repeated_reinitialisation() + { + var (drv, _) = Build(probeReadDelay: TimeSpan.Zero); + await drv.InitializeAsync("{}", CancellationToken.None); + using var stop = new CancellationTokenSource(); + var reader = Task.Run(async () => + { + var reads = 0; + while (!stop.IsCancellationRequested) + { + _ = drv.GetHostStatuses(); + _ = drv.ResolveHost("Speed"); + await drv.ReadAsync(["Speed"], CancellationToken.None); + reads++; + } + return reads; + }); + + for (var i = 0; i < 50; i++) + await drv.InitializeAsync("{}", CancellationToken.None); + stop.Cancel(); + + (await reader).ShouldBeGreaterThan(0); // positive control: the reader actually ran + await drv.DisposeAsync(); + } + + /// F10: the driver exposes only an async dispose; a sync Dispose() that blocked on the async + /// teardown (sync-over-async) is gone, and the runtime takes the path. + [Fact] + public async Task Driver_is_async_disposable_only_and_DisposeAsync_stops_the_probe() + { + typeof(IDisposable).IsAssignableFrom(typeof(AbCipDriver)).ShouldBeFalse(); + var (drv, factory) = Build(probeReadDelay: TimeSpan.Zero); + await drv.InitializeAsync("{}", CancellationToken.None); + await WaitUntilAsync(() => factory.Tags.ContainsKey("ProbeTag"), "probe never started"); + var loop = drv.GetDeviceState(Host)!.ProbeTask!; + + await drv.DisposeAsync(); + await drv.DisposeAsync(); // idempotent + + loop.IsCompleted.ShouldBeTrue(); + factory.Tags["ProbeTag"].Disposed.ShouldBeTrue(); + } +}