Merge lane F: Galaxy driver teardown, probing, knobs (#735)

Claude-Session: https://claude.ai/code/session_01Ub9hGmYLQvn9XETzw7CMsT

src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Browse/GalaxyDiscoverer.cs

@@ -61,8 +61,12 @@ public sealed class GalaxyDiscoverer
     /// </summary>
     /// <param name="builder">The address space builder to populate with discovery results.</param>
     /// <param name="cancellationToken">The cancellation token for the operation.</param>
-    /// <returns>A task that represents the asynchronous operation.</returns>
-    public async Task DiscoverAsync(IAddressSpaceBuilder builder, CancellationToken cancellationToken)
+    /// <returns>
+    ///     The hierarchy this pass fetched, so a caller that needs the object set for something
+    ///     else (the driver selects its <c>$WinPlatform</c> / <c>$AppEngine</c> probe targets from
+    ///     it) reuses the one sweep instead of asking the gateway for the full galaxy twice.
+    /// </returns>
+    public async Task<IReadOnlyList<GalaxyObject>> DiscoverAsync(IAddressSpaceBuilder builder, CancellationToken cancellationToken)
     {
         ArgumentNullException.ThrowIfNull(builder);
         var objects = await _source.GetHierarchyAsync(cancellationToken).ConfigureAwait(false);
@@ -132,6 +136,8 @@ public sealed class GalaxyDiscoverer
                 }
             }
         }
+
+        return objects;
     }
 
     // Resolve (and memoise) the folder for a gobject, recursively creating its parent's folder

src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Browse/GatewayGalaxyHierarchySource.cs

@@ -5,22 +5,83 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Browse;
 
 /// <summary>
 ///     Default <see cref="IGalaxyHierarchySource"/> wrapping the gateway's
-///     <see cref="GalaxyRepositoryClient"/>. Pages internally via the client's overload.
+///     <see cref="GalaxyRepositoryClient"/>. Pages the <c>DiscoverHierarchy</c> RPC itself,
+///     at the configured <c>Repository.DiscoverPageSize</c>: the client's typed
+///     <c>DiscoverHierarchyAsync</c> overloads hard-code a 5000-row page
+///     (<c>DiscoverHierarchyOptions</c> carries no page size), so the knob the AdminUI
+///     authors and the factory parses used to reach nothing. The raw RPC takes the page
+///     size on the wire (<c>DiscoverHierarchyRequest.page_size</c>), so it is honoured here.
 /// </summary>
 public sealed class GatewayGalaxyHierarchySource : IGalaxyHierarchySource
 {
+    /// <summary>The page size used when the configured one is not positive — the client library's own default.</summary>
+    public const int DefaultPageSize = 5000;
+
     private readonly GalaxyRepositoryClient _client;
+    private readonly int _pageSize;
 
     /// <summary>
     ///     Initializes a new instance of the <see cref="GatewayGalaxyHierarchySource"/> class.
     /// </summary>
     /// <param name="client">The gateway's Galaxy repository client for discovering the object hierarchy.</param>
-    public GatewayGalaxyHierarchySource(GalaxyRepositoryClient client)
+    /// <param name="pageSize">
+    ///     Objects per <c>DiscoverHierarchy</c> page (<c>Repository.DiscoverPageSize</c>). A value that is
+    ///     not positive falls back to <see cref="DefaultPageSize"/>.
+    /// </param>
+    public GatewayGalaxyHierarchySource(GalaxyRepositoryClient client, int pageSize = DefaultPageSize)
     {
         _client = client ?? throw new ArgumentNullException(nameof(client));
+        _pageSize = EffectivePageSize(pageSize);
     }
 
+    /// <summary>The page size this source sends on every <c>DiscoverHierarchy</c> request.</summary>
+    public int PageSize => _pageSize;
+
+    /// <summary>Clamp a configured page size to something the gateway accepts: non-positive → <see cref="DefaultPageSize"/>.</summary>
+    /// <param name="configured">The configured value.</param>
+    /// <returns>The page size to send.</returns>
+    public static int EffectivePageSize(int configured) => configured > 0 ? configured : DefaultPageSize;
+
     /// <inheritdoc />
     public Task<IReadOnlyList<GalaxyObject>> GetHierarchyAsync(CancellationToken cancellationToken)
-        => _client.DiscoverHierarchyAsync(cancellationToken);
+        => PageAsync(_client.DiscoverHierarchyRawAsync, _pageSize, cancellationToken);
+
+    /// <summary>
+    ///     Walk every <c>DiscoverHierarchy</c> page at <paramref name="pageSize"/> and concatenate the
+    ///     objects. Mirrors the client library's own loop (including its repeated-page-token guard,
+    ///     which turns a misbehaving gateway into an error rather than an endless walk), with the
+    ///     RPC behind a delegate so the loop is unit-testable without a channel.
+    /// </summary>
+    /// <param name="discover">The raw <c>DiscoverHierarchy</c> RPC.</param>
+    /// <param name="pageSize">Objects per page; sent on every request.</param>
+    /// <param name="cancellationToken">Cancellation token for the walk.</param>
+    /// <returns>Every object across all pages, in gateway order.</returns>
+    internal static async Task<IReadOnlyList<GalaxyObject>> PageAsync(
+        Func<DiscoverHierarchyRequest, CancellationToken, Task<DiscoverHierarchyReply>> discover,
+        int pageSize,
+        CancellationToken cancellationToken)
+    {
+        ArgumentNullException.ThrowIfNull(discover);
+        var effective = EffectivePageSize(pageSize);
+
+        var objects = new List<GalaxyObject>();
+        var seenPageTokens = new HashSet<string>(StringComparer.Ordinal);
+        var pageToken = string.Empty;
+        do
+        {
+            var request = new DiscoverHierarchyRequest { PageSize = effective, PageToken = pageToken };
+            var reply = await discover(request, cancellationToken).ConfigureAwait(false);
+
+            objects.AddRange(reply.Objects);
+            pageToken = reply.NextPageToken ?? string.Empty;
+            if (!string.IsNullOrWhiteSpace(pageToken) && !seenPageTokens.Add(pageToken))
+            {
+                throw new InvalidOperationException(
+                    $"Galaxy DiscoverHierarchy returned a repeated page token '{pageToken}' — refusing to loop.");
+            }
+        }
+        while (!string.IsNullOrWhiteSpace(pageToken));
+
+        return objects;
+    }
 }

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;
+
+/// <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));
+}

