fix(galaxy): seed the probe set at init, refuse pumps mid-teardown, bound the close

Review fix round 1 for lane F: - The ScanState probe watcher is seeded at InitializeAsync from a platform-only DiscoverHierarchy query (IGalaxyPlatformSource / GatewayGalaxyPlatformSource over the owned repository client) and re-seeded on every deploy-watcher rediscovery. The runtime never calls DiscoverAsync on a live driver, so per-platform host status was unreachable in production. A failed fetch logs a Warning and never fails init; re-seeds coalesce so a deploy during an in-flight pass still gets a pass that starts after it. - EnsureEventPumpStarted refuses while the runtime is stopping, BEFORE reusing the live pump (that pump belongs to the dying subscriber), and the production rebuild disposes any stale pump before wiring the new subscriber. - TeardownRuntimeAsync closes the MX session under the caller's token linked with a 5 s SessionCloseDeadline, so a blackholed gateway cannot hold DriverInstanceActor.PostStop for 30 s. - A runtime generation counter, bumped by every teardown, is re-checked before each Rebind in ResubscribeAllAsync, so a replay-gap re-snapshot that straddles a teardown cannot resurrect handles into the cleared registry. Refs #735 Claude-Session: https://claude.ai/code/session_01Ub9hGmYLQvn9XETzw7CMsT
2026-10-22 13:29:11 -04:00
parent 05a0e51fc1
commit 3456756cd1
6 changed files with 754 additions and 11 deletions
@@ -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;
+
+/// <summary>
+///     Default <see cref="IGalaxyPlatformSource"/>: a platform-only <c>DiscoverHierarchy</c> sweep
+///     over the gateway's <see cref="GalaxyRepositoryClient"/> — the request filters on
+///     <c>template_chain_contains</c> (<c>$WinPlatform</c>, <c>$AppEngine</c>) 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 <c>Repository.DiscoverPageSize</c>.
+/// </summary>
+public sealed class GatewayGalaxyPlatformSource : IGalaxyPlatformSource
+{
+    /// <summary>Template-chain tokens that identify a platform / engine object.</summary>
+    public static readonly IReadOnlyList<string> PlatformTemplates = ["$WinPlatform", "$AppEngine"];
+
+    private readonly GalaxyRepositoryClient _client;
+    private readonly int _pageSize;
+
+    /// <summary>Initializes a new instance of the <see cref="GatewayGalaxyPlatformSource"/> class.</summary>
+    /// <param name="client">The gateway's Galaxy repository client.</param>
+    /// <param name="pageSize">Objects per page (<c>Repository.DiscoverPageSize</c>); non-positive → the client default.</param>
+    public GatewayGalaxyPlatformSource(GalaxyRepositoryClient client, int pageSize = GatewayGalaxyHierarchySource.DefaultPageSize)
+    {
+        _client = client ?? throw new ArgumentNullException(nameof(client));
+        _pageSize = GatewayGalaxyHierarchySource.EffectivePageSize(pageSize);
+    }
+
+    /// <inheritdoc />
+    public Task<IReadOnlyList<string>> GetPlatformTagNamesAsync(CancellationToken cancellationToken)
+        => QueryAsync(_client.DiscoverHierarchyRawAsync, _pageSize, cancellationToken);
+
+    /// <summary>
+    ///     The platform-only sweep behind a delegate so it is unit-testable without a channel. Every
+    ///     request carries the template filter and <c>include_attributes = false</c>; 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.
+    /// </summary>
+    /// <param name="discover">The raw <c>DiscoverHierarchy</c> RPC.</param>
+    /// <param name="pageSize">Objects per page.</param>
+    /// <param name="cancellationToken">Cancellation token for the walk.</param>
+    /// <returns>Distinct platform tag names in gateway order.</returns>
+    internal static async Task<IReadOnlyList<string>> QueryAsync(
+        Func<DiscoverHierarchyRequest, CancellationToken, Task<DiscoverHierarchyReply>> discover,
+        int pageSize,
+        CancellationToken cancellationToken)
+    {
+        ArgumentNullException.ThrowIfNull(discover);
+        var effective = GatewayGalaxyHierarchySource.EffectivePageSize(pageSize);
+
+        var names = new List<string>();
+        var seen = new HashSet<string>(StringComparer.OrdinalIgnoreCase);
+        var seenPageTokens = new HashSet<string>(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));
+}
@@ -0,0 +1,18 @@
+namespace ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Browse;
+
+/// <summary>
+///     Driver-side seam for the set of Galaxy platform / engine objects whose <c>ScanState</c>
+///     the <c>PerPlatformProbeWatcher</c> advises. Production wraps the gateway's
+///     <c>DiscoverHierarchy</c> RPC filtered to <c>$WinPlatform</c> / <c>$AppEngine</c> with no
+///     attributes (<see cref="GatewayGalaxyPlatformSource"/>); tests substitute a canned list.
+///     The driver seeds the watcher from this at <c>InitializeAsync</c> and re-seeds on every
+///     deploy-watcher rediscovery — the runtime never calls <c>DiscoverAsync</c> on a live driver,
+///     so without this seed the probe watcher never had a platform set in production.
+/// </summary>
+public interface IGalaxyPlatformSource
+{
+    /// <summary>The <c>tag_name</c> of every deployed <c>$WinPlatform</c> / <c>$AppEngine</c> object.</summary>
+    /// <param name="cancellationToken">Cancellation token for the query.</param>
+    /// <returns>Platform tag names; empty when the galaxy has none.</returns>
+    Task<IReadOnlyList<string>> GetPlatformTagNamesAsync(CancellationToken cancellationToken);
+}
@@ -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;
+
+    /// <summary>
+    ///     Upper bound on the gateway <c>CloseSession</c> round-trip during a teardown. The SDK's
+    ///     default close runs under the 30 s call timeout, which against a blackholed gateway parks
+    ///     <c>ShutdownAsync</c> — and the host's <c>PostStop</c> — for the whole of it. Internal so
+    ///     tests can shrink it.
+    /// </summary>
+    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, <see cref="InitializeAsync"/> starts the
    ///     <see cref="DeployWatcher"/> over it (production builds one over the owned repository client).
    /// </param>
