diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Browse/GatewayGalaxyPlatformSource.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Browse/GatewayGalaxyPlatformSource.cs new file mode 100644 index 00000000..dfed1659 --- /dev/null +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Browse/GatewayGalaxyPlatformSource.cs @@ -0,0 +1,88 @@ +using ZB.MOM.WW.MxGateway.Client; +using ZB.MOM.WW.MxGateway.Contracts.Proto.Galaxy; + +namespace ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Browse; + +/// +/// Default : a platform-only DiscoverHierarchy sweep +/// over the gateway's — the request filters on +/// template_chain_contains ($WinPlatform, $AppEngine) and asks for no +/// attributes, so it is a few rows per galaxy rather than the full object + attribute dump the +/// address-space discovery pulls. Paged at the configured Repository.DiscoverPageSize. +/// +public sealed class GatewayGalaxyPlatformSource : IGalaxyPlatformSource +{ + /// Template-chain tokens that identify a platform / engine object. + public static readonly IReadOnlyList PlatformTemplates = ["$WinPlatform", "$AppEngine"]; + + private readonly GalaxyRepositoryClient _client; + private readonly int _pageSize; + + /// Initializes a new instance of the class. + /// The gateway's Galaxy repository client. + /// Objects per page (Repository.DiscoverPageSize); non-positive → the client default. + public GatewayGalaxyPlatformSource(GalaxyRepositoryClient client, int pageSize = GatewayGalaxyHierarchySource.DefaultPageSize) + { + _client = client ?? throw new ArgumentNullException(nameof(client)); + _pageSize = GatewayGalaxyHierarchySource.EffectivePageSize(pageSize); + } + + /// + public Task> GetPlatformTagNamesAsync(CancellationToken cancellationToken) + => QueryAsync(_client.DiscoverHierarchyRawAsync, _pageSize, cancellationToken); + + /// + /// The platform-only sweep behind a delegate so it is unit-testable without a channel. Every + /// request carries the template filter and include_attributes = false; the reply is + /// re-filtered on the returned template chain (a gateway that ignores the filter still + /// yields only platforms) and reduced to non-empty tag names. + /// + /// The raw DiscoverHierarchy RPC. + /// Objects per page. + /// Cancellation token for the walk. + /// Distinct platform tag names in gateway order. + internal static async Task> QueryAsync( + Func> discover, + int pageSize, + CancellationToken cancellationToken) + { + ArgumentNullException.ThrowIfNull(discover); + var effective = GatewayGalaxyHierarchySource.EffectivePageSize(pageSize); + + var names = new List(); + var seen = new HashSet(StringComparer.OrdinalIgnoreCase); + var seenPageTokens = new HashSet(StringComparer.Ordinal); + var pageToken = string.Empty; + do + { + var request = new DiscoverHierarchyRequest + { + PageSize = effective, + PageToken = pageToken, + IncludeAttributes = false, + }; + request.TemplateChainContains.Add(PlatformTemplates); + + var reply = await discover(request, cancellationToken).ConfigureAwait(false); + foreach (var obj in reply.Objects) + { + if (string.IsNullOrEmpty(obj.TagName) || !IsPlatform(obj)) continue; + if (seen.Add(obj.TagName)) names.Add(obj.TagName); + } + + pageToken = reply.NextPageToken ?? string.Empty; + if (!string.IsNullOrWhiteSpace(pageToken) && !seenPageTokens.Add(pageToken)) + { + throw new InvalidOperationException( + $"Galaxy DiscoverHierarchy (platform query) returned a repeated page token '{pageToken}' — refusing to loop."); + } + } + while (!string.IsNullOrWhiteSpace(pageToken)); + + return names; + } + + private static bool IsPlatform(GalaxyObject obj) => + obj.TemplateChain.Count == 0 // a gateway that strips the chain already filtered server-side + || obj.TemplateChain.Any(t => PlatformTemplates.Contains(t, StringComparer.OrdinalIgnoreCase)); +} diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Browse/IGalaxyPlatformSource.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Browse/IGalaxyPlatformSource.cs new file mode 100644 index 00000000..6589c9aa --- /dev/null +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Browse/IGalaxyPlatformSource.cs @@ -0,0 +1,18 @@ +namespace ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Browse; + +/// +/// Driver-side seam for the set of Galaxy platform / engine objects whose ScanState +/// the PerPlatformProbeWatcher advises. Production wraps the gateway's +/// DiscoverHierarchy RPC filtered to $WinPlatform / $AppEngine with no +/// attributes (); tests substitute a canned list. +/// The driver seeds the watcher from this at InitializeAsync and re-seeds on every +/// deploy-watcher rediscovery — the runtime never calls DiscoverAsync on a live driver, +/// so without this seed the probe watcher never had a platform set in production. +/// +public interface IGalaxyPlatformSource +{ + /// The tag_name of every deployed $WinPlatform / $AppEngine object. + /// Cancellation token for the query. + /// Platform tag names; empty when the galaxy has none. + Task> GetPlatformTagNamesAsync(CancellationToken cancellationToken); +} 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 8cb4bcb5..8fc00fa8 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/GalaxyDriver.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/GalaxyDriver.cs @@ -150,6 +150,31 @@ public sealed class GalaxyDriver // GatewayGalaxyDeployWatchSource over the owned GalaxyRepositoryClient. private readonly IGalaxyDeployWatchSource? _deployWatchSource; + // Platform set for the ScanState probe watcher. Injected seam (tests) or, on the production + // path, a platform-only DiscoverHierarchy query over the owned repository client. Seeded at + // InitializeAsync and re-seeded on every deploy-watcher rediscovery: the runtime never calls + // DiscoverAsync on a live driver (only the /raw picker's throwaway does), so SyncPlatformsAsync + // 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. + private int _reseedRequests; + + // 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 + // late Rebind cannot resurrect handles into a registry the teardown already cleared. + private int _runtimeGeneration; + + /// + /// Upper bound on the gateway CloseSession round-trip during a teardown. The SDK's + /// default close runs under the 30 s call timeout, which against a blackholed gateway parks + /// ShutdownAsync — and the host's PostStop — for the whole of it. Internal so + /// tests can shrink it. + /// + internal TimeSpan SessionCloseDeadline { get; set; } = TimeSpan.FromSeconds(5); + // 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 @@ -258,6 +283,11 @@ public sealed class GalaxyDriver /// Optional deploy-event stream seam. When supplied, starts the /// over it (production builds one over the owned repository client). /// + /// + /// Optional platform-set seam for the ScanState probe watcher. When supplied, + /// seeds the watcher from it and every rediscovery re-seeds + /// (production builds a over the owned repository client). + /// internal GalaxyDriver( string driverInstanceId, GalaxyDriverOptions options, @@ -269,9 +299,11 @@ public sealed class GalaxyDriver IGalaxyAlarmFeed? alarmFeed = null, ILogger? logger = null, ISecretResolver? secretResolver = null, - IGalaxyDeployWatchSource? deployWatchSource = null) + IGalaxyDeployWatchSource? deployWatchSource = null, + IGalaxyPlatformSource? platformSource = null) { _deployWatchSource = deployWatchSource; + _platformSource = platformSource; _driverInstanceId = !string.IsNullOrWhiteSpace(driverInstanceId) ? driverInstanceId : throw new ArgumentException("Driver instance id required.", nameof(driverInstanceId)); @@ -443,6 +475,7 @@ public sealed class GalaxyDriver } await StartDeployWatcherAsync(cancellationToken).ConfigureAwait(false); + await SeedProbePlatformsAsync("initialize", cancellationToken).ConfigureAwait(false); _logger.LogInformation( "GalaxyDriver {InstanceId} initialized — endpoint={Endpoint} clientName={ClientName}", _driverInstanceId, _options.Gateway.Endpoint, _options.MxAccess.ClientName); @@ -454,6 +487,92 @@ 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). + /// + private async Task SeedProbePlatformsAsync(string reason, CancellationToken cancellationToken) + { + var watcher = _probeWatcher; + if (watcher is null) return; + + IGalaxyPlatformSource? source; + try + { + source = await ResolvePlatformSourceAsync(cancellationToken).ConfigureAwait(false); + } + catch (Exception ex) + { + _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; + } + if (source is null) return; + + try + { + var platforms = await source.GetPlatformTagNamesAsync(cancellationToken).ConfigureAwait(false); + 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) + { + _logger.LogWarning(ex, + "GalaxyDriver {InstanceId} platform seed failed ({Reason}) — per-platform host status stays as it was until the next rediscovery.", + _driverInstanceId, reason); + } + } + + 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; + + _ownedRepositoryClient ??= GalaxyRepositoryClient.Create( + await BuildClientOptionsAsync(_options.Gateway, cancellationToken).ConfigureAwait(false)); + return _ownedPlatformSource = + new GatewayGalaxyPlatformSource(_ownedRepositoryClient, _options.Repository.DiscoverPageSize); + } + + /// + /// 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. + /// + private void ReseedProbePlatformsOnRediscovery() + { + 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 () => + { + 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); + }); + } + // 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. @@ -488,6 +607,18 @@ public sealed class GalaxyDriver private async Task BuildProductionRuntimeCoreAsync(CancellationToken cancellationToken) { + // A pump that slipped in between a teardown and this rebuild (a late SubscribeAsync) is bound + // to the dead subscriber; it must never be reused as this runtime's pump. + EventPump? stale; + lock (_pumpLock) { stale = _eventPump; _eventPump = null; } + if (stale is not null) + { + stale.OnDataChange -= OnPumpDataChange; + try { await stale.DisposeAsync().ConfigureAwait(false); } + catch (Exception ex) { _logger.LogWarning(ex, "Stale EventPump dispose failed before runtime rebuild"); } + _subscriptions.Clear(); + } + _ownsRuntime = true; var clientOptions = await BuildClientOptionsAsync(_options.Gateway, cancellationToken).ConfigureAwait(false); _ownedMxClient = MxGatewayClient.Create(clientOptions); @@ -562,6 +693,7 @@ public sealed class GalaxyDriver private async Task TeardownRuntimeAsync(bool final, CancellationToken cancellationToken) { _runtimeStopping = true; + Interlocked.Increment(ref _runtimeGeneration); try { var supervisor = _supervisor; @@ -628,9 +760,16 @@ public sealed class GalaxyDriver if (_ownedMxSession is not null) { - try { await _ownedMxSession.DisposeAsync().ConfigureAwait(false); } - catch (Exception ex) { _logger.LogWarning(ex, "MxSession dispose failed"); } + var session = _ownedMxSession; _ownedMxSession = null; + // Bounded close: the caller's token plus SessionCloseDeadline, so a blackholed gateway + // cannot park this (and the host's PostStop) for the SDK's 30 s call timeout. + using var closeCts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken); + closeCts.CancelAfter(SessionCloseDeadline); + try { await session.CloseAsync(closeCts.Token).ConfigureAwait(false); } + catch (Exception ex) { _logger.LogWarning(ex, "MxSession close failed"); } + try { await session.DisposeAsync().ConfigureAwait(false); } + catch (Exception ex) { _logger.LogWarning(ex, "MxSession dispose failed"); } } if (_ownedMxClient is not null) @@ -655,6 +794,7 @@ public sealed class GalaxyDriver _dataWriter = null; _hierarchySource = null; _alarmAcknowledger = null; + _ownedPlatformSource = null; _ownsRuntime = false; } } @@ -719,7 +859,8 @@ public sealed class GalaxyDriver // completed loop. Recreate the pump so the replayed subscriptions have a consumer. RestartEventPumpForReplay(); - var tagCount = await ResubscribeAllAsync(entries, cancellationToken).ConfigureAwait(false); + var tagCount = await ResubscribeAllAsync(entries, Volatile.Read(ref _runtimeGeneration), cancellationToken) + .ConfigureAwait(false); _logger.LogInformation( "GalaxyDriver {InstanceId} replay completed — {SubCount} subscriptions, {TagCount} tags re-subscribed", @@ -733,14 +874,28 @@ public sealed class GalaxyDriver /// replay must recreate (its loop died with the stream) and a gap must NOT touch (the /// stream carrying the sentinel is still healthy and still delivering). /// + /// The subscriptions to re-issue. + /// + /// The runtime generation this re-subscribe belongs to. Re-checked before every + /// : a teardown bumps it and clears the registry, + /// and a Rebind landing after that would resurrect dead handles into the next runtime. + /// + /// Cancellation token. /// The number of tag references re-subscribed. private async Task ResubscribeAllAsync( IReadOnlyList<(long SubscriptionId, IReadOnlyList Bindings)> entries, + int generation, CancellationToken cancellationToken) { var tagCount = 0; foreach (var (subscriptionId, oldBindings) in entries) { + if (Volatile.Read(ref _runtimeGeneration) != generation) + { + _logger.LogDebug("GalaxyDriver {InstanceId} re-subscribe abandoned — runtime was torn down.", _driverInstanceId); + return tagCount; + } + if (subscriptionId == SubscriptionRegistry.ProbeSubscriptionId && _probeWatcher is { } probeWatcher) { // The probe watcher owns its handles: it re-subscribes its platforms, records the @@ -771,7 +926,9 @@ 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. + // 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); tagCount += refs.Length; } @@ -915,8 +1072,12 @@ public sealed class GalaxyDriver if (startTask.IsFaulted) startTask.GetAwaiter().GetResult(); } - private void OnDeployWatcherRediscoveryNeeded(object? sender, RediscoveryEventArgs args) => + private void OnDeployWatcherRediscoveryNeeded(object? sender, RediscoveryEventArgs args) + { OnRediscoveryNeeded?.Invoke(this, args); + // A redeploy may add or remove platforms / engines; refresh the ScanState probe set. + ReseedProbePlatformsOnRediscovery(); + } /// public Task ReinitializeAsync(string driverConfigJson, CancellationToken cancellationToken) @@ -1437,6 +1598,15 @@ public sealed class GalaxyDriver { lock (_pumpLock) { + // Checked BEFORE the reuse: the flag is set before the teardown takes the pump under this + // same lock, so while it is set the live pump belongs to a dying subscriber. Handing it out + // would register handles into a registry the teardown is about to clear; creating a new one + // would leave a pump on the dead subscriber for the next InitializeAsync to inherit. + if (_runtimeStopping) + { + throw new InvalidOperationException( + "GalaxyDriver runtime is shutting down — no subscription can be served until the next InitializeAsync."); + } if (_eventPump is not null) return _eventPump; _eventPump = new EventPump( _subscriber!, _subscriptions, _logger, @@ -1497,6 +1667,7 @@ public sealed class GalaxyDriver return; } + var generation = Volatile.Read(ref _runtimeGeneration); _ = Task.Run(async () => { try @@ -1504,7 +1675,7 @@ public sealed class GalaxyDriver var entries = _subscriptions.SnapshotEntries(); if (entries.Count == 0) return; - var tagCount = await ResubscribeAllAsync(entries, CancellationToken.None).ConfigureAwait(false); + var tagCount = await ResubscribeAllAsync(entries, generation, CancellationToken.None).ConfigureAwait(false); _logger.LogInformation( "GalaxyDriver {InstanceId} replay-gap re-snapshot completed — {SubCount} subscriptions, {TagCount} tags re-subscribed", _driverInstanceId, entries.Count, tagCount); diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Browse/GatewayGalaxyPlatformSourceTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Browse/GatewayGalaxyPlatformSourceTests.cs new file mode 100644 index 00000000..a385ecfb --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Browse/GatewayGalaxyPlatformSourceTests.cs @@ -0,0 +1,60 @@ +using Shouldly; +using Xunit; +using ZB.MOM.WW.MxGateway.Contracts.Proto.Galaxy; +using ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Browse; + +namespace ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests.Browse; + +/// The platform-only query the driver seeds its ScanState probe watcher from. +public sealed class GatewayGalaxyPlatformSourceTests +{ + private static GalaxyObject Obj(string tag, params string[] chain) + { + var o = new GalaxyObject { TagName = tag }; + o.TemplateChain.Add(chain); + return o; + } + + /// Every request carries the platform template filter, no attributes, and the page size; replies are re-filtered. + [Fact] + public async Task Requests_filter_on_platform_templates_without_attributes_and_replies_are_refiltered() + { + var requests = new List(); + Task Discover(DiscoverHierarchyRequest req, CancellationToken ct) + { + requests.Add(req); + var reply = new DiscoverHierarchyReply(); + if (req.PageToken == "") + { + reply.Objects.Add([Obj("Plat01", "$WinPlatform"), Obj("Tank01", "$UserDefined")]); + reply.NextPageToken = "p2"; + } + else + { + reply.Objects.Add([Obj("Eng01", "$AppEngine", "$WinPlatform"), Obj("Plat01", "$WinPlatform"), Obj("", "$WinPlatform")]); + } + return Task.FromResult(reply); + } + + var names = await GatewayGalaxyPlatformSource.QueryAsync(Discover, pageSize: 50, CancellationToken.None); + + names.ShouldBe(["Plat01", "Eng01"]); + requests.Count.ShouldBe(2); + requests.ShouldAllBe(r => r.PageSize == 50 && r.IncludeAttributes == false); + requests.ShouldAllBe(r => r.TemplateChainContains.Contains("$WinPlatform") && r.TemplateChainContains.Contains("$AppEngine")); + } + + /// A gateway that strips the template chain is trusted to have filtered server-side. + [Fact] + public async Task Objects_without_a_template_chain_are_kept() + { + Task Discover(DiscoverHierarchyRequest req, CancellationToken ct) + { + var reply = new DiscoverHierarchyReply(); + reply.Objects.Add(new GalaxyObject { TagName = "Plat01" }); + return Task.FromResult(reply); + } + + (await GatewayGalaxyPlatformSource.QueryAsync(Discover, pageSize: 0, CancellationToken.None)).ShouldBe(["Plat01"]); + } +} 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 new file mode 100644 index 00000000..3a7b241e --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/GalaxyDriverFixRound1Tests.cs @@ -0,0 +1,396 @@ +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; + +/// +/// Lane F fix round 1 (review of #735): the probe watcher is seeded at InitializeAsync +/// and re-seeded on rediscovery (the runtime never calls DiscoverAsync on a live driver); +/// a SubscribeAsync landing mid-teardown cannot create a pump on the dying subscriber; a +/// gateway close that never answers cannot park ShutdownAsync; a re-subscribe that +/// straddles a teardown never rebinds into the next runtime's registry. +/// +public sealed class GalaxyDriverFixRound1Tests +{ + private const int WaitMs = 10_000; + + private static GalaxyDriverOptions Opts(bool watchDeploy = false) => new( + new GalaxyGatewayOptions("https://mxgw.test:5001", "key"), + new GalaxyMxAccessOptions("FixRound1"), + new GalaxyRepositoryOptions(WatchDeployEvents: watchDeploy), + new GalaxyReconnectOptions()); + + // ───────────────────────────── fakes ───────────────────────────── + + private sealed class ScriptedPlatformSource : IGalaxyPlatformSource + { + private int _calls; + public int Calls => Volatile.Read(ref _calls); + public Func> Script { get; init; } = _ => ["Plat01"]; + + public Task> GetPlatformTagNamesAsync(CancellationToken cancellationToken) + => Task.FromResult(Script(Interlocked.Increment(ref _calls))); + } + + private sealed class ParkableSubscriber : IGalaxySubscriber + { + private readonly Channel _events = Channel.CreateUnbounded(); + private int _nextHandle; + private int _subscribeCalls; + + public Dictionary HandleByRef { get; } = new(StringComparer.OrdinalIgnoreCase); + public List Subscribed { get; } = []; + public int SubscribeCalls => Volatile.Read(ref _subscribeCalls); + + /// When set, UnsubscribeBulk parks until . + public bool ParkUnsubscribe { get; set; } + public TaskCompletionSource UnsubscribeEntered { get; } = new(TaskCreationOptions.RunContinuationsAsynchronously); + private readonly TaskCompletionSource _releaseUnsubscribe = new(TaskCreationOptions.RunContinuationsAsynchronously); + public void ReleaseUnsubscribe() => _releaseUnsubscribe.TrySetResult(); + + /// When > 0, that SubscribeBulk call (1-based) parks until . + public int ParkSubscribeCall { get; set; } + public TaskCompletionSource SubscribeEntered { get; } = new(TaskCreationOptions.RunContinuationsAsynchronously); + private readonly TaskCompletionSource _releaseSubscribe = new(TaskCreationOptions.RunContinuationsAsynchronously); + public void ReleaseSubscribe() => _releaseSubscribe.TrySetResult(); + + public async Task> SubscribeBulkAsync( + IReadOnlyList fullReferences, int bufferedUpdateIntervalMs, CancellationToken cancellationToken) + { + var call = Interlocked.Increment(ref _subscribeCalls); + if (call == ParkSubscribeCall) + { + SubscribeEntered.TrySetResult(); + await _releaseSubscribe.Task.WaitAsync(cancellationToken); + } + var results = new List(); + lock (Subscribed) + { + foreach (var r in fullReferences) + { + var h = Interlocked.Increment(ref _nextHandle); + Subscribed.Add(r); + HandleByRef[r] = h; + results.Add(new SubscribeResult { TagAddress = r, ItemHandle = h, WasSuccessful = true }); + } + } + return results; + } + + public async Task UnsubscribeBulkAsync(IReadOnlyList itemHandles, CancellationToken cancellationToken) + { + if (ParkUnsubscribe) + { + UnsubscribeEntered.TrySetResult(); + await _releaseUnsubscribe.Task.WaitAsync(cancellationToken); + } + } + + public async IAsyncEnumerable StreamEventsAsync([EnumeratorCancellation] CancellationToken cancellationToken) + { + await foreach (var ev in _events.Reader.ReadAllAsync(cancellationToken).ConfigureAwait(false)) + { + yield return ev; + } + } + + public ValueTask EmitBoolAsync(int handle, bool value) => _events.Writer.WriteAsync(new MxEvent + { + Family = MxEventFamily.OnDataChange, + ItemHandle = handle, + Value = new MxValue { BoolValue = value }, + Quality = 192, + SourceTimestamp = Timestamp.FromDateTime(DateTime.UtcNow), + }); + } + + private sealed class ChannelDeployWatchSource : IGalaxyDeployWatchSource + { + public Channel Events { get; } = Channel.CreateUnbounded(); + public int Opened; + + public async IAsyncEnumerable WatchAsync( + DateTimeOffset? lastSeenDeployTime, [EnumeratorCancellation] CancellationToken cancellationToken) + { + Interlocked.Increment(ref Opened); + await foreach (var ev in Events.Reader.ReadAllAsync(cancellationToken).ConfigureAwait(false)) + { + yield return ev; + } + } + + public ValueTask PushAsync(ulong sequence, DateTimeOffset deployTime) => Events.Writer.WriteAsync(new DeployEvent + { + Sequence = sequence, + TimeOfLastDeployPresent = true, + TimeOfLastDeploy = Timestamp.FromDateTimeOffset(deployTime), + }); + } + + 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); + } + } + + private static HostState? StateOf(GalaxyDriver driver, string host) => + driver.GetHostStatuses().FirstOrDefault(s => s.HostName == host)?.State; + + // ───────────────────────────── IMPORTANT 1: seed at init, re-seed on rediscovery ───────────────────────────── + + /// A fresh InitializeAsync with NO DiscoverAsync yields per-platform host statuses. + [Fact] + public async Task InitializeAsync_seeds_the_probe_set_and_ScanState_updates_host_status_without_DiscoverAsync() + { + var sub = new ParkableSubscriber(); + var platforms = new ScriptedPlatformSource(); + await using var driver = new GalaxyDriver("drv", Opts(), hierarchySource: null, subscriber: sub, platformSource: platforms); + + await driver.InitializeAsync("{}", CancellationToken.None); + + platforms.Calls.ShouldBe(1); + sub.Subscribed.ShouldContain("Plat01.ScanState", "the probe watcher must be seeded at init"); + driver.ProbeWatcherForTests.ShouldNotBeNull().WatchedPlatforms.ShouldBe(["Plat01"]); + + 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 seeded platform on GetHostStatuses"); + } + + /// A failed seed does not fail init; the next deploy-watcher rediscovery re-seeds. + [Fact] + public async Task Failed_seed_does_not_fail_init_and_rediscovery_reseeds() + { + var sub = new ParkableSubscriber(); + var deploys = new ChannelDeployWatchSource(); + var platforms = new ScriptedPlatformSource + { + Script = call => call == 1 ? throw new IOException("repository unavailable") : ["Plat01", "Eng01"], + }; + await using var driver = new GalaxyDriver("drv", Opts(watchDeploy: true), hierarchySource: null, + subscriber: sub, platformSource: platforms, deployWatchSource: deploys); + var raised = 0; + driver.OnRediscoveryNeeded += (_, _) => Interlocked.Increment(ref raised); + + await driver.InitializeAsync("{}", CancellationToken.None); + driver.GetHealth().State.ShouldBe(DriverState.Healthy, "a platform-seed failure must not fail init"); + sub.Subscribed.ShouldBeEmpty(); + + await WaitUntilAsync(() => Volatile.Read(ref deploys.Opened) == 1, "the deploy watch must open"); + await deploys.PushAsync(1, new DateTimeOffset(2026, 1, 1, 0, 0, 0, TimeSpan.Zero)); // bootstrap (suppressed) + await deploys.PushAsync(2, new DateTimeOffset(2026, 1, 2, 0, 0, 0, TimeSpan.Zero)); // a real deploy + + await WaitUntilAsync(() => Volatile.Read(ref raised) == 1, "positive control: the deploy must raise rediscovery"); + await WaitUntilAsync(() => driver.ProbeWatcherForTests?.WatchedPlatforms.Count == 2, + "the rediscovery must re-seed the probe set"); + platforms.Calls.ShouldBe(2); + sub.Subscribed.ShouldBe(["Plat01.ScanState", "Eng01.ScanState"], ignoreOrder: true); + } + + private sealed class ParkingPlatformSource : IGalaxyPlatformSource + { + private int _calls; + public int Calls => Volatile.Read(ref _calls); + public int ParkCall { get; init; } + public TaskCompletionSource Entered { get; } = new(TaskCreationOptions.RunContinuationsAsynchronously); + private readonly TaskCompletionSource _release = new(TaskCreationOptions.RunContinuationsAsynchronously); + public void Release() => _release.TrySetResult(); + + public async Task> GetPlatformTagNamesAsync(CancellationToken cancellationToken) + { + var call = Interlocked.Increment(ref _calls); + if (call == ParkCall) + { + Entered.TrySetResult(); + await _release.Task.WaitAsync(cancellationToken); + return ["Plat01"]; // the pre-deploy view this pass fetched + } + return call == 1 ? [] : ["Plat01", "Eng02"]; + } + } + + /// + /// A deploy that lands while a re-seed pass is mid-fetch gets a pass of its own that starts + /// after it — otherwise the probe set would stay on the stale view until the next deploy. + /// + [Fact] + public async Task A_deploy_during_an_in_flight_reseed_gets_its_own_pass() + { + var sub = new ParkableSubscriber(); + var deploys = new ChannelDeployWatchSource(); + var platforms = new ParkingPlatformSource { ParkCall = 2 }; // call 1 = init seed, call 2 = first rediscovery + await using var driver = new GalaxyDriver("drv", Opts(watchDeploy: true), hierarchySource: null, + subscriber: sub, platformSource: platforms, deployWatchSource: deploys); + var raised = 0; + driver.OnRediscoveryNeeded += (_, _) => Interlocked.Increment(ref raised); + try + { + await driver.InitializeAsync("{}", CancellationToken.None); + await WaitUntilAsync(() => Volatile.Read(ref deploys.Opened) == 1, "the deploy watch must open"); + await deploys.PushAsync(1, new DateTimeOffset(2026, 1, 1, 0, 0, 0, TimeSpan.Zero)); // bootstrap + await deploys.PushAsync(2, new DateTimeOffset(2026, 1, 2, 0, 0, 0, TimeSpan.Zero)); + (await Task.WhenAny(platforms.Entered.Task, Task.Delay(WaitMs))).ShouldBe(platforms.Entered.Task, + "the first rediscovery pass must be parked mid-fetch"); + + await deploys.PushAsync(3, new DateTimeOffset(2026, 1, 3, 0, 0, 0, TimeSpan.Zero)); + await WaitUntilAsync(() => Volatile.Read(ref raised) == 2, "positive control: the second deploy raised rediscovery"); + } + finally + { + platforms.Release(); + } + + await WaitUntilAsync(() => platforms.Calls == 3, "the deploy that landed mid-pass must get its own re-seed"); + await WaitUntilAsync(() => driver.ProbeWatcherForTests?.WatchedPlatforms.Count == 2, + "the final probe set must reflect the latest deploy, not the parked pass's stale view"); + } + + /// The production path (owned runtime) queries the platform source at init, with no DiscoverAsync. + [Fact] + public async Task Production_path_queries_the_platform_source_at_init() + { + var platforms = new ScriptedPlatformSource(); + await using var driver = new GalaxyDriver("drv", Opts(), hierarchySource: null, platformSource: platforms) + { + SessionOpenOverrideForTests = _ => Task.CompletedTask, + }; + + await driver.InitializeAsync("{}", CancellationToken.None); + + platforms.Calls.ShouldBe(1, "the owned runtime must seed its probe set at init"); + driver.GetHealth().State.ShouldBe(DriverState.Healthy); + } + + // ───────────────────────────── IMPORTANT 2: no pump on a dying subscriber ───────────────────────────── + + /// + /// A SubscribeAsync that lands while the teardown is parked (after the stopping flag, before + /// the subscriber is released) is refused and leaves no pump; before the shutdown the same + /// call works (positive control). + /// + [Fact] + public async Task SubscribeAsync_during_a_teardown_is_refused_and_leaves_no_pump() + { + var sub = new ParkableSubscriber(); + 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 driver.SubscribeAsync(["Tank01.Level"], TimeSpan.Zero, CancellationToken.None); + driver.HasEventPumpForTests.ShouldBeTrue("positive control: subscribe works on a live runtime"); + + sub.ParkUnsubscribe = true; + var shutdown = driver.ShutdownAsync(CancellationToken.None); + var subscribeCallsBefore = sub.SubscribeCalls; + try + { + (await Task.WhenAny(sub.UnsubscribeEntered.Task, Task.Delay(WaitMs))).ShouldBe(sub.UnsubscribeEntered.Task, + "the teardown must be parked inside the probe watcher's unsubscribe"); + shutdown.IsCompleted.ShouldBeFalse(); + + // The pump is still live here (the teardown takes it after the probe watcher) — reusing it + // is exactly the race: the call must be refused, not served on the dying subscriber. + driver.HasEventPumpForTests.ShouldBeTrue("the interleaving under test: the old pump is not yet taken"); + await Should.ThrowAsync( + () => driver.SubscribeAsync(["Tank02.Level"], TimeSpan.Zero, CancellationToken.None)); + sub.SubscribeCalls.ShouldBe(subscribeCallsBefore, "a refused subscribe must not reach the gateway"); + } + finally + { + // Never leave the teardown parked: it holds the lifecycle gate the await-using dispose needs. + sub.ReleaseUnsubscribe(); + } + await shutdown.WaitAsync(TimeSpan.FromMilliseconds(WaitMs)); + driver.HasEventPumpForTests.ShouldBeFalse("no pump may survive the teardown"); + driver.TrackedSubscriptionCountForTests.ShouldBe(0); + + await driver.InitializeAsync("{}", CancellationToken.None); + driver.HasEventPumpForTests.ShouldBeFalse("the next runtime starts with no inherited pump"); + } + + // ───────────────────────────── IMPORTANT 3: bounded gateway close ───────────────────────────── + + /// A gateway CloseSession that never answers cannot park ShutdownAsync past the close deadline. + [Fact] + public async Task ShutdownAsync_is_bounded_when_the_gateway_close_never_answers() + { + await using var driver = new GalaxyDriver("drv", Opts(), hierarchySource: null, platformSource: new ScriptedPlatformSource()) + { + SessionOpenOverrideForTests = _ => Task.CompletedTask, + SessionCloseDeadline = TimeSpan.FromMilliseconds(300), + }; + await driver.InitializeAsync("{}", CancellationToken.None); + var closeEntered = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var closeToken = CancellationToken.None; + driver.OwnedMxSessionForTests.ShouldNotBeNull().CloseOverrideForTests = async ct => + { + closeToken = ct; + closeEntered.TrySetResult(); + await Task.Delay(Timeout.Infinite, ct); // a blackholed gateway + }; + + var sw = System.Diagnostics.Stopwatch.StartNew(); + await driver.ShutdownAsync(CancellationToken.None).WaitAsync(TimeSpan.FromMilliseconds(WaitMs)); + sw.Stop(); + + closeEntered.Task.IsCompleted.ShouldBeTrue("positive control: the close was attempted"); + closeToken.IsCancellationRequested.ShouldBeTrue("the close must have been cancelled by the deadline"); + sw.Elapsed.ShouldBeLessThan(TimeSpan.FromSeconds(5), "ShutdownAsync must return at the close deadline, not the SDK's 30 s"); + driver.OwnedMxSessionForTests.ShouldBeNull(); + driver.GetHealth().State.ShouldBe(DriverState.Unknown); + } + + // ───────────────────────────── MINOR: generation guard on a straddling re-subscribe ───────────────────────────── + + /// + /// A replay whose SubscribeBulk completes after a teardown must not Rebind into the cleared + /// registry; the same replay without a teardown rebinds (positive control). + /// + [Fact] + public async Task A_resubscribe_that_straddles_a_teardown_does_not_rebind() + { + // Positive control first. + { + var sub = new ParkableSubscriber(); + 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); + await driver.InvokeReplayForTestAsync(CancellationToken.None); + driver.TrackedSubscriptionCountForTests.ShouldBe(1, "a replay on a live runtime rebinds the subscription"); + sub.SubscribeCalls.ShouldBe(2); + } + + { + 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); + + var replay = driver.InvokeReplayForTestAsync(CancellationToken.None); + (await Task.WhenAny(sub.SubscribeEntered.Task, Task.Delay(WaitMs))).ShouldBe(sub.SubscribeEntered.Task, + "the replay must be parked inside its SubscribeBulk"); + + await driver.ShutdownAsync(CancellationToken.None); + driver.TrackedSubscriptionCountForTests.ShouldBe(0); + + sub.ReleaseSubscribe(); + await replay.WaitAsync(TimeSpan.FromMilliseconds(WaitMs)); + + driver.TrackedSubscriptionCountForTests.ShouldBe(0, "a Rebind after the teardown would resurrect dead handles"); + } + } +} 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 index 5a146656..10ef4ff3 100644 --- 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 @@ -125,6 +125,13 @@ public sealed class GalaxyDriverShutdownTests }); } + /// Canned platform set so a production-path init never dials the (fake) gateway for one. + private sealed class NoPlatforms : IGalaxyPlatformSource + { + public Task> GetPlatformTagNamesAsync(CancellationToken cancellationToken) + => Task.FromResult>([]); + } + private sealed class RecordingAlarmFeed : IGalaxyAlarmFeed { public int Starts; @@ -172,8 +179,9 @@ public sealed class GalaxyDriverShutdownTests await driver.ShutdownAsync(CancellationToken.None); - (await Task.WhenAny(sub.StreamCancelled.Task, Task.Delay(WaitMs))).ShouldBe(sub.StreamCancelled.Task, - "ShutdownAsync must cancel the StreamEvents stream"); + // No wait: the pump's Cancel runs the stream's registration synchronously and ShutdownAsync + // awaited the pump's dispose, so the signal is either already set or the stream was never cancelled. + sub.StreamCancelled.Task.IsCompleted.ShouldBeTrue("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); @@ -289,7 +297,8 @@ public sealed class GalaxyDriverShutdownTests { var opens = 0; var source = new CountingDeployWatchSource(); - await using var driver = new GalaxyDriver("drv", Opts(), hierarchySource: null, deployWatchSource: source) + await using var driver = new GalaxyDriver("drv", Opts(), hierarchySource: null, deployWatchSource: source, + platformSource: new NoPlatforms()) { SessionOpenOverrideForTests = _ => { opens++; return Task.CompletedTask; }, }; @@ -338,7 +347,8 @@ public sealed class GalaxyDriverShutdownTests 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) + await using var driver = new GalaxyDriver("drv", Opts(watchDeploy: false), hierarchySource: null, + platformSource: new NoPlatforms()) { SessionOpenOverrideForTests = _ => ++attempts == 1 ? throw new InvalidOperationException("gateway refused")