src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Browse/IGalaxyPlatformSource.cs

@@ -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);
+}

src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/GalaxyDriverProbe.cs

@@ -54,12 +54,18 @@ public sealed class GalaxyDriverProbe : IDriverProbe
         if (string.IsNullOrWhiteSpace(host) || port <= 0)
             return new(false, "Config has no host/port to probe.", null);
 
+        // The config's own ProbeTimeoutSeconds caps the probe deadline. It can only tighten the
+        // caller's budget, never extend it — the caller's token fires at `timeout` regardless.
+        timeout = EffectiveTimeout(timeout, opts.ProbeTimeoutSeconds);
+        using var probeCts = CancellationTokenSource.CreateLinkedTokenSource(ct);
+        probeCts.CancelAfter(timeout);
+
         // --- TCP preflight: fast-fail for closed ports / unreachable hosts ---
         var sw = Stopwatch.StartNew();
         try
         {
             using var socket = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
-            await socket.ConnectAsync(host, port, ct);
+            await socket.ConnectAsync(host, port, probeCts.Token);
         }
         catch (SocketException ex)
         {
@@ -86,10 +92,7 @@ public sealed class GalaxyDriverProbe : IDriverProbe
         {
             client = GalaxyRepositoryClient.Create(BuildProbeClientOptions(opts.Gateway, timeout));
 
-            using var deadlineCts = CancellationTokenSource.CreateLinkedTokenSource(ct);
-            deadlineCts.CancelAfter(timeout);
-
-            await client.TestConnectionAsync(deadlineCts.Token).ConfigureAwait(false);
+            await client.TestConnectionAsync(probeCts.Token).ConfigureAwait(false);
             sw.Stop();
 
             // No RpcException ⇒ the gateway answered the gRPC call successfully.
@@ -157,6 +160,26 @@ public sealed class GalaxyDriverProbe : IDriverProbe
         _ => (false, $"Reachable at {host}:{port} but gateway gRPC handshake failed: {code}"),
     };
 
+    /// <summary>
+    ///     The deadline the probe actually runs under: the caller's budget, tightened by the
+    ///     config's <see cref="GalaxyDriverOptions.ProbeTimeoutSeconds"/> when that is positive and
+    ///     shorter. The knob was authored by the AdminUI and parsed by the factory but consumed by
+    ///     nothing — the probe always ran on the caller's budget alone. A non-positive caller budget
+    ///     is floored to 1 s before the comparison (the same floor <see cref="BuildProbeClientOptions"/>
+    ///     applies to the gRPC call budget), so the probe runs briefly rather than cancelling before
+    ///     it starts; a longer config value never raises it above that floor.
+    /// </summary>
+    /// <param name="callerTimeout">The budget the host handed the probe.</param>
+    /// <param name="configuredSeconds">The config's <c>ProbeTimeoutSeconds</c>; ignored when not positive.</param>
+    /// <returns>The effective probe deadline.</returns>
+    internal static TimeSpan EffectiveTimeout(TimeSpan callerTimeout, int configuredSeconds)
+    {
+        var caller = callerTimeout > TimeSpan.Zero ? callerTimeout : TimeSpan.FromSeconds(1);
+        if (configuredSeconds <= 0) return caller;
+        var configured = TimeSpan.FromSeconds(configuredSeconds);
+        return configured < caller ? configured : caller;
+    }
+
     /// <summary>
     ///     Builds the gRPC client options for the probe ping from the gateway config WITHOUT
     ///     resolving the API-key secret ref (the host owns secret resolution; the probe sends the

src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Health/PerPlatformProbeWatcher.cs

@@ -38,12 +38,19 @@ public sealed class PerPlatformProbeWatcher : IDisposable
     private readonly ILogger _logger;
     private readonly int _bufferedUpdateIntervalMs;
 
+    // The driver's fan-out registry. The EventPump resolves every inbound event through it, so
+    // the probe handles must be registered there (under SubscriptionRegistry.ProbeSubscriptionId)
+    // or every ScanState event is dropped as "stale after unsubscribe". Null when the watcher is
+    // driven directly (unit tests of the decoding table).
+    private readonly SubscriptionRegistry? _registry;
+
     // Tracked platform → gw item handle. Item handle 0 means the gw rejected the subscribe;
     // we keep the entry so SyncPlatformsAsync doesn't try to subscribe it again on every call.
     private readonly ConcurrentDictionary<string, int> _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;
 
     /// <summary>Initializes a new instance of the PerPlatformProbeWatcher class.</summary>
     /// <param name="subscriber">The Galaxy subscriber for managing probe subscriptions.</param>
@@ -55,9 +62,30 @@ public sealed class PerPlatformProbeWatcher : IDisposable
         HostStatusAggregator aggregator,
         ILogger? logger = null,
         int bufferedUpdateIntervalMs = 0)