+    /// <param name="platformSource">
+    ///     Optional platform-set seam for the ScanState probe watcher. When supplied,
+    ///     <see cref="InitializeAsync"/> seeds the watcher from it and every rediscovery re-seeds
+    ///     (production builds a <see cref="GatewayGalaxyPlatformSource"/> over the owned repository client).
+    /// </param>
    internal GalaxyDriver(
        string driverInstanceId,
        GalaxyDriverOptions options,
@@ -269,9 +299,11 @@ public sealed class GalaxyDriver
        IGalaxyAlarmFeed? alarmFeed = null,
        ILogger<GalaxyDriver>? 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
        }
    }

+    /// <summary>
+    ///     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
+    ///     <c>InitializeAsync</c> (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).
+    /// </summary>
+    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<IGalaxyPlatformSource?> 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);
+    }
+
+    /// <summary>
+    ///     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.
+    /// </summary>
+    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).
    /// </summary>
+    /// <param name="entries">The subscriptions to re-issue.</param>
+    /// <param name="generation">
+    ///     The runtime generation this re-subscribe belongs to. Re-checked before every
+    ///     <see cref="SubscriptionRegistry.Rebind"/>: a teardown bumps it and clears the registry,
+    ///     and a Rebind landing after that would resurrect dead handles into the next runtime.
+    /// </param>
+    /// <param name="cancellationToken">Cancellation token.</param>
    /// <returns>The number of tag references re-subscribed.</returns>
    private async Task<int> ResubscribeAllAsync(
        IReadOnlyList<(long SubscriptionId, IReadOnlyList<TagBinding> 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();
+    }

    /// <inheritdoc />
    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);