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 8fc00fa8..d0fac54a 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/GalaxyDriver.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/GalaxyDriver.cs @@ -157,10 +157,22 @@ public sealed class GalaxyDriver // from DiscoverAsync alone left the production watcher with no platforms, ever. private readonly IGalaxyPlatformSource? _platformSource; private IGalaxyPlatformSource? _ownedPlatformSource; - // Rediscovery re-seed requests not yet served. The first request starts the worker; later ones - // only bump the count, and the worker loops until it has served every request it saw — so a - // deploy that lands while a pass is mid-fetch still gets a pass that starts after it. + // Probe-seed requests (init + every rediscovery) not yet served. The first request starts the + // background worker; later ones only bump the count, and the worker loops until it has served + // every request it saw — so a request that lands while a pass is mid-fetch (a deploy, or the + // next runtime's init after a teardown) still gets a pass that starts after it. private int _reseedRequests; + private volatile string _seedReason = "initialize"; + + // Cancels the in-flight background seed of the CURRENT runtime. Created by InitializeAsync, + // cancelled (never disposed — it has no timer or wait handle, and the worker may still read its + // token) by the teardown. + private CancellationTokenSource? _seedCts; + + // Serialises "is this registration still for the live runtime?" against the teardown's registry + // Clear, so a subscribe / re-subscribe whose SubscribeBulk straddled a teardown either lands + // before the Clear (and is wiped by it) or sees the bumped generation and backs out. + private readonly Lock _registryGate = new(); // Bumped by every runtime teardown. Background work started under one runtime (the replay-gap // re-snapshot, a rediscovery re-seed) re-checks it before touching the registry / watcher, so a @@ -475,7 +487,11 @@ public sealed class GalaxyDriver } await StartDeployWatcherAsync(cancellationToken).ConfigureAwait(false); - await SeedProbePlatformsAsync("initialize", cancellationToken).ConfigureAwait(false); + // The platform seed is a gateway DiscoverHierarchy round-trip, which can spend the whole of + // a cold gateway cache's first-load wait — longer than Galaxy's Tier-A Initialize budget. It + // runs in the background: InitializeAsync neither waits on it nor fails because of it. + await PrepareProbeSeedAsync(cancellationToken).ConfigureAwait(false); + RequestProbeSeed("initialize"); _logger.LogInformation( "GalaxyDriver {InstanceId} initialized — endpoint={Endpoint} clientName={ClientName}", _driverInstanceId, _options.Gateway.Endpoint, _options.MxAccess.ClientName); @@ -488,39 +504,95 @@ public sealed class GalaxyDriver } /// - /// Fetch the galaxy's platform / engine set and hand it to the probe watcher. Best-effort: - /// a failed fetch logs a Warning and leaves the watcher as it was — it must never fail an - /// InitializeAsync (the value path does not depend on it), and the next rediscovery - /// re-seeds. No-op when there is no watcher or no platform source (seam mode without one). + /// Under the lifecycle gate, before the first seed request of a runtime: resolve the platform + /// source (production builds it over the owned repository client — no network here) and arm + /// the runtime's seed cancellation. Resolving here, not in the worker, keeps the background + /// pass from creating a repository client after a teardown released it. A failure is a + /// Warning: the value path does not depend on the probe set. /// - private async Task SeedProbePlatformsAsync(string reason, CancellationToken cancellationToken) + private async Task PrepareProbeSeedAsync(CancellationToken cancellationToken) { - var watcher = _probeWatcher; - if (watcher is null) return; - - IGalaxyPlatformSource? source; + if (_probeWatcher is null) return; + _seedCts ??= new CancellationTokenSource(); + if (_platformSource is not null || !_ownsRuntime || _ownedPlatformSource is not null) return; try { - source = await ResolvePlatformSourceAsync(cancellationToken).ConfigureAwait(false); + _ownedRepositoryClient ??= GalaxyRepositoryClient.Create( + await BuildClientOptionsAsync(_options.Gateway, cancellationToken).ConfigureAwait(false)); + _ownedPlatformSource = + new GatewayGalaxyPlatformSource(_ownedRepositoryClient, _options.Repository.DiscoverPageSize); } - catch (Exception ex) + catch (Exception ex) when (ex is not OperationCanceledException || !cancellationToken.IsCancellationRequested) { _logger.LogWarning(ex, - "GalaxyDriver {InstanceId} could not build the platform query ({Reason}) — per-platform host status stays unseeded until the next rediscovery.", - _driverInstanceId, reason); - return; + "GalaxyDriver {InstanceId} could not build the platform query — per-platform host status stays unseeded until the next rediscovery.", + _driverInstanceId); } - if (source is null) return; + } + + /// + /// Ask for a background probe seed (init and every deploy-watcher rediscovery). Requests + /// coalesce onto one worker; each worker pass serves the runtime that is live when the pass + /// STARTS, and a pass whose fetch completes after a teardown is discarded — so a request made + /// under a new runtime always gets a pass of its own. + /// + private void RequestProbeSeed(string reason) + { + if (_runtimeStopping || _probeWatcher is null) return; + _seedReason = reason; + if (Interlocked.Increment(ref _reseedRequests) != 1) return; // the running worker will serve it + + _ = Task.Run(async () => + { + int served; + do + { + served = Volatile.Read(ref _reseedRequests); + var generation = Volatile.Read(ref _runtimeGeneration); + var seedCts = Volatile.Read(ref _seedCts); + if (!_runtimeStopping && seedCts is not null) + { + await SeedProbePlatformsAsync(_seedReason, generation, seedCts.Token).ConfigureAwait(false); + } + } + while (Interlocked.Add(ref _reseedRequests, -served) != 0); + }); + } + + /// + /// One seed pass: fetch the platform / engine set and hand it to the probe watcher of the + /// runtime identified by . Never throws — a failed or cancelled + /// fetch logs and leaves the watcher as it was (the next rediscovery retries), and a result + /// that lands after that runtime was torn down is discarded. + /// + private async Task SeedProbePlatformsAsync(string reason, int generation, CancellationToken cancellationToken) + { + var watcher = _probeWatcher; + var source = _platformSource ?? _ownedPlatformSource; + if (watcher is null || source is null) return; try { var platforms = await source.GetPlatformTagNamesAsync(cancellationToken).ConfigureAwait(false); + if (Volatile.Read(ref _runtimeGeneration) != generation || _runtimeStopping + || !ReferenceEquals(watcher, _probeWatcher)) + { + _logger.LogDebug( + "GalaxyDriver {InstanceId} discarded a platform seed ({Reason}) — its runtime was torn down.", + _driverInstanceId, reason); + return; + } await watcher.SyncPlatformsAsync(platforms, cancellationToken).ConfigureAwait(false); _logger.LogInformation( "GalaxyDriver {InstanceId} probe watcher seeded with {Count} platform(s) ({Reason}).", _driverInstanceId, platforms.Count, reason); } - catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) { throw; } + catch (Exception ex) when (cancellationToken.IsCancellationRequested + || Volatile.Read(ref _runtimeGeneration) != generation) + { + _logger.LogDebug(ex, "GalaxyDriver {InstanceId} platform seed ({Reason}) abandoned by a teardown.", + _driverInstanceId, reason); + } catch (Exception ex) { _logger.LogWarning(ex, @@ -529,50 +601,48 @@ public sealed class GalaxyDriver } } - private async Task ResolvePlatformSourceAsync(CancellationToken cancellationToken) - { - if (_platformSource is not null) return _platformSource; - if (!_ownsRuntime) return null; // seam mode without an injected source: nothing to query - if (_ownedPlatformSource is not null) return _ownedPlatformSource; + /// Deploy-watcher rediscovery: a redeploy may add or remove platforms / engines. + private void ReseedProbePlatformsOnRediscovery() => RequestProbeSeed("rediscovery"); - _ownedRepositoryClient ??= GalaxyRepositoryClient.Create( - await BuildClientOptionsAsync(_options.Gateway, cancellationToken).ConfigureAwait(false)); - return _ownedPlatformSource = - new GatewayGalaxyPlatformSource(_ownedRepositoryClient, _options.Repository.DiscoverPageSize); + /// + /// Register only while is still the + /// live runtime. Atomic with the teardown's registry Clear (both under ). + /// + private bool TryRegisterForGeneration(long subscriptionId, IReadOnlyList bindings, int generation) + { + lock (_registryGate) + { + if (Volatile.Read(ref _runtimeGeneration) != generation) return false; + _subscriptions.Register(subscriptionId, bindings); + return true; + } + } + + private void ClearRegistry() + { + lock (_registryGate) _subscriptions.Clear(); } /// - /// Deploy-watcher rediscovery: re-seed the platform set off the watcher's thread, one pass at - /// a time, and only if the runtime it was raised under is still the live one. + /// Best-effort release of the live handles a SubscribeBulk produced for a call that then lost + /// its runtime — the subscriber may already be dead, so a failure is logged and swallowed. /// - private void ReseedProbePlatformsOnRediscovery() + private async Task ReleaseOrphanedHandlesAsync(IGalaxySubscriber subscriber, IReadOnlyList bindings) { - if (_runtimeStopping || _probeWatcher is null) return; - if (Interlocked.Increment(ref _reseedRequests) != 1) return; // the running worker will serve it - var generation = Volatile.Read(ref _runtimeGeneration); - - _ = Task.Run(async () => + var live = bindings.Where(b => b.ItemHandle > 0).Select(b => b.ItemHandle).ToArray(); + if (live.Length == 0) return; + try { await subscriber.UnsubscribeBulkAsync(live, CancellationToken.None).ConfigureAwait(false); } + catch (Exception ex) { - int served; - do - { - served = Volatile.Read(ref _reseedRequests); - try - { - if (Volatile.Read(ref _runtimeGeneration) == generation && !_runtimeStopping) - { - await SeedProbePlatformsAsync("rediscovery", CancellationToken.None).ConfigureAwait(false); - } - } - catch (Exception ex) - { - _logger.LogWarning(ex, "GalaxyDriver {InstanceId} rediscovery re-seed failed.", _driverInstanceId); - } - } - while (Interlocked.Add(ref _reseedRequests, -served) != 0); - }); + _logger.LogWarning(ex, + "GalaxyDriver {InstanceId} could not release {Count} handle(s) orphaned by a teardown.", + _driverInstanceId, live.Length); + } } + private static InvalidOperationException TornDownDuringSubscribe() => new( + "GalaxyDriver runtime was torn down while the subscribe was in flight — retry after the next InitializeAsync."); + // The watcher publishes its probe handles into the driver's registry (reserved id) so the // EventPump's fan-out reaches OnPumpDataChange for ScanState events — without that every // probe event was dropped before it got anywhere near the watcher. @@ -616,8 +686,9 @@ public sealed class GalaxyDriver stale.OnDataChange -= OnPumpDataChange; try { await stale.DisposeAsync().ConfigureAwait(false); } catch (Exception ex) { _logger.LogWarning(ex, "Stale EventPump dispose failed before runtime rebuild"); } - _subscriptions.Clear(); } + // Unconditionally: whatever the registry holds names handles of a session this rebuild replaces. + ClearRegistry(); _ownsRuntime = true; var clientOptions = await BuildClientOptionsAsync(_options.Gateway, cancellationToken).ConfigureAwait(false); @@ -694,6 +765,7 @@ public sealed class GalaxyDriver { _runtimeStopping = true; Interlocked.Increment(ref _runtimeGeneration); + Interlocked.Exchange(ref _seedCts, null)?.Cancel(); // a seed in flight belongs to this runtime try { var supervisor = _supervisor; @@ -735,7 +807,7 @@ public sealed class GalaxyDriver } // 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(); + ClearRegistry(); // 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. @@ -928,8 +1000,11 @@ public sealed class GalaxyDriver // Rebind so the EventPump fan-out reverse map points at the post-reconnect // handles; otherwise events on the new handles miss every subscription. Not after a // teardown: the fresh handles belong to a subscriber that is already being released. - if (Volatile.Read(ref _runtimeGeneration) != generation) return tagCount; - _subscriptions.Rebind(subscriptionId, newBindings); + lock (_registryGate) + { + if (Volatile.Read(ref _runtimeGeneration) != generation) return tagCount; + _subscriptions.Rebind(subscriptionId, newBindings); + } tagCount += refs.Length; } @@ -1308,6 +1383,8 @@ public sealed class GalaxyDriver deadlineCts.CancelAfter(ReadViaSubscribeDeadline); var cancellationToken = deadlineCts.Token; + var generation = Volatile.Read(ref _runtimeGeneration); // see SubscribeAsync + var readSubscriber = _subscriber!; var pump = EnsureEventPumpStarted(); var subscriptionId = _subscriptions.NextSubscriptionId(); @@ -1341,7 +1418,7 @@ public sealed class GalaxyDriver // The RPC gets only the CALLER's token: the internal deadline bounds our wait, not the RPC, so a slow // gateway can never surface an OperationCanceledException to a caller that did not cancel, and the // handles it eventually creates are known (and unsubscribed below) rather than lost to a cancelled call. - var subscribeTask = _subscriber!.SubscribeBulkAsync(fullReferences, bufferedIntervalMs, callerToken); + var subscribeTask = readSubscriber.SubscribeBulkAsync(fullReferences, bufferedIntervalMs, callerToken); try { results = await subscribeTask.WaitAsync(cancellationToken).ConfigureAwait(false); @@ -1350,7 +1427,7 @@ public sealed class GalaxyDriver { // Internal deadline fired while the subscribe RPC was still in flight. pump.OnDataChange -= handler; - var subscriber = _subscriber!; + var subscriber = readSubscriber; var logger = _logger; _ = subscribeTask.ContinueWith(t => { @@ -1402,7 +1479,12 @@ public sealed class GalaxyDriver ServerTimestampUtc: DateTime.UtcNow)); } } - _subscriptions.Register(subscriptionId, bindings); + if (!TryRegisterForGeneration(subscriptionId, bindings, generation)) + { + pump.OnDataChange -= handler; + await ReleaseOrphanedHandlesAsync(readSubscriber, bindings).ConfigureAwait(false); + throw TornDownDuringSubscribe(); + } try { @@ -1441,7 +1523,7 @@ public sealed class GalaxyDriver { try { - await _subscriber!.UnsubscribeBulkAsync(liveHandles, CancellationToken.None) + await readSubscriber.UnsubscribeBulkAsync(liveHandles, CancellationToken.None) .ConfigureAwait(false); } catch (Exception ex) @@ -1502,6 +1584,10 @@ public sealed class GalaxyDriver "Call InitializeAsync and wait for a successful connection before subscribing."); } + // Captured before the pump/stopping check: a teardown that starts after this point bumps the + // generation, so the Register below can tell its SubscribeBulk straddled a teardown. + var generation = Volatile.Read(ref _runtimeGeneration); + var subscriber = _subscriber; var pump = EnsureEventPumpStarted(); var subscriptionId = _subscriptions.NextSubscriptionId(); @@ -1509,7 +1595,7 @@ public sealed class GalaxyDriver { // Empty subscriptions register but never bind anything — keeps Unsubscribe // symmetric for callers that conditionally add tags later. - _subscriptions.Register(subscriptionId, []); + if (!TryRegisterForGeneration(subscriptionId, [], generation)) throw TornDownDuringSubscribe(); return new GalaxySubscriptionHandle(subscriptionId); } @@ -1525,7 +1611,7 @@ public sealed class GalaxyDriver // and infrastructure callers (probe watcher, deploy watcher) hit the fallback. var requested = (int)Math.Max(0, publishingInterval.TotalMilliseconds); var bufferedIntervalMs = requested > 0 ? requested : _options.MxAccess.PublishingIntervalMs; - var results = await _subscriber + var results = await subscriber .SubscribeBulkAsync(dialledRefs, bufferedIntervalMs, cancellationToken) .ConfigureAwait(false); @@ -1553,7 +1639,13 @@ public sealed class GalaxyDriver } } - _subscriptions.Register(subscriptionId, bindings); + if (!TryRegisterForGeneration(subscriptionId, bindings, generation)) + { + // The teardown cleared the registry while SubscribeBulk was in flight; registering now + // would carry dead handles into the next runtime. Release what we got and fail the call. + await ReleaseOrphanedHandlesAsync(subscriber, bindings).ConfigureAwait(false); + throw TornDownDuringSubscribe(); + } _ = pump; // keep the pump alive for the subscription's lifetime return new GalaxySubscriptionHandle(subscriptionId); } 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 ae4c41ac..64080d52 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 @@ -49,7 +49,8 @@ public sealed class PerPlatformProbeWatcher : IDisposable private readonly ConcurrentDictionary _itemHandlesByPlatform = new(StringComparer.OrdinalIgnoreCase); private readonly Lock _syncLock = new(); - private bool _disposed; + // Written under _syncLock by StopAsync; read under it wherever a write-back must not outlive a stop. + private volatile bool _disposed; /// Initializes a new instance of the PerPlatformProbeWatcher class. /// The Galaxy subscriber for managing probe subscriptions. @@ -100,10 +101,18 @@ public sealed class PerPlatformProbeWatcher : IDisposable private IReadOnlyList BuildBindings() => [.. _itemHandlesByPlatform.Select(kvp => new TagBinding(kvp.Key + ProbeSuffix, kvp.Value))]; - /// Mirror the tracked handle set into the driver registry (no-op without one). + /// + /// Mirror the tracked handle set into the driver registry (no-op without one, or once stopped — + /// a stopped watcher must never write its entry back into a registry the teardown cleared). + /// private void PublishBindings() { - if (_registry is null) return; + lock (_syncLock) PublishBindingsLocked(); + } + + private void PublishBindingsLocked() + { + if (_registry is null || _disposed) return; var bindings = BuildBindings(); if (bindings.Count == 0) _registry.Remove(SubscriptionRegistry.ProbeSubscriptionId); else _registry.Upsert(SubscriptionRegistry.ProbeSubscriptionId, bindings); @@ -136,24 +145,51 @@ public sealed class PerPlatformProbeWatcher : IDisposable var results = await _subscriber.SubscribeBulkAsync( probeAddresses, _bufferedUpdateIntervalMs, cancellationToken).ConfigureAwait(false); + var acquired = new (string Platform, int ItemHandle, string? Error)[platforms.Count]; for (var i = 0; i < platforms.Count; i++) { - var platform = platforms[i]; var match = results.FirstOrDefault(r => string.Equals( r.TagAddress, probeAddresses[i], StringComparison.OrdinalIgnoreCase)); + acquired[i] = (platforms[i], match is { WasSuccessful: true } ? match.ItemHandle : 0, + match?.ErrorMessage ?? ""); + } - var itemHandle = match is { WasSuccessful: true } ? match.ItemHandle : 0; - _itemHandlesByPlatform[platform] = itemHandle; - - if (itemHandle <= 0) + // Re-check the stop AFTER the await, atomically with the record + publish: a StopAsync that + // ran while SubscribeBulk was in flight already cleared the map and the registry entry, so + // recording now would resurrect the probe entry into the next runtime's registry. + bool stopped; + lock (_syncLock) + { + stopped = _disposed; + if (!stopped) { - _logger.LogWarning( - "PerPlatformProbeWatcher subscribe failed for {Platform}: {Error}", - platform, match?.ErrorMessage ?? ""); + foreach (var (platform, itemHandle, _) in acquired) _itemHandlesByPlatform[platform] = itemHandle; + PublishBindingsLocked(); } } - PublishBindings(); + if (stopped) + { + var orphans = acquired.Where(a => a.ItemHandle > 0).Select(a => a.ItemHandle).ToArray(); + if (orphans.Length > 0) + { + try { await _subscriber.UnsubscribeBulkAsync(orphans, CancellationToken.None).ConfigureAwait(false); } + catch (Exception ex) + { + _logger.LogWarning(ex, + "PerPlatformProbeWatcher could not release {Count} probe handle(s) acquired after stop", orphans.Length); + } + } + return; + } + + foreach (var (platform, itemHandle, error) in acquired) + { + if (itemHandle <= 0) + { + _logger.LogWarning("PerPlatformProbeWatcher subscribe failed for {Platform}: {Error}", platform, error); + } + } } /// Snapshot of platform tag names currently watched. @@ -274,12 +310,15 @@ public sealed class PerPlatformProbeWatcher : IDisposable /// 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; - - var liveHandles = _itemHandlesByPlatform.Values.Where(h => h > 0).ToArray(); - _itemHandlesByPlatform.Clear(); - _registry?.Remove(SubscriptionRegistry.ProbeSubscriptionId); + int[] liveHandles; + lock (_syncLock) + { + if (_disposed) return; + _disposed = true; + liveHandles = [.. _itemHandlesByPlatform.Values.Where(h => h > 0)]; + _itemHandlesByPlatform.Clear(); + _registry?.Remove(SubscriptionRegistry.ProbeSubscriptionId); + } if (liveHandles.Length > 0) { try { await _subscriber.UnsubscribeBulkAsync(liveHandles, cancellationToken).ConfigureAwait(false); } diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/GalaxyDriverFixRound1Tests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/GalaxyDriverFixRound1Tests.cs index 3a7b241e..9876c09f 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/GalaxyDriverFixRound1Tests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/GalaxyDriverFixRound1Tests.cs @@ -21,9 +21,9 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests.Runtime; /// public sealed class GalaxyDriverFixRound1Tests { - private const int WaitMs = 10_000; + internal const int WaitMs = 10_000; - private static GalaxyDriverOptions Opts(bool watchDeploy = false) => new( + internal static GalaxyDriverOptions Opts(bool watchDeploy = false) => new( new GalaxyGatewayOptions("https://mxgw.test:5001", "key"), new GalaxyMxAccessOptions("FixRound1"), new GalaxyRepositoryOptions(WatchDeployEvents: watchDeploy), @@ -41,7 +41,7 @@ public sealed class GalaxyDriverFixRound1Tests => Task.FromResult(Script(Interlocked.Increment(ref _calls))); } - private sealed class ParkableSubscriber : IGalaxySubscriber + internal sealed class ParkableSubscriber : IGalaxySubscriber { private readonly Channel _events = Channel.CreateUnbounded(); private int _nextHandle; @@ -86,8 +86,12 @@ public sealed class GalaxyDriverFixRound1Tests return results; } + /// Every handle passed to UnsubscribeBulk, in call order. + public List Unsubscribed { get; } = []; + public async Task UnsubscribeBulkAsync(IReadOnlyList itemHandles, CancellationToken cancellationToken) { + lock (Unsubscribed) Unsubscribed.AddRange(itemHandles); if (ParkUnsubscribe) { UnsubscribeEntered.TrySetResult(); @@ -113,7 +117,7 @@ public sealed class GalaxyDriverFixRound1Tests }); } - private sealed class ChannelDeployWatchSource : IGalaxyDeployWatchSource + internal sealed class ChannelDeployWatchSource : IGalaxyDeployWatchSource { public Channel Events { get; } = Channel.CreateUnbounded(); public int Opened; @@ -136,7 +140,7 @@ public sealed class GalaxyDriverFixRound1Tests }); } - private static async Task WaitUntilAsync(Func condition, string because) + internal static async Task WaitUntilAsync(Func condition, string because) { var deadline = DateTime.UtcNow.AddMilliseconds(WaitMs); while (!condition()) @@ -146,7 +150,7 @@ public sealed class GalaxyDriverFixRound1Tests } } - private static HostState? StateOf(GalaxyDriver driver, string host) => + internal static HostState? StateOf(GalaxyDriver driver, string host) => driver.GetHostStatuses().FirstOrDefault(s => s.HostName == host)?.State; // ───────────────────────────── IMPORTANT 1: seed at init, re-seed on rediscovery ───────────────────────────── @@ -161,8 +165,11 @@ public sealed class GalaxyDriverFixRound1Tests await driver.InitializeAsync("{}", CancellationToken.None); + // The seed runs in the background (fix round 2, N1) — init does not wait on it. + await WaitUntilAsync(() => driver.ProbeWatcherForTests?.WatchedPlatforms.Count == 1, + "the probe watcher must be seeded after init"); platforms.Calls.ShouldBe(1); - sub.Subscribed.ShouldContain("Plat01.ScanState", "the probe watcher must be seeded at init"); + lock (sub.Subscribed) sub.Subscribed.ShouldContain("Plat01.ScanState"); driver.ProbeWatcherForTests.ShouldNotBeNull().WatchedPlatforms.ShouldBe(["Plat01"]); await driver.SubscribeAsync(["Tank01.Level"], TimeSpan.Zero, CancellationToken.None); // starts the pump @@ -188,6 +195,7 @@ public sealed class GalaxyDriverFixRound1Tests await driver.InitializeAsync("{}", CancellationToken.None); driver.GetHealth().State.ShouldBe(DriverState.Healthy, "a platform-seed failure must not fail init"); + await WaitUntilAsync(() => platforms.Calls == 1, "the (background) init seed must have run"); sub.Subscribed.ShouldBeEmpty(); await WaitUntilAsync(() => Volatile.Read(ref deploys.Opened) == 1, "the deploy watch must open"); @@ -271,7 +279,7 @@ public sealed class GalaxyDriverFixRound1Tests await driver.InitializeAsync("{}", CancellationToken.None); - platforms.Calls.ShouldBe(1, "the owned runtime must seed its probe set at init"); + await WaitUntilAsync(() => platforms.Calls == 1, "the owned runtime must seed its probe set after init"); driver.GetHealth().State.ShouldBe(DriverState.Healthy); } @@ -289,6 +297,8 @@ public sealed class GalaxyDriverFixRound1Tests var platforms = new ScriptedPlatformSource(); // gives the probe watcher a handle to unsubscribe await using var driver = new GalaxyDriver("drv", Opts(), hierarchySource: null, subscriber: sub, platformSource: platforms); await driver.InitializeAsync("{}", CancellationToken.None); + await WaitUntilAsync(() => driver.ProbeWatcherForTests?.WatchedPlatforms.Count == 1, + "the (background) seed must give the probe watcher its handle first"); await driver.SubscribeAsync(["Tank01.Level"], TimeSpan.Zero, CancellationToken.None); driver.HasEventPumpForTests.ShouldBeTrue("positive control: subscribe works on a live runtime"); diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/GalaxyDriverFixRound2Tests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/GalaxyDriverFixRound2Tests.cs new file mode 100644 index 00000000..ddd97023 --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/GalaxyDriverFixRound2Tests.cs @@ -0,0 +1,313 @@ +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; +using ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Browse; +using ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Health; +using ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Runtime; +using static ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests.Runtime.GalaxyDriverFixRound1Tests; + +namespace ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests.Runtime; + +/// +/// Lane F fix round 2 (re-review of #735): +/// +/// N1 — the init platform seed runs in the background: a slow or cancelled repository +/// fetch can neither delay nor fail InitializeAsync (Galaxy's Tier-A Initialize budget is +/// 5 s, and a cold gateway cache can spend all of it), and a seed that lands after a teardown +/// is discarded. +/// N2 — a seed requested under a new runtime always gets a pass of its own, even when a +/// worker from the previous runtime is still running. +/// N3 — a subscribe / subscribe-once read whose SubscribeBulk straddles a teardown fails +/// and unsubscribes instead of registering into the next runtime's registry. +/// The probe watcher's own subscribe re-checks its stop after the await. +/// +/// +public sealed class GalaxyDriverFixRound2Tests +{ + private static readonly TimeSpan InitBudget = TimeSpan.FromSeconds(2); + + /// A platform source whose fetches park until released (optionally ignoring cancellation). + private sealed class GatedPlatformSource : IGalaxyPlatformSource + { + private int _calls; + private readonly TaskCompletionSource _release = new(TaskCreationOptions.RunContinuationsAsynchronously); + + public int Calls => Volatile.Read(ref _calls); + public int ParkCall { get; init; } = 1; + public bool IgnoreCancellation { get; init; } + public Func> Result { get; init; } = _ => ["Plat01"]; + public TaskCompletionSource Entered { get; } = new(TaskCreationOptions.RunContinuationsAsynchronously); + public TaskCompletionSource ParkedToken { get; } = new(TaskCreationOptions.RunContinuationsAsynchronously); + public void Release() => _release.TrySetResult(); + + public async Task> GetPlatformTagNamesAsync(CancellationToken cancellationToken) + { + var call = Interlocked.Increment(ref _calls); + if (call == ParkCall) + { + ParkedToken.TrySetResult(cancellationToken); + Entered.TrySetResult(); + if (IgnoreCancellation) await _release.Task; + else await _release.Task.WaitAsync(cancellationToken); + } + return Result(call); + } + } + + private sealed class ThrowingPlatformSource(Exception ex) : IGalaxyPlatformSource + { + private int _calls; + public int Calls => Volatile.Read(ref _calls); + + public Task> GetPlatformTagNamesAsync(CancellationToken cancellationToken) + { + Interlocked.Increment(ref _calls); + return Task.FromException>(ex); + } + } + + private static async Task Entered(TaskCompletionSource tcs, string because) => + (await Task.WhenAny(tcs.Task, Task.Delay(WaitMs))).ShouldBe(tcs.Task, because); + + // ───────────────────────────── N1: background init seed ───────────────────────────── + + /// + /// A repository fetch that never completes cannot hold InitializeAsync: init returns Healthy + /// inside its budget, and the teardown cancels the parked fetch. + /// + [Fact] + public async Task InitializeAsync_returns_Healthy_while_the_platform_fetch_never_completes() + { + var sub = new ParkableSubscriber(); + var platforms = new GatedPlatformSource(); + await using var driver = new GalaxyDriver("drv", Opts(), hierarchySource: null, subscriber: sub, platformSource: platforms); + try + { + var sw = System.Diagnostics.Stopwatch.StartNew(); + await driver.InitializeAsync("{}", CancellationToken.None).WaitAsync(InitBudget); + sw.Stop(); + + driver.GetHealth().State.ShouldBe(DriverState.Healthy); + await Entered(platforms.Entered, "positive control: the seed fetch was started"); + sw.Elapsed.ShouldBeLessThan(InitBudget); + + await driver.ShutdownAsync(CancellationToken.None).WaitAsync(TimeSpan.FromMilliseconds(WaitMs)); + var token = await platforms.ParkedToken.Task; + token.IsCancellationRequested.ShouldBeTrue("the teardown must cancel a seed still in flight"); + } + finally + { + platforms.Release(); + } + } + + /// A fetch that is cancelled (the gateway's own deadline) does not fail init. + [Fact] + public async Task A_cancelled_platform_fetch_does_not_fail_InitializeAsync() + { + var sub = new ParkableSubscriber(); + var platforms = new ThrowingPlatformSource(new OperationCanceledException("gateway deadline")); + await using var driver = new GalaxyDriver("drv", Opts(), hierarchySource: null, subscriber: sub, platformSource: platforms); + + await driver.InitializeAsync("{}", CancellationToken.None).WaitAsync(InitBudget); + + driver.GetHealth().State.ShouldBe(DriverState.Healthy); + await WaitUntilAsync(() => platforms.Calls == 1, "positive control: the seed fetch ran"); + driver.ProbeWatcherForTests.ShouldNotBeNull().WatchedPlatforms.ShouldBeEmpty(); + + // The driver keeps serving after the failed seed. + await driver.SubscribeAsync(["Tank01.Level"], TimeSpan.Zero, CancellationToken.None); + driver.TrackedSubscriptionCountForTests.ShouldBe(1); + } + + /// + /// The production failure mode: the host's Tier-A Initialize timeout cancels the init token + /// while the (cold-cache) fetch is still running. Init must already have returned Healthy, and + /// the seed — tied to the runtime, not to the init call — must survive that cancellation and + /// still land. + /// + [Fact] + public async Task Init_token_cancellation_mid_fetch_neither_fails_init_nor_cancels_the_seed() + { + var sub = new ParkableSubscriber(); + var platforms = new GatedPlatformSource(); + await using var driver = new GalaxyDriver("drv", Opts(), hierarchySource: null, subscriber: sub, platformSource: platforms); + using var initCts = new CancellationTokenSource(); + try + { + initCts.CancelAfter(TimeSpan.FromMilliseconds(300)); // the Polly Initialize timeout, shrunk + await driver.InitializeAsync("{}", initCts.Token).WaitAsync(InitBudget); + driver.GetHealth().State.ShouldBe(DriverState.Healthy); + + await Entered(platforms.Entered, "the seed fetch must be in flight"); + await WaitUntilAsync(() => initCts.IsCancellationRequested, "the init token fires while the fetch is parked"); + (await platforms.ParkedToken.Task).IsCancellationRequested.ShouldBeFalse( + "the seed belongs to the runtime; the init call's timeout must not cancel it"); + } + finally + { + platforms.Release(); + } + + await WaitUntilAsync(() => driver.ProbeWatcherForTests?.WatchedPlatforms.Count == 1, + "the seed must still land after the init token was cancelled"); + } + + /// Positive control: a fetch that completes after init still populates host statuses. + [Fact] + public async Task A_fetch_completing_after_init_populates_host_statuses() + { + var sub = new ParkableSubscriber(); + var platforms = new GatedPlatformSource(); + await using var driver = new GalaxyDriver("drv", Opts(), hierarchySource: null, subscriber: sub, platformSource: platforms); + try + { + await driver.InitializeAsync("{}", CancellationToken.None).WaitAsync(InitBudget); + await Entered(platforms.Entered, "the seed must be parked"); + driver.ProbeWatcherForTests.ShouldNotBeNull().WatchedPlatforms.ShouldBeEmpty(); + } + finally + { + platforms.Release(); + } + + await WaitUntilAsync(() => driver.ProbeWatcherForTests?.WatchedPlatforms.Count == 1, + "the late seed must reach the probe watcher"); + await driver.SubscribeAsync(["Tank01.Level"], TimeSpan.Zero, CancellationToken.None); // starts the pump + await sub.EmitBoolAsync(sub.HandleByRef["Plat01.ScanState"], true); + await WaitUntilAsync(() => StateOf(driver, "Plat01") == HostState.Running, + "a ScanState event must surface the late-seeded platform"); + } + + // ───────────────────────────── N1 generation + N2: straddling worker ───────────────────────────── + + /// + /// A seed from the previous runtime that completes after a teardown + re-init is discarded, + /// and the new runtime's own seed request still gets its own pass. + /// + [Fact] + public async Task A_seed_straddling_a_teardown_is_discarded_and_the_new_runtime_is_seeded() + { + var sub = new ParkableSubscriber(); + var platforms = new GatedPlatformSource + { + IgnoreCancellation = true, // a fetch that finishes regardless — only the generation check can stop it + Result = call => call == 1 ? ["Stale01"] : ["Plat01"], + }; + await using var driver = new GalaxyDriver("drv", Opts(), hierarchySource: null, subscriber: sub, platformSource: platforms); + try + { + await driver.InitializeAsync("{}", CancellationToken.None).WaitAsync(InitBudget); + await Entered(platforms.Entered, "runtime 1's seed must be parked"); + + await driver.ShutdownAsync(CancellationToken.None).WaitAsync(TimeSpan.FromMilliseconds(WaitMs)); + await driver.InitializeAsync("{}", CancellationToken.None).WaitAsync(InitBudget); + } + finally + { + platforms.Release(); + } + + await WaitUntilAsync(() => platforms.Calls == 2, "the request made under the new runtime must get its own pass"); + await WaitUntilAsync(() => driver.ProbeWatcherForTests?.WatchedPlatforms.Count == 1, + "the new runtime must be seeded"); + driver.ProbeWatcherForTests!.WatchedPlatforms.ShouldBe(["Plat01"]); + lock (sub.Subscribed) sub.Subscribed.ShouldNotContain("Stale01.ScanState", "runtime 1's late result must be discarded"); + } + + // ───────────────────────────── probe watcher: re-check after the SubscribeBulk await ───────────────────────────── + + /// + /// A watcher sync whose SubscribeBulk completes after StopAsync must not write the probe entry + /// back into the registry, and must release the handles it just acquired. Positive control: + /// the same sync without a stop registers. + /// + [Fact] + public async Task Watcher_sync_straddling_a_stop_does_not_resurrect_the_probe_entry() + { + { + var registry = new SubscriptionRegistry(); + using var watcher = new PerPlatformProbeWatcher(new ParkableSubscriber(), new HostStatusAggregator(), registry); + await watcher.SyncPlatformsAsync(["A"], CancellationToken.None); + registry.Contains(SubscriptionRegistry.ProbeSubscriptionId).ShouldBeTrue("positive control"); + } + + { + var registry = new SubscriptionRegistry(); + var sub = new ParkableSubscriber { ParkSubscribeCall = 1 }; + var watcher = new PerPlatformProbeWatcher(sub, new HostStatusAggregator(), registry); + + var sync = watcher.SyncPlatformsAsync(["A"], CancellationToken.None); + await Entered(sub.SubscribeEntered, "the sync must be parked in SubscribeBulk"); + await watcher.StopAsync(CancellationToken.None); + + sub.ReleaseSubscribe(); + await sync.WaitAsync(TimeSpan.FromMilliseconds(WaitMs)); + + registry.Contains(SubscriptionRegistry.ProbeSubscriptionId).ShouldBeFalse("a stopped watcher must not re-register"); + watcher.WatchedPlatforms.ShouldBeEmpty(); + lock (sub.Unsubscribed) sub.Unsubscribed.ShouldContain(sub.HandleByRef["A.ScanState"], + "the handle acquired after the stop must be released"); + } + } + + // ───────────────────────────── N3: subscribe / read straddling a teardown ───────────────────────────── + + /// + /// A SubscribeAsync whose SubscribeBulk completes after the teardown fails, registers nothing + /// and unsubscribes its handles; a subscribe on the live runtime registers (positive control). + /// + [Fact] + public async Task SubscribeAsync_straddling_a_teardown_fails_and_registers_nothing() + { + var sub = new ParkableSubscriber { ParkSubscribeCall = 2 }; + await using var driver = new GalaxyDriver("drv", Opts(), hierarchySource: null, subscriber: sub); + await driver.InitializeAsync("{}", CancellationToken.None); + + await driver.SubscribeAsync(["Tank01.Level"], TimeSpan.Zero, CancellationToken.None); + driver.TrackedSubscriptionCountForTests.ShouldBe(1, "positive control: a live-runtime subscribe registers"); + + var straddling = driver.SubscribeAsync(["Tank02.Level"], TimeSpan.Zero, CancellationToken.None); + try + { + await Entered(sub.SubscribeEntered, "the subscribe must be parked in SubscribeBulk"); + await driver.ShutdownAsync(CancellationToken.None).WaitAsync(TimeSpan.FromMilliseconds(WaitMs)); + driver.TrackedSubscriptionCountForTests.ShouldBe(0); + } + finally + { + sub.ReleaseSubscribe(); + } + + await Should.ThrowAsync(() => straddling.WaitAsync(TimeSpan.FromMilliseconds(WaitMs))); + driver.TrackedSubscriptionCountForTests.ShouldBe(0, "the straddling subscribe must not land in the registry"); + lock (sub.Unsubscribed) sub.Unsubscribed.ShouldContain(sub.HandleByRef["Tank02.Level"]); + + await driver.InitializeAsync("{}", CancellationToken.None); + driver.TrackedSubscriptionCountForTests.ShouldBe(0, "the next runtime starts with an empty registry"); + } + + /// The subscribe-once read path: same straddle, same outcome. + [Fact] + public async Task Subscribe_once_read_straddling_a_teardown_fails_and_registers_nothing() + { + var sub = new ParkableSubscriber { ParkSubscribeCall = 1 }; + await using var driver = new GalaxyDriver("drv", Opts(), hierarchySource: null, subscriber: sub); + await driver.InitializeAsync("{}", CancellationToken.None); + + var read = driver.ReadAsync(["Tank01.Level"], CancellationToken.None); + try + { + await Entered(sub.SubscribeEntered, "the read must be parked in SubscribeBulk"); + await driver.ShutdownAsync(CancellationToken.None).WaitAsync(TimeSpan.FromMilliseconds(WaitMs)); + } + finally + { + sub.ReleaseSubscribe(); + } + + await Should.ThrowAsync(() => read.WaitAsync(TimeSpan.FromMilliseconds(WaitMs))); + driver.TrackedSubscriptionCountForTests.ShouldBe(0); + lock (sub.Unsubscribed) sub.Unsubscribed.ShouldContain(sub.HandleByRef["Tank01.Level"]); + } +}