+        : this(subscriber, aggregator, registry: null, logger, bufferedUpdateIntervalMs)
+    {
+    }
+
+    /// <summary>
+    ///     Driver-side ctor: the watcher publishes its probe handles into <paramref name="registry"/>
+    ///     under <see cref="SubscriptionRegistry.ProbeSubscriptionId"/> so the <see cref="EventPump"/>
+    ///     fans ScanState events out to <see cref="OnProbeValueChanged"/>.
+    /// </summary>
+    /// <param name="subscriber">The Galaxy subscriber for managing probe subscriptions.</param>
+    /// <param name="aggregator">The host status aggregator for tracking platform connectivity.</param>
+    /// <param name="registry">The driver's subscription registry; null to keep the handles private.</param>
+    /// <param name="logger">Optional logger for diagnostic messages.</param>
+    /// <param name="bufferedUpdateIntervalMs">Buffered update interval in milliseconds; must be >= 0.</param>
+    internal PerPlatformProbeWatcher(
+        IGalaxySubscriber subscriber,
+        HostStatusAggregator aggregator,
+        SubscriptionRegistry? registry,
+        ILogger? logger = null,
+        int bufferedUpdateIntervalMs = 0)
     {
         _subscriber = subscriber ?? throw new ArgumentNullException(nameof(subscriber));
         _aggregator = aggregator ?? throw new ArgumentNullException(nameof(aggregator));
+        _registry = registry;
         _logger = logger ?? NullLogger.Instance;
         if (bufferedUpdateIntervalMs < 0)
         {
@@ -67,6 +95,103 @@ public sealed class PerPlatformProbeWatcher : IDisposable
         _bufferedUpdateIntervalMs = bufferedUpdateIntervalMs;
     }
 
+    /// <summary>Current platform → probe-handle bindings, as the registry sees them. Test seam.</summary>
+    internal IReadOnlyList<TagBinding> CurrentBindings => BuildBindings();
+
+    private IReadOnlyList<TagBinding> BuildBindings() =>
+        [.. _itemHandlesByPlatform.Select(kvp => new TagBinding(kvp.Key + ProbeSuffix, kvp.Value))];
+
+    /// <summary>
+    ///     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).
+    /// </summary>
+    private void PublishBindings()
+    {
+        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);
+    }
+
+    /// <summary>
+    ///     Re-issue SubscribeBulk for every tracked platform and replace the handles — the
+    ///     reconnect replay path, where the gateway session was recreated and every pre-reconnect
+    ///     handle is dead. Platforms the gateway now rejects keep a 0 handle (no events expected).
+    /// </summary>
+    /// <param name="cancellationToken">A cancellation token to cancel the operation.</param>
+    /// <returns>The number of probe references re-subscribed.</returns>
+    public async Task<int> ResubscribeAsync(CancellationToken cancellationToken)
+    {
+        ObjectDisposedException.ThrowIf(_disposed, this);
+        var platforms = _itemHandlesByPlatform.Keys.ToArray();
+        if (platforms.Length == 0) return 0;
+
+        await SubscribeAndRecordAsync(platforms, cancellationToken).ConfigureAwait(false);
+        return platforms.Length;
+    }
+
+    private async Task SubscribeAndRecordAsync(IReadOnlyList<string> platforms, CancellationToken cancellationToken)
+    {
+        var probeAddresses = platforms.Select(p => p + ProbeSuffix).ToArray();
+        // PR 6.3 — use the configured bufferedUpdateIntervalMs (defaults to 0 = gw cadence
+        // when the driver hasn't overridden MxAccess.PublishingIntervalMs). Probe ScanState
+        // changes are rare so a coarser interval is usually fine; deployments that need
+        // tighter health visibility can dial it down through GalaxyDriverOptions.
+        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 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 ?? "<no result returned>");
+        }
+
+        // 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)
+            {
+                foreach (var (platform, itemHandle, _) in acquired) _itemHandlesByPlatform[platform] = itemHandle;
+                PublishBindingsLocked();
+            }
+        }
+
+        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);
+            }
+        }
+    }
+
     /// <summary>Snapshot of platform tag names currently watched.</summary>
     public IReadOnlyCollection<string> WatchedPlatforms => [.. _itemHandlesByPlatform.Keys];
 
