diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/GalaxyDriver.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/GalaxyDriver.cs index 3216e136..a75ba588 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/GalaxyDriver.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/GalaxyDriver.cs @@ -146,9 +146,61 @@ public sealed class GalaxyDriver private HostConnectivityForwarder? _transportForwarder; private PerPlatformProbeWatcher? _probeWatcher; + // Injected deploy-watch seam (tests). Null in production, where StartDeployWatcherAsync builds a + // GatewayGalaxyDeployWatchSource over the owned GalaxyRepositoryClient. + private readonly IGalaxyDeployWatchSource? _deployWatchSource; + + // True while the seams (_subscriber / _dataWriter / _hierarchySource / _alarmFeed / + // _alarmAcknowledger) were built by BuildProductionRuntimeAsync and are therefore ours to dispose + // and null on ShutdownAsync — the host's comms-loss recovery is ShutdownAsync → InitializeAsync on + // this SAME object, and InitializeAsync rebuilds only when the seams are null. Injected seams + // belong to the injector: ShutdownAsync stops every loop it started on top of them but leaves the + // seams in place, so a seam-driven test can cycle the lifecycle the way the watchdog does. + private bool _ownsRuntime; + + // Serialises InitializeAsync / ShutdownAsync / DisposeAsync so a late watchdog re-init can never + // interleave with a teardown and resurrect half of the runtime it is tearing down. + private readonly SemaphoreSlim _lifecycleGate = new(1, 1); + + // Set for the duration of a runtime teardown so the pump's stream-fault / replay-gap callbacks and + // the supervisor's replay cannot start a fresh pump underneath it. + private volatile bool _runtimeStopping; + private DriverHealth _health = new(DriverState.Unknown, null, null); private bool _disposed; + /// + /// Test seam — forwarded to on the + /// session creates, so the owned-runtime lifecycle + /// (client + session construction, teardown, rebuild) can be exercised without a gateway. The + /// gRPC channels it builds are real but lazy: nothing dials until a call is made. + /// + internal Func? SessionOpenOverrideForTests { get; set; } + + /// Test seam: the owned gateway client, null when no production runtime is live. + internal MxGatewayClient? OwnedMxClientForTests => _ownedMxClient; + + /// Test seam: the owned MX session, null when no production runtime is live. + internal GalaxyMxSession? OwnedMxSessionForTests => _ownedMxSession; + + /// Test seam: the owned repository client, null until the deploy watcher or discovery built it. + internal GalaxyRepositoryClient? OwnedRepositoryClientForTests => _ownedRepositoryClient; + + /// Test seam: the live reconnect supervisor, null outside a production runtime. + internal ReconnectSupervisor? SupervisorForTests => _supervisor; + + /// Test seam: the live per-platform probe watcher, null until InitializeAsync. + internal PerPlatformProbeWatcher? ProbeWatcherForTests => _probeWatcher; + + /// Test seam: the live deploy watcher, null until InitializeAsync starts one. + internal DeployWatcher? DeployWatcherForTests => _deployWatcher; + + /// Test seam: true while an is running. + internal bool HasEventPumpForTests { get { lock (_pumpLock) return _eventPump is not null; } } + + /// Test seam: the number of driver subscriptions the registry still tracks. + internal int TrackedSubscriptionCountForTests => _subscriptions.TrackedSubscriptionCount; + /// /// Server-pushed data-change notification. Fires from the /// 's background loop; handlers should be cheap (or queue @@ -202,6 +254,10 @@ public sealed class GalaxyDriver /// null-object resolver (returns null for every name) so seam-injecting tests that /// don't exercise a secret: ref need not supply one. /// + /// + /// Optional deploy-event stream seam. When supplied, starts the + /// over it (production builds one over the owned repository client). + /// internal GalaxyDriver( string driverInstanceId, GalaxyDriverOptions options, @@ -212,8 +268,10 @@ public sealed class GalaxyDriver IGalaxyAlarmAcknowledger? alarmAcknowledger = null, IGalaxyAlarmFeed? alarmFeed = null, ILogger? logger = null, - ISecretResolver? secretResolver = null) + ISecretResolver? secretResolver = null, + IGalaxyDeployWatchSource? deployWatchSource = null) { + _deployWatchSource = deployWatchSource; _driverInstanceId = !string.IsNullOrWhiteSpace(driverInstanceId) ? driverInstanceId : throw new ArgumentException("Driver instance id required.", nameof(driverInstanceId)); @@ -359,28 +417,47 @@ public sealed class GalaxyDriver ObjectDisposedException.ThrowIf(_disposed, this); if (HasConfigBody(driverConfigJson)) RefuseNonEquivalentConfig(driverConfigJson); - // Tests inject seams via the internal ctor; production InitializeAsync builds - // the gateway client + session + per-capability runtime components from - // GalaxyDriverOptions. When seams are pre-injected we leave them alone (the - // test exercises the wired surface without a real gw round-trip). - if (_subscriber is null && _dataWriter is null && _hierarchySource is null) + await _lifecycleGate.WaitAsync(cancellationToken).ConfigureAwait(false); + try { - await BuildProductionRuntimeAsync(cancellationToken).ConfigureAwait(false); - } - else - { - _logger.LogDebug( - "GalaxyDriver {InstanceId} initializing with pre-injected seams — production runtime build skipped", - _driverInstanceId); - } + ObjectDisposedException.ThrowIf(_disposed, this); - await StartDeployWatcherAsync(cancellationToken).ConfigureAwait(false); - _logger.LogInformation( - "GalaxyDriver {InstanceId} initialized — endpoint={Endpoint} clientName={ClientName}", - _driverInstanceId, _options.Gateway.Endpoint, _options.MxAccess.ClientName); - _health = new DriverHealth(DriverState.Healthy, DateTime.UtcNow, null); + // Tests inject seams via the internal ctor; production InitializeAsync builds + // the gateway client + session + per-capability runtime components from + // GalaxyDriverOptions. When seams are pre-injected we leave them alone (the + // test exercises the wired surface without a real gw round-trip). After a + // ShutdownAsync the production seams are null again, so the watchdog's + // Shutdown → Initialize genuinely rebuilds the runtime on this same object. + if (_subscriber is null && _dataWriter is null && _hierarchySource is null) + { + await BuildProductionRuntimeAsync(cancellationToken).ConfigureAwait(false); + } + else + { + _logger.LogDebug( + "GalaxyDriver {InstanceId} initializing with pre-injected seams — production runtime build skipped", + _driverInstanceId); + // The per-platform probe watcher needs only a subscriber, so it runs on the seam path + // too — that is what lets the ScanState wiring be exercised without a gateway. + if (_subscriber is not null) _probeWatcher ??= BuildProbeWatcher(_subscriber); + } + + await StartDeployWatcherAsync(cancellationToken).ConfigureAwait(false); + _logger.LogInformation( + "GalaxyDriver {InstanceId} initialized — endpoint={Endpoint} clientName={ClientName}", + _driverInstanceId, _options.Gateway.Endpoint, _options.MxAccess.ClientName); + _health = new DriverHealth(DriverState.Healthy, DateTime.UtcNow, null); + } + finally + { + _lifecycleGate.Release(); + } } + private PerPlatformProbeWatcher BuildProbeWatcher(IGalaxySubscriber subscriber) => + new(subscriber, _hostStatuses, _logger, + bufferedUpdateIntervalMs: _options.MxAccess.PublishingIntervalMs); + /// /// Build the production gw client + session + per-capability runtime components /// from _options. Sets up the reconnect supervisor's reopen / replay @@ -389,9 +466,32 @@ public sealed class GalaxyDriver /// private async Task BuildProductionRuntimeAsync(CancellationToken cancellationToken) { + try + { + await BuildProductionRuntimeCoreAsync(cancellationToken).ConfigureAwait(false); + } + catch (Exception ex) + { + // A failed connect must not leave a half-built runtime behind: the watchdog retries + // InitializeAsync every 10 s, and the seam check above would see the orphaned client + + // session as "already built" — or, worse, rebuild on top of them. Release what was + // created, and land the failed state before AND after the teardown (IDriver remark). + _health = new DriverHealth(DriverState.Faulted, _health.LastSuccessfulRead, ex.Message); + await TeardownRuntimeAsync(final: false, CancellationToken.None).ConfigureAwait(false); + _health = new DriverHealth(DriverState.Faulted, _health.LastSuccessfulRead, ex.Message); + throw; + } + } + + private async Task BuildProductionRuntimeCoreAsync(CancellationToken cancellationToken) + { + _ownsRuntime = true; var clientOptions = await BuildClientOptionsAsync(_options.Gateway, cancellationToken).ConfigureAwait(false); _ownedMxClient = MxGatewayClient.Create(clientOptions); - _ownedMxSession = new GalaxyMxSession(_options.MxAccess, _logger); + _ownedMxSession = new GalaxyMxSession(_options.MxAccess, _logger) + { + OpenAndRegisterOverrideForTests = SessionOpenOverrideForTests, + }; await _ownedMxSession.ConnectAsync(clientOptions, cancellationToken).ConfigureAwait(false); // PR 6.1 — wrap the gw-facing seams in tracing decorators so every Subscribe / @@ -424,9 +524,7 @@ public sealed class GalaxyDriver _supervisor.StateChanged += OnSupervisorStateChanged; - _probeWatcher = new PerPlatformProbeWatcher( - _subscriber, _hostStatuses, _logger, - bufferedUpdateIntervalMs: _options.MxAccess.PublishingIntervalMs); + _probeWatcher = BuildProbeWatcher(_subscriber); // Wire the alarm acknowledger + feed to the live gateway client. Both are // session-less — the gateway serves alarms from an always-on central monitor — @@ -436,6 +534,133 @@ public sealed class GalaxyDriver _ownedMxClient.StreamAlarmsAsync, _logger, _options.MxAccess.ClientName); } + /// + /// Stop every background loop and release every gateway resource this driver holds — the + /// teardown half shared by ( = false: + /// a later rebuilds) and + /// ( = true). Before this existed, ShutdownAsync was a + /// no-op and everything below outlived the driver on every respawn, removal and + /// /raw browse: the gateway accumulated live MX sessions and advised items, and + /// this process accumulated channels and stream loops. + /// + /// + /// Ordering matters: + /// + /// The supervisor goes first and is awaited, so no recovery loop can reopen the + /// session or start a pump while the rest is being released. + /// The probe watcher unsubscribes while the session is still open (its handles + /// die with the session anyway, but a clean release keeps the gateway's advise + /// count honest). + /// Streams (pump, alarm feed, deploy watcher) are cancelled and awaited. + /// The session closes (gateway CloseSession), then both owned channels. + /// + /// Every step is best-effort: one failing release never prevents the next. + /// + private async Task TeardownRuntimeAsync(bool final, CancellationToken cancellationToken) + { + _runtimeStopping = true; + try + { + var supervisor = _supervisor; + _supervisor = null; + if (supervisor is not null) + { + supervisor.StateChanged -= OnSupervisorStateChanged; + try { await supervisor.StopAsync().ConfigureAwait(false); } + catch (Exception ex) { _logger.LogWarning(ex, "ReconnectSupervisor stop failed"); } + } + + var deployWatcher = _deployWatcher; + _deployWatcher = null; + if (deployWatcher is not null) + { + deployWatcher.OnRediscoveryNeeded -= OnDeployWatcherRediscoveryNeeded; + try { await deployWatcher.StopAsync().ConfigureAwait(false); } + catch (Exception ex) { _logger.LogWarning(ex, "DeployWatcher stop failed"); } + } + + var probeWatcher = _probeWatcher; + _probeWatcher = null; + if (probeWatcher is not null) + { + try { await probeWatcher.StopAsync(cancellationToken).ConfigureAwait(false); } + catch (Exception ex) { _logger.LogWarning(ex, "ProbeWatcher stop failed"); } + } + + try { _transportForwarder?.Dispose(); } catch (Exception ex) { _logger.LogWarning(ex, "Transport forwarder dispose failed"); } + _transportForwarder = null; + + EventPump? pump; + lock (_pumpLock) { pump = _eventPump; _eventPump = null; } + if (pump is not null) + { + pump.OnDataChange -= OnPumpDataChange; + try { await pump.DisposeAsync().ConfigureAwait(false); } + catch (Exception ex) { _logger.LogWarning(ex, "EventPump dispose failed"); } + } + // Every item handle the registry knows died with the pump's session; the host + // re-subscribes (fresh ids) once the next InitializeAsync reports Connected. + _subscriptions.Clear(); + + // The alarm feed: owned → disposed and nulled so InitializeAsync builds a fresh one; + // injected → detached (the injector owns its lifetime) unless this is the final dispose. + IGalaxyAlarmFeed? alarmFeed; + bool releaseFeed; + lock (_alarmFeedLock) + { + alarmFeed = _alarmFeed; + releaseFeed = alarmFeed is not null && (_ownsRuntime || final); + if (alarmFeed is not null && _alarmFeedWired) + { + alarmFeed.OnAlarmTransition -= OnAlarmFeedTransition; + _alarmFeedWired = false; + } + if (releaseFeed) _alarmFeed = null; + } + if (releaseFeed) + { + try { await alarmFeed!.DisposeAsync().ConfigureAwait(false); } + catch (Exception ex) { _logger.LogWarning(ex, "Alarm feed dispose failed"); } + } + + if (_ownedMxSession is not null) + { + try { await _ownedMxSession.DisposeAsync().ConfigureAwait(false); } + catch (Exception ex) { _logger.LogWarning(ex, "MxSession dispose failed"); } + _ownedMxSession = null; + } + + if (_ownedMxClient is not null) + { + try { await _ownedMxClient.DisposeAsync().ConfigureAwait(false); } + catch (Exception ex) { _logger.LogWarning(ex, "MxClient dispose failed"); } + _ownedMxClient = null; + } + + if (_ownedRepositoryClient is not null) + { + try { await _ownedRepositoryClient.DisposeAsync().ConfigureAwait(false); } + catch (Exception ex) { _logger.LogWarning(ex, "RepositoryClient dispose failed"); } + _ownedRepositoryClient = null; + } + + if (_ownsRuntime || final) + { + // Null the seams so the next InitializeAsync rebuilds (production) — injected seams + // are only dropped on the final dispose. + _subscriber = null; + _dataWriter = null; + _hierarchySource = null; + _alarmAcknowledger = null; + _ownsRuntime = false; + } + } + finally + { + _runtimeStopping = false; + } + } + /// /// Reopen callback for : recreate the gw session. After a /// gateway restart the existing session handle is faulted / stale but non-null, so a plain @@ -465,6 +690,9 @@ public sealed class GalaxyDriver /// private async Task ReplayAsync(CancellationToken cancellationToken) { + // A teardown cancels the supervisor's cycle between reopen and replay; a replay that ran on + // anyway would start a fresh pump underneath the shutdown. + cancellationToken.ThrowIfCancellationRequested(); if (_subscriber is null) return; var entries = _subscriptions.SnapshotEntries(); if (entries.Count == 0) return; @@ -581,9 +809,9 @@ public sealed class GalaxyDriver } // EnsureEventPumpStarted creates + starts a fresh pump under the lock. Skip the - // recreate if the driver is being disposed — Dispose already tore the pump down - // and a fresh one here would leak past the driver's lifetime. - if (_disposed) return; + // recreate if the driver is being disposed or its runtime torn down — the teardown + // already released the pump and a fresh one here would leak past it. + if (_disposed || _runtimeStopping) return; EnsureEventPumpStarted(); } @@ -631,28 +859,53 @@ public sealed class GalaxyDriver private async Task StartDeployWatcherAsync(CancellationToken cancellationToken) { if (!_options.Repository.WatchDeployEvents) return; - if (_ownedRepositoryClient is null && _hierarchySource is null) return; - // Reuse the lazily-built repository client (DiscoverAsync constructs it on demand). - // If discovery hasn't run yet, build the client here so the watcher has a target. - // Build under a null-check (not ??=) so if BuildDefaultHierarchySourceAsync later - // runs it reuses this client rather than overwriting the field and leaking the - // first instance — the client-options build is now async (secret: arm). - _ownedRepositoryClient ??= ZB.MOM.WW.MxGateway.Client.GalaxyRepositoryClient.Create( - await BuildClientOptionsAsync(_options.Gateway, cancellationToken).ConfigureAwait(false)); + var source = _deployWatchSource; + if (source is null) + { + // Seam mode without an injected watch source has no gateway to watch. (Production + // always owns its runtime here.) The old guard returned whenever BOTH the repository + // client and the hierarchy source were null — which on a production runtime instance + // is every first InitializeAsync, since discovery runs on the /raw picker's throwaway + // instance, not here — so the watcher never started and the re-browse chip never + // fired for a live driver. + if (!_ownsRuntime) return; - var source = new GatewayGalaxyDeployWatchSource(_ownedRepositoryClient); - _deployWatcher = new DeployWatcher(source, _logger); - _deployWatcher.OnRediscoveryNeeded += (_, args) => OnRediscoveryNeeded?.Invoke(this, args); + // Reuse the lazily-built repository client (DiscoverAsync constructs it on demand). + // Build under a null-check (not an overwrite) so if BuildDefaultHierarchySourceAsync + // later runs it reuses this client rather than leaking the first instance. + _ownedRepositoryClient ??= GalaxyRepositoryClient.Create( + await BuildClientOptionsAsync(_options.Gateway, cancellationToken).ConfigureAwait(false)); + source = new GatewayGalaxyDeployWatchSource(_ownedRepositoryClient); + } + + // One watcher per driver: a previous one (an InitializeAsync without a ShutdownAsync in + // between) is stopped before its replacement starts — every live InitializeAsync used to + // stack another watch-stream loop, each raising OnRediscoveryNeeded per real deploy. + var previous = _deployWatcher; + _deployWatcher = null; + if (previous is not null) + { + previous.OnRediscoveryNeeded -= OnDeployWatcherRediscoveryNeeded; + try { await previous.StopAsync().ConfigureAwait(false); } + catch (Exception ex) { _logger.LogWarning(ex, "Previous DeployWatcher stop failed — replacing it anyway."); } + } + + var watcher = new DeployWatcher(source, _logger); + watcher.OnRediscoveryNeeded += OnDeployWatcherRediscoveryNeeded; + _deployWatcher = watcher; // StartAsync schedules the background loop and returns Task.CompletedTask immediately. // It throws InvalidOperationException synchronously if called twice (programming error). // Don't discard the return value — observe any synchronous throw. - var startTask = _deployWatcher.StartAsync(CancellationToken.None); + var startTask = watcher.StartAsync(CancellationToken.None); // The task is already completed (StartAsync is synchronous); surface any synchronous fault. if (startTask.IsFaulted) startTask.GetAwaiter().GetResult(); } + private void OnDeployWatcherRediscoveryNeeded(object? sender, RediscoveryEventArgs args) => + OnRediscoveryNeeded?.Invoke(this, args); + /// public Task ReinitializeAsync(string driverConfigJson, CancellationToken cancellationToken) { @@ -731,13 +984,32 @@ public sealed class GalaxyDriver && a.Reconnect == b.Reconnect && a.ProbeTimeoutSeconds == b.ProbeTimeoutSeconds; + /// + /// Stop the driver and release the gateway runtime: supervisor, deploy watcher, probe + /// watcher, event pump, alarm feed, MX session (gateway CloseSession) and both owned + /// gRPC channels. The seams are nulled afterwards, so a later + /// on this same object — the host's comms-loss recovery and the way ReinitializeAsync + /// is sequenced — rebuilds a fresh runtime instead of reusing a dead one. The driver object + /// itself stays usable; is the final, idempotent teardown. + /// /// - public Task ShutdownAsync(CancellationToken cancellationToken) + public async Task ShutdownAsync(CancellationToken cancellationToken) { - if (_disposed) return Task.CompletedTask; + if (_disposed) return; _logger.LogInformation("GalaxyDriver {InstanceId} shutting down", _driverInstanceId); - _health = new DriverHealth(DriverState.Unknown, _health.LastSuccessfulRead, null); - return Task.CompletedTask; + await _lifecycleGate.WaitAsync(cancellationToken).ConfigureAwait(false); + try + { + if (_disposed) return; + await TeardownRuntimeAsync(final: false, cancellationToken).ConfigureAwait(false); + // Terminal health lands AFTER the teardown (IDriver remark): a loop unwinding during the + // release could otherwise overwrite it with the state it was reporting when cancelled. + _health = new DriverHealth(DriverState.Unknown, _health.LastSuccessfulRead, null); + } + finally + { + _lifecycleGate.Release(); + } } /// @@ -1192,7 +1464,7 @@ public sealed class GalaxyDriver /// private void OnEventPumpReplayGap(ReplayGap gap) { - if (_subscriber is null) return; + if (_subscriber is null || _runtimeStopping) return; if (!_options.Reconnect.ReplayOnSessionLost) { @@ -1249,6 +1521,14 @@ public sealed class GalaxyDriver /// private void OnEventPumpStreamFault(Exception cause) { + if (_runtimeStopping) + { + // The pump is being torn down with the rest of the runtime — a fault it reports on the + // way out is shutdown noise, not something to recover from. + _logger.LogDebug(cause, "GalaxyDriver {InstanceId} event stream ended during runtime teardown.", _driverInstanceId); + return; + } + var supervisor = _supervisor; if (supervisor is null) { @@ -1517,54 +1797,19 @@ public sealed class GalaxyDriver public async ValueTask DisposeAsync() { if (_disposed) return; - _disposed = true; - - // Synchronous sub-components first — none of these block. - try { _deployWatcher?.Dispose(); } catch (Exception ex) { _logger.LogWarning(ex, "DeployWatcher dispose failed"); } - try { _supervisor?.Dispose(); } catch (Exception ex) { _logger.LogWarning(ex, "ReconnectSupervisor dispose failed"); } - try { _probeWatcher?.Dispose(); } catch (Exception ex) { _logger.LogWarning(ex, "ProbeWatcher dispose failed"); } - try { _transportForwarder?.Dispose(); } catch (Exception ex) { _logger.LogWarning(ex, "Transport forwarder dispose failed"); } - - // Async sub-components: await each so we don't block a thread-pool thread - // on a slow shutdown (e.g. EventPump draining its channel, gRPC stream closing). - EventPump? pump; - lock (_pumpLock) { pump = _eventPump; _eventPump = null; } - if (pump is not null) + await _lifecycleGate.WaitAsync().ConfigureAwait(false); + try { - try { await pump.DisposeAsync().ConfigureAwait(false); } - catch (Exception ex) { _logger.LogWarning(ex, "EventPump dispose failed"); } + if (_disposed) return; + _disposed = true; + // Same release sequence as ShutdownAsync, plus the injected seams this instance will + // never need again. Idempotent: a ShutdownAsync before this finds nothing left to do. + await TeardownRuntimeAsync(final: true, CancellationToken.None).ConfigureAwait(false); } - - IGalaxyAlarmFeed? alarmFeed; - lock (_alarmFeedLock) { alarmFeed = _alarmFeed; _alarmFeed = null; } - if (alarmFeed is not null) + finally { - try { await alarmFeed.DisposeAsync().ConfigureAwait(false); } - catch (Exception ex) { _logger.LogWarning(ex, "Alarm feed dispose failed"); } + _lifecycleGate.Release(); } - - if (_ownedMxSession is not null) - { - try { await _ownedMxSession.DisposeAsync().ConfigureAwait(false); } - catch (Exception ex) { _logger.LogWarning(ex, "MxSession dispose failed"); } - _ownedMxSession = null; - } - - if (_ownedMxClient is not null) - { - try { await _ownedMxClient.DisposeAsync().ConfigureAwait(false); } - catch (Exception ex) { _logger.LogWarning(ex, "MxClient dispose failed"); } - _ownedMxClient = null; - } - - if (_ownedRepositoryClient is not null) - { - try { await _ownedRepositoryClient.DisposeAsync().ConfigureAwait(false); } - catch (Exception ex) { _logger.LogWarning(ex, "RepositoryClient dispose failed"); } - _ownedRepositoryClient = null; - } - - _hierarchySource = null; } /// diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Health/PerPlatformProbeWatcher.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Health/PerPlatformProbeWatcher.cs index b766e38b..12afb837 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Health/PerPlatformProbeWatcher.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Health/PerPlatformProbeWatcher.cs @@ -196,20 +196,28 @@ public sealed class PerPlatformProbeWatcher : IDisposable }; } - /// Disposes the probe watcher and unsubscribes all tracked platforms. - public void Dispose() + /// + /// Stops the watcher and unsubscribes every tracked platform's probe handle on the gateway + /// (best-effort — a transport error is logged and swallowed). The async counterpart of + /// ; the driver's ShutdownAsync uses it so the unsubscribe + /// round-trip is awaited rather than blocked on. Idempotent. + /// + /// Bounds the unsubscribe round-trip. + /// A task that completes once the handles have been released (or the attempt gave up). + public async Task StopAsync(CancellationToken cancellationToken) { if (_disposed) return; _disposed = true; - // Best-effort unsubscribe everything we know about. Run synchronously through - // GetAwaiter().GetResult() since Dispose is sync; transport errors are swallowed. var liveHandles = _itemHandlesByPlatform.Values.Where(h => h > 0).ToArray(); _itemHandlesByPlatform.Clear(); if (liveHandles.Length > 0) { - try { _subscriber.UnsubscribeBulkAsync(liveHandles, CancellationToken.None).GetAwaiter().GetResult(); } - catch (Exception ex) { _logger.LogWarning(ex, "PerPlatformProbeWatcher dispose unsubscribe failed"); } + try { await _subscriber.UnsubscribeBulkAsync(liveHandles, cancellationToken).ConfigureAwait(false); } + catch (Exception ex) { _logger.LogWarning(ex, "PerPlatformProbeWatcher stop unsubscribe failed"); } } } + + /// Disposes the probe watcher and unsubscribes all tracked platforms (blocking on the round-trip). + public void Dispose() => StopAsync(CancellationToken.None).GetAwaiter().GetResult(); } diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/EventPump.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/EventPump.cs index d6e4fb4b..e39385b4 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/EventPump.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/EventPump.cs @@ -187,9 +187,11 @@ internal sealed class EventPump : IAsyncDisposable NotifyStreamFault(cause); } } - catch (OperationCanceledException) when (ct.IsCancellationRequested) + catch (Exception) when (ct.IsCancellationRequested) { - // Clean shutdown — no log. + // Clean shutdown — no log. A cancelled gRPC stream may surface as RpcException(Cancelled) + // rather than OperationCanceledException; either way the pump was told to stop, so this + // is not a transport fault and must not wake the reconnect supervisor mid-teardown. } catch (Exception ex) { diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/ReconnectSupervisor.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/ReconnectSupervisor.cs index 15022173..7b6c6410 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/ReconnectSupervisor.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/ReconnectSupervisor.cs @@ -62,6 +62,16 @@ public sealed class ReconnectSupervisor : IDisposable private Task? _recoveryLoop; private CancellationTokenSource? _loopCts; private bool _disposed; + private int _cyclesStarted; + private int _cyclesReleased; + + /// Recovery cycles started so far (one per that left Healthy). Test seam. + internal int CyclesStarted => Volatile.Read(ref _cyclesStarted); + + /// Recovery cycles whose has been disposed. Equals + /// once every loop has exited — a cycle's CTS used to be released only + /// if it happened to be the LAST one when ran. Test seam. + internal int CyclesReleased => Volatile.Read(ref _cyclesReleased); /// Fires after every state transition. public event EventHandler? StateChanged; @@ -136,8 +146,13 @@ public sealed class ReconnectSupervisor : IDisposable TransitionLocked(State.TransportLost, cause.Message); - _loopCts = new CancellationTokenSource(); - _recoveryLoop = Task.Run(() => RecoveryLoopAsync(_loopCts.Token)); + // Each cycle owns its CTS and releases it when its loop exits (see RecoveryLoopAsync's + // finally). Before, every cycle allocated a fresh CTS and only the last one was ever + // disposed — one leaked CancellationTokenSource per outage for the driver's lifetime. + var cts = new CancellationTokenSource(); + _loopCts = cts; + Interlocked.Increment(ref _cyclesStarted); + _recoveryLoop = Task.Run(() => RecoveryLoopAsync(cts)); } } @@ -156,7 +171,26 @@ public sealed class ReconnectSupervisor : IDisposable } } - private async Task RecoveryLoopAsync(CancellationToken ct) + private async Task RecoveryLoopAsync(CancellationTokenSource cycleCts) + { + try + { + await RecoveryLoopCoreAsync(cycleCts.Token).ConfigureAwait(false); + } + finally + { + // Release this cycle's CTS once nothing can observe its token any more. Stop/Dispose may + // race us here: they Cancel (guarded against ODE) and Dispose (idempotent) the same object. + lock (_stateLock) + { + if (ReferenceEquals(_loopCts, cycleCts)) _loopCts = null; + } + cycleCts.Dispose(); + Interlocked.Increment(ref _cyclesReleased); + } + } + + private async Task RecoveryLoopCoreAsync(CancellationToken ct) { var attempt = 0; while (!ct.IsCancellationRequested) @@ -240,8 +274,13 @@ public sealed class ReconnectSupervisor : IDisposable } } - /// Disposes the supervisor and cancels any in-flight recovery loop. - public void Dispose() + /// + /// Cancels any in-flight recovery loop and waits for it to exit, without blocking the caller's + /// thread. The async counterpart of ; the driver's ShutdownAsync + /// uses it so a loop parked inside a reopen round-trip is awaited, not blocked on. Idempotent. + /// + /// A task that completes once the recovery loop has exited. + public async Task StopAsync() { if (_disposed) return; _disposed = true; @@ -250,13 +289,18 @@ public sealed class ReconnectSupervisor : IDisposable Task? loop; lock (_stateLock) { cts = _loopCts; loop = _recoveryLoop; _loopCts = null; _recoveryLoop = null; } - cts?.Cancel(); + // The loop's own finally may have released the CTS already — a cancelled-then-disposed CTS + // throws on Cancel, and a disposed one is a no-op on Dispose. + try { cts?.Cancel(); } catch (ObjectDisposedException) { /* loop already released it */ } if (loop is not null) { - try { loop.GetAwaiter().GetResult(); } catch { /* shutdown */ } + try { await loop.ConfigureAwait(false); } catch { /* shutdown */ } } cts?.Dispose(); } + + /// Disposes the supervisor and cancels any in-flight recovery loop (blocking until it exits). + public void Dispose() => StopAsync().GetAwaiter().GetResult(); } /// diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/SubscriptionRegistry.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/SubscriptionRegistry.cs index 512099f5..9dbd7755 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/SubscriptionRegistry.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/SubscriptionRegistry.cs @@ -100,6 +100,17 @@ internal sealed class SubscriptionRegistry } + /// + /// Drop every subscription and the whole reverse map. Used by the driver's runtime teardown: + /// once the gateway session is closed every item handle in here is dead, and the host + /// re-issues its subscriptions (fresh ids) after the next InitializeAsync. + /// + public void Clear() + { + _bySubscriptionId.Clear(); + _subscribersByItemHandle.Clear(); + } + /// Snapshot every active binding for diagnostic output. /// A flattened list of every tag binding across all tracked subscriptions. public IReadOnlyList SnapshotAllBindings() => diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/GalaxyDriverShutdownTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/GalaxyDriverShutdownTests.cs new file mode 100644 index 00000000..5a146656 --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/GalaxyDriverShutdownTests.cs @@ -0,0 +1,375 @@ +using System.Runtime.CompilerServices; +using System.Threading.Channels; +using Google.Protobuf.WellKnownTypes; +using Shouldly; +using Xunit; +using ZB.MOM.WW.MxGateway.Contracts.Proto; +using ZB.MOM.WW.MxGateway.Contracts.Proto.Galaxy; +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; +using ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Browse; +using ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Config; +using ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Runtime; + +namespace ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests.Runtime; + +/// +/// used to be a no-op (review 2026-10-21, G-1/G-2/G-9): +/// the gRPC runtime outlived the driver on every respawn, removal and /raw browse, and +/// the host's comms-loss recovery (ShutdownAsync → InitializeAsync on the SAME +/// object) neither tore down nor rebuilt anything. These tests pin the teardown at every +/// seam a fake can reach — the StreamEvents stream, the deploy-watch stream, the alarm feed, +/// the owned channels + session — and the rebuild on the next InitializeAsync. +/// +public sealed class GalaxyDriverShutdownTests +{ + private const int WaitMs = 10_000; + + private static GalaxyDriverOptions Opts(bool watchDeploy = true) => new( + new GalaxyGatewayOptions("https://mxgw.test:5001", "key"), + new GalaxyMxAccessOptions("ShutdownTest"), + new GalaxyRepositoryOptions(WatchDeployEvents: watchDeploy), + new GalaxyReconnectOptions()); + + // ───────────────────────────── fakes ───────────────────────────── + + /// + /// Subscriber whose event stream records the token it was opened with and completes a + /// signal when that token fires — the observable for "the StreamEvents stream was cancelled". + /// + private sealed class StreamingSubscriber : IGalaxySubscriber + { + private readonly Channel _events = Channel.CreateUnbounded(); + private int _nextHandle; + + public Dictionary HandleByRef { get; } = new(StringComparer.OrdinalIgnoreCase); + public List Unsubscribed { get; } = []; + public int StreamsOpened; + public TaskCompletionSource StreamCancelled { get; } = new(TaskCreationOptions.RunContinuationsAsynchronously); + + public Task> SubscribeBulkAsync( + IReadOnlyList fullReferences, int bufferedUpdateIntervalMs, CancellationToken cancellationToken) + { + var results = new List(); + foreach (var r in fullReferences) + { + var h = Interlocked.Increment(ref _nextHandle); + HandleByRef[r] = h; + results.Add(new SubscribeResult { TagAddress = r, ItemHandle = h, WasSuccessful = true }); + } + return Task.FromResult>(results); + } + + public Task UnsubscribeBulkAsync(IReadOnlyList itemHandles, CancellationToken cancellationToken) + { + lock (Unsubscribed) Unsubscribed.AddRange(itemHandles); + return Task.CompletedTask; + } + + public async IAsyncEnumerable StreamEventsAsync([EnumeratorCancellation] CancellationToken cancellationToken) + { + Interlocked.Increment(ref StreamsOpened); + using var _ = cancellationToken.Register(() => StreamCancelled.TrySetResult()); + await foreach (var ev in _events.Reader.ReadAllAsync(cancellationToken).ConfigureAwait(false)) + { + yield return ev; + } + } + + public ValueTask EmitAsync(int handle, double value) => _events.Writer.WriteAsync(new MxEvent + { + Family = MxEventFamily.OnDataChange, + ItemHandle = handle, + Value = new MxValue { DoubleValue = value }, + Quality = 192, + SourceTimestamp = Timestamp.FromDateTime(DateTime.UtcNow), + }); + } + + /// + /// Deploy-watch source that counts OPEN enumerations (incremented on open, decremented when + /// the enumerator is cancelled or completes) — the observable for "exactly one watcher". + /// + private sealed class CountingDeployWatchSource : IGalaxyDeployWatchSource + { + private int _open; + private int _opened; + public Channel Current { get; private set; } = Channel.CreateUnbounded(); + + public int Open => Volatile.Read(ref _open); + public int Opened => Volatile.Read(ref _opened); + + public async IAsyncEnumerable WatchAsync( + DateTimeOffset? lastSeenDeployTime, [EnumeratorCancellation] CancellationToken cancellationToken) + { + Interlocked.Increment(ref _open); + Interlocked.Increment(ref _opened); + var channel = Current = Channel.CreateUnbounded(); + try + { + await foreach (var ev in channel.Reader.ReadAllAsync(cancellationToken).ConfigureAwait(false)) + { + yield return ev; + } + } + finally + { + Interlocked.Decrement(ref _open); + } + } + + public ValueTask PushAsync(ulong sequence, DateTimeOffset deployTime) => Current.Writer.WriteAsync(new DeployEvent + { + Sequence = sequence, + TimeOfLastDeployPresent = true, + TimeOfLastDeploy = Timestamp.FromDateTimeOffset(deployTime), + }); + } + + private sealed class RecordingAlarmFeed : IGalaxyAlarmFeed + { + public int Starts; + public int Disposes; + public event EventHandler? OnAlarmTransition; + public void Start() => Starts++; + public void Emit() => OnAlarmTransition?.Invoke(this, new GalaxyAlarmTransition( + "Tank01.Level.HiHi", "Tank01", "AnalogLimitAlarm.HiHi", GalaxyAlarmTransitionKind.Raise, + AlarmSeverity.High, 800, 750, null, DateTime.UtcNow, "", "", "Process", "hi-hi")); + public ValueTask DisposeAsync() { Disposes++; return ValueTask.CompletedTask; } + } + + private static async Task WaitUntilAsync(Func condition, string because) + { + var deadline = DateTime.UtcNow.AddMilliseconds(WaitMs); + while (!condition()) + { + if (DateTime.UtcNow > deadline) throw new ShouldAssertException(because); + await Task.Delay(10); + } + } + + // ───────────────────────────── G-1: streams + pump ───────────────────────────── + + /// + /// After ShutdownAsync the StreamEvents stream is cancelled, the pump is gone and the + /// registry is empty. Positive control: an event delivered through that same stream + /// reached OnDataChange before the shutdown. + /// + [Fact] + public async Task ShutdownAsync_cancels_the_event_stream_and_releases_the_pump() + { + var sub = new StreamingSubscriber(); + await using var driver = new GalaxyDriver("drv", Opts(watchDeploy: false), hierarchySource: null, subscriber: sub); + await driver.InitializeAsync("{}", CancellationToken.None); + + var delivered = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + driver.OnDataChange += (_, e) => delivered.TrySetResult(e); + await driver.SubscribeAsync(["Tank01.Level"], TimeSpan.Zero, CancellationToken.None); + driver.HasEventPumpForTests.ShouldBeTrue(); + + await sub.EmitAsync(sub.HandleByRef["Tank01.Level"], 42.0); + (await Task.WhenAny(delivered.Task, Task.Delay(WaitMs))).ShouldBe(delivered.Task, + "positive control: the stream must deliver before the shutdown"); + + await driver.ShutdownAsync(CancellationToken.None); + + (await Task.WhenAny(sub.StreamCancelled.Task, Task.Delay(WaitMs))).ShouldBe(sub.StreamCancelled.Task, + "ShutdownAsync must cancel the StreamEvents stream"); + driver.HasEventPumpForTests.ShouldBeFalse(); + driver.TrackedSubscriptionCountForTests.ShouldBe(0, "handles died with the session; the host re-subscribes after re-init"); + driver.GetHealth().State.ShouldBe(DriverState.Unknown); + driver.GetHealth().LastSuccessfulRead.ShouldNotBeNull("the liveness evidence survives a shutdown"); + } + + /// + /// An injected alarm feed is detached on shutdown (the injector owns it, so it is NOT + /// disposed) and re-wired by the next SubscribeAlarmsAsync after a re-init. Positive + /// control on both sides of the shutdown. + /// + [Fact] + public async Task ShutdownAsync_detaches_the_alarm_feed_and_reinit_rewires_it() + { + var feed = new RecordingAlarmFeed(); + var sub = new StreamingSubscriber(); + await using var driver = new GalaxyDriver("drv", Opts(watchDeploy: false), hierarchySource: null, + subscriber: sub, alarmFeed: feed); + var received = 0; + driver.OnAlarmEvent += (_, _) => Interlocked.Increment(ref received); + + await driver.InitializeAsync("{}", CancellationToken.None); + await driver.SubscribeAlarmsAsync(["Tank01.Level.HiHi"], CancellationToken.None); + feed.Emit(); + received.ShouldBe(1, "positive control: the feed is wired before the shutdown"); + + await driver.ShutdownAsync(CancellationToken.None); + feed.Emit(); + received.ShouldBe(1, "a detached feed must not reach OnAlarmEvent"); + feed.Disposes.ShouldBe(0, "an injected feed belongs to the injector"); + + await driver.InitializeAsync("{}", CancellationToken.None); + await driver.SubscribeAlarmsAsync(["Tank01.Level.HiHi"], CancellationToken.None); + feed.Emit(); + received.ShouldBe(2, "re-init + re-subscribe must re-wire the feed"); + feed.Starts.ShouldBe(2); + } + + // ───────────────────────────── G-2: one DeployWatcher ───────────────────────────── + + /// + /// Five watchdog cycles (ShutdownAsync → InitializeAsync) leave exactly one open watch + /// stream. Positive control: a real deploy change then raises OnRediscoveryNeeded exactly + /// once — five stacked watchers would raise five times. + /// + [Fact] + public async Task Five_watchdog_cycles_leave_exactly_one_DeployWatcher() + { + var source = new CountingDeployWatchSource(); + var sub = new StreamingSubscriber(); + await using var driver = new GalaxyDriver("drv", Opts(), hierarchySource: null, subscriber: sub, + deployWatchSource: source); + var raised = 0; + driver.OnRediscoveryNeeded += (_, _) => Interlocked.Increment(ref raised); + + await driver.InitializeAsync("{}", CancellationToken.None); + for (var cycle = 0; cycle < 5; cycle++) + { + // Let this cycle's watch stream actually open before the next shutdown, so every cycle + // holds a live stream the teardown has to cancel (a loop cancelled before its first + // open would make the count vacuous). + await WaitUntilAsync(() => source.Opened == cycle + 1, $"cycle {cycle}: the watch stream must open"); + await driver.ShutdownAsync(CancellationToken.None); + await driver.InitializeAsync("{}", CancellationToken.None); + } + + await WaitUntilAsync(() => source.Opened == 6 && source.Open == 1, + $"expected six opens and exactly one open watch stream, saw opened={source.Opened} open={source.Open}"); + driver.DeployWatcherForTests.ShouldNotBeNull(); + + // Bootstrap (suppressed) then a genuine change. + await source.PushAsync(1, new DateTimeOffset(2026, 1, 1, 0, 0, 0, TimeSpan.Zero)); + await source.PushAsync(2, new DateTimeOffset(2026, 1, 2, 0, 0, 0, TimeSpan.Zero)); + await WaitUntilAsync(() => Volatile.Read(ref raised) >= 1, "positive control: a deploy change must raise rediscovery"); + await Task.Delay(200); + Volatile.Read(ref raised).ShouldBe(1, "stacked watchers would each raise for the same deploy"); + + await driver.ShutdownAsync(CancellationToken.None); + await WaitUntilAsync(() => source.Open == 0, "ShutdownAsync must cancel the deploy-watch stream"); + driver.DeployWatcherForTests.ShouldBeNull(); + } + + /// A second InitializeAsync without a shutdown in between replaces the watcher rather than stacking one. + [Fact] + public async Task InitializeAsync_twice_replaces_the_previous_DeployWatcher() + { + var source = new CountingDeployWatchSource(); + var sub = new StreamingSubscriber(); + await using var driver = new GalaxyDriver("drv", Opts(), hierarchySource: null, subscriber: sub, + deployWatchSource: source); + + await driver.InitializeAsync("{}", CancellationToken.None); + var first = driver.DeployWatcherForTests; + await WaitUntilAsync(() => source.Opened == 1, "the first watch stream must open"); + await driver.InitializeAsync("{}", CancellationToken.None); + + await WaitUntilAsync(() => source.Opened == 2 && source.Open == 1, + $"expected two opens and one open watch stream, saw opened={source.Opened} open={source.Open}"); + driver.DeployWatcherForTests.ShouldNotBeSameAs(first); + } + + // ───────────────────────────── G-1/G-9: owned runtime, channels + session ───────────────────────────── + + /// + /// The production path (no injected seams): InitializeAsync builds the owned client, session + /// and supervisor; ShutdownAsync closes the session, disposes both channels and nulls the + /// seams; the next InitializeAsync opens a NEW session on NEW channels — the watchdog's + /// recovery now genuinely rebuilds. The session open is replaced by a seam; the gRPC + /// channels are real but lazy (nothing dials until a call is made). + /// + [Fact] + public async Task ShutdownAsync_closes_the_session_disposes_both_channels_and_reinit_rebuilds() + { + var opens = 0; + var source = new CountingDeployWatchSource(); + await using var driver = new GalaxyDriver("drv", Opts(), hierarchySource: null, deployWatchSource: source) + { + SessionOpenOverrideForTests = _ => { opens++; return Task.CompletedTask; }, + }; + + await driver.InitializeAsync("{}", CancellationToken.None); + opens.ShouldBe(1); + var client = driver.OwnedMxClientForTests.ShouldNotBeNull(); + var session = driver.OwnedMxSessionForTests.ShouldNotBeNull(); + session.IsConnected.ShouldBeTrue(); + driver.SupervisorForTests.ShouldNotBeNull(); + driver.ProbeWatcherForTests.ShouldNotBeNull(); + driver.DeployWatcherForTests.ShouldNotBeNull(); + driver.GetHealth().State.ShouldBe(DriverState.Healthy); + var stamp = driver.GetHealth().LastSuccessfulRead.ShouldNotBeNull(); + + await driver.ShutdownAsync(CancellationToken.None); + + session.IsConnected.ShouldBeFalse("ShutdownAsync must close the MX session"); + await Should.ThrowAsync(() => client.OpenSessionAsync(cancellationToken: CancellationToken.None), + "the owned gateway channel must be disposed"); + driver.OwnedMxClientForTests.ShouldBeNull(); + driver.OwnedMxSessionForTests.ShouldBeNull(); + driver.SupervisorForTests.ShouldBeNull(); + driver.ProbeWatcherForTests.ShouldBeNull(); + driver.DeployWatcherForTests.ShouldBeNull(); + await WaitUntilAsync(() => source.Open == 0, "the deploy-watch stream must be cancelled"); + driver.GetHealth().State.ShouldBe(DriverState.Unknown); + driver.GetHealth().LastSuccessfulRead.ShouldBe(stamp); + + // The watchdog's second half: a fresh runtime on the same object. + await driver.InitializeAsync("{}", CancellationToken.None); + opens.ShouldBe(2, "re-init must open a NEW session, not reuse the dead one"); + driver.OwnedMxClientForTests.ShouldNotBeNull().ShouldNotBeSameAs(client); + driver.OwnedMxSessionForTests.ShouldNotBeNull().ShouldNotBeSameAs(session); + driver.OwnedMxSessionForTests!.IsConnected.ShouldBeTrue(); + driver.SupervisorForTests.ShouldNotBeNull(); + driver.GetHealth().State.ShouldBe(DriverState.Healthy); + await WaitUntilAsync(() => source.Open == 1, "re-init must start one deploy watcher"); + } + + /// + /// A connect that fails mid-build releases the partial runtime (no orphaned client per + /// watchdog retry), lands Faulted, and the next InitializeAsync starts from nothing. + /// + [Fact] + public async Task Failed_InitializeAsync_releases_the_partial_runtime_and_the_retry_rebuilds() + { + var attempts = 0; + await using var driver = new GalaxyDriver("drv", Opts(watchDeploy: false), hierarchySource: null) + { + SessionOpenOverrideForTests = _ => ++attempts == 1 + ? throw new InvalidOperationException("gateway refused") + : Task.CompletedTask, + }; + + var ex = await Should.ThrowAsync(() => driver.InitializeAsync("{}", CancellationToken.None)); + ex.Message.ShouldBe("gateway refused"); + driver.OwnedMxClientForTests.ShouldBeNull("a failed connect must not orphan the client"); + driver.OwnedMxSessionForTests.ShouldBeNull(); + driver.GetHealth().State.ShouldBe(DriverState.Faulted); + driver.GetHealth().LastError.ShouldBe("gateway refused"); + + await driver.InitializeAsync("{}", CancellationToken.None); + attempts.ShouldBe(2); + driver.OwnedMxSessionForTests.ShouldNotBeNull().IsConnected.ShouldBeTrue(); + driver.GetHealth().State.ShouldBe(DriverState.Healthy); + } + + /// DisposeAsync after ShutdownAsync is harmless, and ShutdownAsync after DisposeAsync is a no-op. + [Fact] + public async Task Shutdown_then_dispose_and_dispose_then_shutdown_are_both_harmless() + { + var sub = new StreamingSubscriber(); + var driver = new GalaxyDriver("drv", Opts(watchDeploy: false), hierarchySource: null, subscriber: sub); + await driver.InitializeAsync("{}", CancellationToken.None); + await driver.SubscribeAsync(["Tank01.Level"], TimeSpan.Zero, CancellationToken.None); + + await driver.ShutdownAsync(CancellationToken.None); + await Should.NotThrowAsync(async () => await driver.DisposeAsync()); + await Should.NotThrowAsync(() => driver.ShutdownAsync(CancellationToken.None)); + await Should.ThrowAsync(() => driver.InitializeAsync("{}", CancellationToken.None)); + } +} diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/ReconnectSupervisorCycleCtsTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/ReconnectSupervisorCycleCtsTests.cs new file mode 100644 index 00000000..1660751d --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/ReconnectSupervisorCycleCtsTests.cs @@ -0,0 +1,92 @@ +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Runtime; + +namespace ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests.Runtime; + +/// +/// Review 2026-10-21 G-3: allocated a +/// new per recovery cycle and only the last one was ever +/// disposed. Each cycle now releases its own CTS when its loop exits, and +/// awaits the loop instead of blocking on it. +/// +public sealed class ReconnectSupervisorCycleCtsTests +{ + private const int WaitMs = 10_000; + + private static ReconnectOptions FastBackoff() => + new(InitialBackoffOverride: TimeSpan.FromMilliseconds(5), MaxBackoffOverride: TimeSpan.FromMilliseconds(20)); + + /// Three completed cycles ⇒ three released CTSs; the in-flight one is not released early (positive control). + [Fact] + public async Task Every_completed_recovery_cycle_releases_its_own_cts() + { + var holdReopen = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var reopenEntered = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var hold = false; + using var sup = new ReconnectSupervisor( + reopen: async _ => + { + if (hold) + { + reopenEntered.TrySetResult(); + await holdReopen.Task; + } + }, + replay: _ => Task.CompletedTask, + options: FastBackoff()); + + for (var i = 0; i < 3; i++) + { + sup.ReportTransportFailure(new IOException($"drop {i}")); + await sup.WaitForHealthyAsync(new CancellationTokenSource(WaitMs).Token); + } + await WaitUntilAsync(() => sup.CyclesReleased == 3, "each completed cycle must release its CTS"); + sup.CyclesStarted.ShouldBe(3); + + // Positive control: a cycle parked inside reopen still holds its CTS. + hold = true; + sup.ReportTransportFailure(new IOException("drop 3")); + (await Task.WhenAny(reopenEntered.Task, Task.Delay(WaitMs))).ShouldBe(reopenEntered.Task); + sup.CyclesStarted.ShouldBe(4); + sup.CyclesReleased.ShouldBe(3, "an in-flight cycle's CTS is live until its loop exits"); + + holdReopen.TrySetResult(); + await sup.WaitForHealthyAsync(new CancellationTokenSource(WaitMs).Token); + await WaitUntilAsync(() => sup.CyclesReleased == 4, "the fourth cycle must release its CTS on exit"); + } + + /// StopAsync cancels a parked cycle, awaits its exit and releases its CTS without blocking. + [Fact] + public async Task StopAsync_cancels_the_in_flight_cycle_and_releases_its_cts() + { + var reopenEntered = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var sup = new ReconnectSupervisor( + reopen: async ct => + { + reopenEntered.TrySetResult(); + await Task.Delay(Timeout.Infinite, ct); + }, + replay: _ => Task.CompletedTask, + options: FastBackoff()); + + sup.ReportTransportFailure(new IOException("drop")); + (await Task.WhenAny(reopenEntered.Task, Task.Delay(WaitMs))).ShouldBe(reopenEntered.Task); + + var stop = sup.StopAsync(); + (await Task.WhenAny(stop, Task.Delay(WaitMs))).ShouldBe(stop, "StopAsync must cancel the parked reopen and return"); + sup.CyclesReleased.ShouldBe(1); + await Should.NotThrowAsync(sup.StopAsync); // idempotent + Should.NotThrow(sup.Dispose); + } + + private static async Task WaitUntilAsync(Func condition, string because) + { + var deadline = DateTime.UtcNow.AddMilliseconds(WaitMs); + while (!condition()) + { + if (DateTime.UtcNow > deadline) throw new ShouldAssertException(because); + await Task.Delay(10); + } + } +}