@@ -109,6 +234,8 @@ public sealed class PerPlatformProbeWatcher : IDisposable
 
         if (toRemove.Count > 0)
         {
+            // Drop the dead handles from the fan-out map before the (slow, fallible) unsubscribe.
+            PublishBindings();
             try
             {
                 await _subscriber.UnsubscribeBulkAsync(
@@ -124,30 +251,7 @@ public sealed class PerPlatformProbeWatcher : IDisposable
 
         if (toAdd.Count == 0) return;
 
-        var probeAddresses = toAdd.Select(p => p + ProbeSuffix).ToArray();
-        // PR 6.3 — use the configured bufferedUpdateIntervalMs (defaults to 0 = gw cadence
-        // when the driver hasn't overridden MxAccess.PublishingIntervalMs). Probe ScanState
-        // changes are rare so a coarser interval is usually fine; deployments that need
-        // tighter health visibility can dial it down through GalaxyDriverOptions.
-        var results = await _subscriber.SubscribeBulkAsync(
-            probeAddresses, _bufferedUpdateIntervalMs, cancellationToken).ConfigureAwait(false);
-
-        for (var i = 0; i < toAdd.Count; i++)
-        {
-            var platform = toAdd[i];
-            var match = results.FirstOrDefault(r => string.Equals(
-                r.TagAddress, probeAddresses[i], StringComparison.OrdinalIgnoreCase));
-
-            var itemHandle = match is { WasSuccessful: true } ? match.ItemHandle : 0;
-            _itemHandlesByPlatform[platform] = itemHandle;
-
-            if (itemHandle <= 0)
-            {
-                _logger.LogWarning(
-                    "PerPlatformProbeWatcher subscribe failed for {Platform}: {Error}",
-                    platform, match?.ErrorMessage ?? "<no result returned>");
-            }
-        }
+        await SubscribeAndRecordAsync(toAdd, cancellationToken).ConfigureAwait(false);
     }
 
     /// <summary>
@@ -196,20 +300,32 @@ public sealed class PerPlatformProbeWatcher : IDisposable
         };
     }
 
-    /// <summary>Disposes the probe watcher and unsubscribes all tracked platforms.</summary>
-    public void Dispose()
+    /// <summary>
+    ///     Stops the watcher and unsubscribes every tracked platform's probe handle on the gateway
+    ///     (best-effort — a transport error is logged and swallowed). The async counterpart of
+    ///     <see cref="Dispose"/>; the driver's <c>ShutdownAsync</c> uses it so the unsubscribe
+    ///     round-trip is awaited rather than blocked on. Idempotent.
+    /// </summary>
+    /// <param name="cancellationToken">Bounds the unsubscribe round-trip.</param>
+    /// <returns>A task that completes once the handles have been released (or the attempt gave up).</returns>
+    public async Task StopAsync(CancellationToken cancellationToken)
     {
-        if (_disposed) return;
-        _disposed = true;
-
-        // Best-effort unsubscribe everything we know about. Run synchronously through
-        // GetAwaiter().GetResult() since Dispose is sync; transport errors are swallowed.
-        var liveHandles = _itemHandlesByPlatform.Values.Where(h => h > 0).ToArray();
-        _itemHandlesByPlatform.Clear();
+        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 { _subscriber.UnsubscribeBulkAsync(liveHandles, CancellationToken.None).GetAwaiter().GetResult(); }
-            catch (Exception ex) { _logger.LogWarning(ex, "PerPlatformProbeWatcher dispose unsubscribe failed"); }
+            try { await _subscriber.UnsubscribeBulkAsync(liveHandles, cancellationToken).ConfigureAwait(false); }
+            catch (Exception ex) { _logger.LogWarning(ex, "PerPlatformProbeWatcher stop unsubscribe failed"); }
         }
     }
+
+    /// <summary>Disposes the probe watcher and unsubscribes all tracked platforms (blocking on the round-trip).</summary>
+    public void Dispose() => StopAsync(CancellationToken.None).GetAwaiter().GetResult();
 }

src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/EventPump.cs

@@ -187,9 +187,11 @@ internal sealed class EventPump : IAsyncDisposable
                 NotifyStreamFault(cause);
             }
         }
-        catch (OperationCanceledException) when (ct.IsCancellationRequested)
+        catch (Exception) when (ct.IsCancellationRequested)
         {
-            // Clean shutdown — no log.
+            // Clean shutdown — no log. A cancelled gRPC stream may surface as RpcException(Cancelled)
+            // rather than OperationCanceledException; either way the pump was told to stop, so this
+            // is not a transport fault and must not wake the reconnect supervisor mid-teardown.
         }
         catch (Exception ex)
         {
@@ -283,14 +285,19 @@ internal sealed class EventPump : IAsyncDisposable
                 EventsDispatched.Add(1, _clientTag);
             }
         }
-        catch (OperationCanceledException) when (ct.IsCancellationRequested)
+        catch (Exception) when (ct.IsCancellationRequested)
         {
             // Clean shutdown.
         }
         catch (Exception ex)
         {
+            // The read loop is still draining the stream into a channel nobody reads: the pump is
+            // as dead as after a transport fault, but used to raise nothing — health stayed
+            // Healthy and the comms-loss watchdog never tripped (a clean termination evading the
+            // watchdog). Report it as a stream fault so the supervisor recreates the pump.
             _logger.LogWarning(ex,
-                "Galaxy EventPump dispatch loop ended with an exception — events past this point will be lost until restart.");
+                "Galaxy EventPump dispatch loop ended with an exception — notifying reconnect supervisor so the pump is recreated.");
+            NotifyStreamFault(ex);
         }
     }
 

src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/GalaxyMxSession.cs

@@ -60,6 +60,12 @@ public sealed class GalaxyMxSession : IAsyncDisposable
     /// </summary>
     internal Func<CancellationToken, Task>? OpenAndRegisterOverrideForTests { get; set; }
 
+    /// <summary>
+    ///     Test seam — replaces the gateway <c>CloseSession</c> round-trip in <see cref="CloseAsync"/>
+    ///     so the bounded-close behaviour can be exercised against a never-answering fake.
+    /// </summary>
+    internal Func<CancellationToken, Task>? CloseOverrideForTests { get; set; }
+
     /// <summary>
     ///     Connect the underlying gateway client + open an MXAccess session + register the
     ///     configured client name. Idempotent — second calls are no-ops while a connect has
@@ -96,7 +102,7 @@ public sealed class GalaxyMxSession : IAsyncDisposable
         {
             // A partial open (e.g. OpenSession succeeded but Register threw) must not leave a half-open
             // handle that blocks the next reconnect. Tear the partial state down so a retry rebuilds cleanly.
-            await TeardownSessionAsync().ConfigureAwait(false);
+            await TeardownSessionAsync(CancellationToken.None).ConfigureAwait(false);
             throw;
         }
     }
@@ -112,10 +118,23 @@ public sealed class GalaxyMxSession : IAsyncDisposable
     public async Task RecreateAsync(MxGatewayClientOptions clientOptions, CancellationToken cancellationToken)
     {
         ObjectDisposedException.ThrowIf(_disposed, this);
-        await TeardownSessionAsync().ConfigureAwait(false);
+        await TeardownSessionAsync(cancellationToken).ConfigureAwait(false);
         await ConnectAsync(clientOptions, cancellationToken).ConfigureAwait(false);
     }
 
+    /// <summary>
+    ///     Close the gateway session and release the owned client under
+    ///     <paramref name="cancellationToken"/>. The SDK's <c>DisposeAsync</c> closes with NO token,
+    ///     i.e. under the 30 s <c>DefaultCallTimeout</c> — against a blackholed gateway that parks
+    ///     the driver's <c>ShutdownAsync</c> (and the host's <c>PostStop</c>) for the full timeout.
+    ///     A close that is cancelled or fails is NOT retried through the SDK dispose (which would
+    ///     park again); the channel is dropped and the gateway reaps the session on its idle lease.
+    ///     The session can be re-opened afterwards (this does not set <c>_disposed</c>).
+    /// </summary>
+    /// <param name="cancellationToken">Bounds the gateway <c>CloseSession</c> round-trip.</param>
+    /// <returns>A task that completes once the session and client are released.</returns>
+    public Task CloseAsync(CancellationToken cancellationToken) => TeardownSessionAsync(cancellationToken);
+
     /// <summary>
     ///     Test seam — attach a session opened externally (e.g. against an in-process gw
     ///     fake). Skips the gateway-client construction so tests can drive the session
@@ -143,24 +162,45 @@ public sealed class GalaxyMxSession : IAsyncDisposable
     {
         if (_disposed) return;
         _disposed = true;
-        await TeardownSessionAsync().ConfigureAwait(false);
+        await TeardownSessionAsync(CancellationToken.None).ConfigureAwait(false);
     }
 
     /// <summary>
-    ///     Resets the connect flag and best-effort disposes + nulls the owned session and client.
-    ///     Shared by <see cref="DisposeAsync"/> (final teardown) and <see cref="RecreateAsync"/> /
-    ///     the partial-open recovery path (rebuild teardown) — it does NOT set <c>_disposed</c>, so
-    ///     a torn-down session can be re-opened.
+    ///     Resets the connect flag and best-effort closes + nulls the owned session and client.
+    ///     Shared by <see cref="DisposeAsync"/> (final teardown), <see cref="CloseAsync"/> (bounded
+    ///     teardown) and <see cref="RecreateAsync"/> / the partial-open recovery path (rebuild
+    ///     teardown) — it does NOT set <c>_disposed</c>, so a torn-down session can be re-opened.
     /// </summary>
-    private async Task TeardownSessionAsync()
+    private async Task TeardownSessionAsync(CancellationToken cancellationToken)
     {
+        var wasConnected = _connected;
         _connected = false;
 
+        // The override models the gateway round-trip, so it runs only when there was a live
+        // session to close — a second teardown (CloseAsync then DisposeAsync) must not close twice.
+        var closeOverride = CloseOverrideForTests;
+        if (closeOverride is not null && wasConnected)
+        {
+            try { await closeOverride(cancellationToken).ConfigureAwait(false); }
+            catch (Exception ex) { _logger.LogWarning(ex, "GalaxyMxSession close failed or timed out (best-effort)"); }
+        }
+
         if (_session is not null)
         {
-            try { await _session.DisposeAsync().ConfigureAwait(false); }
-            catch (Exception ex) { _logger.LogWarning(ex, "GalaxyMxSession session dispose failed (best-effort)"); }
+            var session = _session;
             _session = null;
+            try
+            {
+                // CloseAsync(ct) is the bounded gateway round-trip; DisposeAsync only re-runs an
+                // un-replied close (with no token), so it is reached only after a successful one.
+                await session.CloseAsync(cancellationToken).ConfigureAwait(false);
+                await session.DisposeAsync().ConfigureAwait(false);
+            }
+            catch (Exception ex)
+            {
+                _logger.LogWarning(ex,
+                    "GalaxyMxSession close failed or timed out (best-effort) — dropping the channel; the gateway reaps the session on its idle lease.");
+            }
         }
 
         if (_ownedClient is not null)

src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/GatewayGalaxyDataWriter.cs

@@ -120,15 +120,87 @@ public sealed class GatewayGalaxyDataWriter : IGalaxyDataWriter
         // Bind the per-batch MX write operations behind the internal IMxWriteOps seam (see below) so the
         // whole write pipeline is unit-testable without the sealed SDK session (archreview 06/S-1).
         var ops = new SessionMxWriteOps(session, serverHandle);
+        return await WriteBatchAsync(ops, writes, securityResolver, cancellationToken).ConfigureAwait(false);
+    }
 
+    /// <summary>
+    ///     Upper bound on write lanes in flight at once. A first-touch write is three round-trips
+    ///     (AddItem → AdviseSupervisory → Write) and the batch used to run them strictly serially;
+    ///     distinct tags are independent, so their lanes now overlap, bounded so a large batch cannot
+    ///     flood the gateway's STA pump. Acks stay serial by design (see <c>GalaxyDriver.AcknowledgeAsync</c>).
+    /// </summary>
+    internal const int MaxConcurrentWriteLanes = 8;
+
+    /// <summary>Test seam: drive the real batch pipeline against a fake <see cref="IMxWriteOps"/>.</summary>
+    /// <param name="ops">The (fake) MX write operations.</param>
+    /// <param name="writes">The batch.</param>
+    /// <param name="securityResolver">Per-tag classification resolver.</param>
+    /// <param name="ct">The cancellation token.</param>
+    /// <returns>One result per request, in input order.</returns>
+    internal Task<IReadOnlyList<WriteResult>> WriteBatchForTestAsync(
+        IMxWriteOps ops, IReadOnlyList<WriteRequest> writes,
+        Func<string, SecurityClassification> securityResolver, CancellationToken ct)
+        => WriteBatchAsync(ops, writes, securityResolver, ct);
+
+    /// <summary>
+    ///     Run a batch as one lane per distinct tag (OrdinalIgnoreCase): writes to the SAME tag
+    ///     stay in input order inside their lane, lanes run concurrently up to
+    ///     <see cref="MaxConcurrentWriteLanes"/>, and results land at their request's index.
+    ///     A single-tag batch takes the direct path with no gate.
+    /// </summary>
+    private async Task<IReadOnlyList<WriteResult>> WriteBatchAsync(
+        IMxWriteOps ops, IReadOnlyList<WriteRequest> writes,
+        Func<string, SecurityClassification> securityResolver, CancellationToken ct)
+    {
         var results = new WriteResult[writes.Count];
+        if (writes.Count == 0) return results;
+
+        // Group request indices by tag, first-seen order preserved so the lane start order is
+        // deterministic and a same-tag sequence keeps its authored order.
+        var lanes = new Dictionary<string, List<int>>(StringComparer.OrdinalIgnoreCase);
+        var laneOrder = new List<List<int>>();
         for (var i = 0; i < writes.Count; i++)
         {
-            results[i] = await WriteOneAsync(ops, writes[i],
-                securityResolver(writes[i].FullReference), cancellationToken)
-                .ConfigureAwait(false);
+            if (!lanes.TryGetValue(writes[i].FullReference, out var lane))
+            {
+                lane = [];
+                lanes[writes[i].FullReference] = lane;
+                laneOrder.Add(lane);
+            }
+            lane.Add(i);
         }
+
+        if (laneOrder.Count == 1)
+        {
+            await RunLaneAsync(laneOrder[0]).ConfigureAwait(false);
+            return results;
+        }
+
+        using var gate = new SemaphoreSlim(MaxConcurrentWriteLanes, MaxConcurrentWriteLanes);
+        var tasks = new Task[laneOrder.Count];
+        for (var l = 0; l < laneOrder.Count; l++)
+        {
+            var lane = laneOrder[l];
+            tasks[l] = RunGatedAsync(lane);
+        }
+        await Task.WhenAll(tasks).ConfigureAwait(false);
         return results;
+
+        async Task RunGatedAsync(List<int> lane)
+        {
+            await gate.WaitAsync(ct).ConfigureAwait(false);
+            try { await RunLaneAsync(lane).ConfigureAwait(false); }
+            finally { gate.Release(); }
+        }
+
+        async Task RunLaneAsync(List<int> lane)
+        {
+            foreach (var i in lane)
+            {
+                results[i] = await WriteOneAsync(ops, writes[i],
+                    securityResolver(writes[i].FullReference), ct).ConfigureAwait(false);
+            }
+        }
     }
 
     /// <summary>Test seam: drive the real single-write pipeline against a fake <see cref="IMxWriteOps"/>

src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/ReconnectSupervisor.cs

@@ -62,6 +62,16 @@ public sealed class ReconnectSupervisor : IDisposable
     private Task? _recoveryLoop;
     private CancellationTokenSource? _loopCts;
     private bool _disposed;
+    private int _cyclesStarted;
+    private int _cyclesReleased;
+
+    /// <summary>Recovery cycles started so far (one per <see cref="ReportTransportFailure"/> that left Healthy). Test seam.</summary>
+    internal int CyclesStarted => Volatile.Read(ref _cyclesStarted);
+
+    /// <summary>Recovery cycles whose <see cref="CancellationTokenSource"/> has been disposed. Equals
+    /// <see cref="CyclesStarted"/> once every loop has exited — a cycle's CTS used to be released only
+    /// if it happened to be the LAST one when <see cref="Dispose"/> ran. Test seam.</summary>
+    internal int CyclesReleased => Volatile.Read(ref _cyclesReleased);
 
     /// <summary>Fires after every state transition.</summary>
     public event EventHandler<StateTransition>? StateChanged;
@@ -136,8 +146,13 @@ public sealed class ReconnectSupervisor : IDisposable
 
             TransitionLocked(State.TransportLost, cause.Message);
 
-            _loopCts = new CancellationTokenSource();
-            _recoveryLoop = Task.Run(() => RecoveryLoopAsync(_loopCts.Token));
+            // Each cycle owns its CTS and releases it when its loop exits (see RecoveryLoopAsync's
+            // finally). Before, every cycle allocated a fresh CTS and only the last one was ever
+            // disposed — one leaked CancellationTokenSource per outage for the driver's lifetime.
+            var cts = new CancellationTokenSource();
+            _loopCts = cts;
+            Interlocked.Increment(ref _cyclesStarted);
+            _recoveryLoop = Task.Run(() => RecoveryLoopAsync(cts));
         }
     }
 
@@ -156,7 +171,26 @@ public sealed class ReconnectSupervisor : IDisposable
         }
     }
 
-    private async Task RecoveryLoopAsync(CancellationToken ct)
+    private async Task RecoveryLoopAsync(CancellationTokenSource cycleCts)
+    {
+        try
+        {
+            await RecoveryLoopCoreAsync(cycleCts.Token).ConfigureAwait(false);
+        }
+        finally
+        {
+            // Release this cycle's CTS once nothing can observe its token any more. Stop/Dispose may
+            // race us here: they Cancel (guarded against ODE) and Dispose (idempotent) the same object.
+            lock (_stateLock)
+            {
+                if (ReferenceEquals(_loopCts, cycleCts)) _loopCts = null;
+            }
+            cycleCts.Dispose();
+            Interlocked.Increment(ref _cyclesReleased);
+        }
+    }
+
+    private async Task RecoveryLoopCoreAsync(CancellationToken ct)
     {
         var attempt = 0;
         while (!ct.IsCancellationRequested)
@@ -240,8 +274,13 @@ public sealed class ReconnectSupervisor : IDisposable
         }
     }
 
-    /// <summary>Disposes the supervisor and cancels any in-flight recovery loop.</summary>
-    public void Dispose()
+    /// <summary>
+    ///     Cancels any in-flight recovery loop and waits for it to exit, without blocking the caller's
+    ///     thread. The async counterpart of <see cref="Dispose"/>; the driver's <c>ShutdownAsync</c>
+    ///     uses it so a loop parked inside a reopen round-trip is awaited, not blocked on. Idempotent.
+    /// </summary>
+    /// <returns>A task that completes once the recovery loop has exited.</returns>
+    public async Task StopAsync()
     {
         if (_disposed) return;
         _disposed = true;
@@ -250,13 +289,18 @@ public sealed class ReconnectSupervisor : IDisposable
         Task? loop;
         lock (_stateLock) { cts = _loopCts; loop = _recoveryLoop; _loopCts = null; _recoveryLoop = null; }
 
-        cts?.Cancel();
+        // The loop's own finally may have released the CTS already — a cancelled-then-disposed CTS
+        // throws on Cancel, and a disposed one is a no-op on Dispose.
+        try { cts?.Cancel(); } catch (ObjectDisposedException) { /* loop already released it */ }
         if (loop is not null)
         {
-            try { loop.GetAwaiter().GetResult(); } catch { /* shutdown */ }
+            try { await loop.ConfigureAwait(false); } catch { /* shutdown */ }
         }
         cts?.Dispose();
     }
+
+    /// <summary>Disposes the supervisor and cancels any in-flight recovery loop (blocking until it exits).</summary>
+    public void Dispose() => StopAsync().GetAwaiter().GetResult();
 }
 
 /// <summary>

src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/SubscriptionRegistry.cs

@@ -18,10 +18,39 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Runtime;
 /// </remarks>
 internal sealed class SubscriptionRegistry
 {
+    /// <summary>
+    ///     Reserved subscription id for the per-platform <c>ScanState</c> probe handles
+    ///     (<c>PerPlatformProbeWatcher</c>). Driver subscription ids are positive and
+    ///     monotonic (<see cref="NextSubscriptionId"/>), so a negative id can never collide.
+    ///     The probe handles used to live only in the watcher's own map, so the
+    ///     <see cref="EventPump"/> — which resolves subscribers through THIS registry —
+    ///     dropped every probe event as "stale after unsubscribe" and per-platform host
+    ///     status never updated.
+    /// </summary>
+    public const long ProbeSubscriptionId = -1;
+
     private readonly ConcurrentDictionary<long, SubscriptionEntry> _bySubscriptionId = new();
     private readonly ConcurrentDictionary<int, ImmutableHashSet<long>> _subscribersByItemHandle = new();
     private long _nextSubscriptionId;
 
+    /// <summary>True when <paramref name="subscriptionId"/> is currently registered.</summary>
+    /// <param name="subscriptionId">The subscription identifier.</param>
+    /// <returns>Whether the id is tracked.</returns>
+    public bool Contains(long subscriptionId) => _bySubscriptionId.ContainsKey(subscriptionId);
+
+    /// <summary>
+    ///     <see cref="Rebind"/> when the id is already tracked (so its stale reverse-map entries
+    ///     are dropped), <see cref="Register"/> otherwise. Used by the probe watcher, whose single
+    ///     reserved subscription is re-issued on every platform-set change.
+    /// </summary>
+    /// <param name="subscriptionId">The subscription identifier.</param>
+    /// <param name="bindings">The tag bindings for the subscription.</param>
+    public void Upsert(long subscriptionId, IReadOnlyList<TagBinding> bindings)
+    {
+        if (Contains(subscriptionId)) Rebind(subscriptionId, bindings);
+        else Register(subscriptionId, bindings);
+    }
+
     /// <summary>Gets the number of tracked subscriptions.</summary>
     public int TrackedSubscriptionCount => _bySubscriptionId.Count;
     /// <summary>Gets the number of tracked item handles.</summary>
@@ -100,6 +129,17 @@ internal sealed class SubscriptionRegistry
     }
 
 
+    /// <summary>
+    ///     Drop every subscription and the whole reverse map. Used by the driver's runtime teardown:
+    ///     once the gateway session is closed every item handle in here is dead, and the host
+    ///     re-issues its subscriptions (fresh ids) after the next <c>InitializeAsync</c>.
+    /// </summary>
+    public void Clear()
+    {
+        _bySubscriptionId.Clear();
+        _subscribersByItemHandle.Clear();
+    }
+
     /// <summary>Snapshot every active binding for diagnostic output.</summary>
     /// <returns>A flattened list of every tag binding across all tracked subscriptions.</returns>
     public IReadOnlyList<TagBinding> SnapshotAllBindings() =>

src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Components/Shared/Drivers/Forms/GalaxyDriverForm.razor

@@ -92,7 +92,7 @@
             <div class="col-md-3">
                 <label class="form-label">Discover page size</label>
                 <InputNumber @bind-Value="_form.Galaxy.RepositoryDiscoverPageSize" @bind-Value:after="EmitAsync" class="form-control form-control-sm" />
-                <div class="form-text">Default 5000 objects per page.</div>
+                <div class="form-text">Objects per DiscoverHierarchy page the driver asks the gateway for. Default 5000.</div>
             </div>
             <div class="col-md-3">
                 <div class="form-check form-switch mt-4">
@@ -138,7 +138,7 @@
             <div class="col-md-3">
                 <label class="form-label">Admin UI probe timeout (seconds)</label>
                 <InputNumber @bind-Value="_form.Galaxy.ProbeTimeoutSeconds" @bind-Value:after="EmitAsync" class="form-control form-control-sm" />
-                <div class="form-text">Max 60. Used by Test Connect. Default 30.</div>
+                <div class="form-text">Caps the Test Connect probe deadline (the probe's own budget still applies, so this only shortens it). Max 60. Default 30.</div>
             </div>
         </div>
     </div>