fix(galaxy): stranded subscriptions after a no-replay reconnect go Bad + Degraded (#862)

With Reconnect.ReplayOnSessionLost=false a recreated gateway session carries none of the old advises, so every pre-reconnect subscription went silently deaf: nodes kept their last value as Good and health returned to Healthy (only a Warning log). ReplayAsync now records the stranded subscriptions and publishes BadNoCommunication (StatusCodeMap.BadNoCommunication, 0x80310000; value null — no driver value cache) for every tag of each, straight to OnDataChange so LastSuccessfulRead is not stamped. GetHealth reports Degraded naming the setting while any stranded subscription is still registered; it clears on UnsubscribeAsync (the host's re-subscribe path) or runtime teardown, and pruned ids never pin it. A subscribe after the reconnect is an ordinary new-session subscription. With nothing else flowing, the comms-loss watchdog re-initializes the driver after 90 s (a delayed replay). Tests: Runtime/StrandedSubscriptionTests (6). Docs: drivers/Galaxy.md, which also no longer claims the replay uses the gateway's ReplaySubscriptions RPC. Claude-Session: https://claude.ai/code/session_01GT67AD99zQQqpLCVm3p7xk
2026-11-11 11:13:19 -05:00
parent 793935614f
commit 33e10c2266
5 changed files with 365 additions and 11 deletions
@@ -84,7 +84,7 @@ Project root files:
 - **`Gateway`** — endpoint, API key secret ref (`secret:galaxy/<name>` · `secret:otopcua/galaxy/<name>` · `env:OTOPCUA_GALAXY_<NAME>` · legacy `env:GALAXY_MXGW_<NAME>` · `dev:KEY` · literal; anything else, incl. `file:`, is refused — #815 (an already-deployed driver keeps resolving a pre-#815 ref on init and every gateway reconnect in deprecated warn-only mode until its next deploy), [`docs/security.md`](../security.md) §"Credential references are namespaced per driver"), TLS knobs, connect/call/stream timeouts. `StreamTimeoutSeconds = 0` keeps the long-lived `StreamEvents` RPC open for the driver's lifetime.
 - **`MxAccess`** — `ClientName` (must be unique per OtOpcUa instance — redundancy pairs enforce uniqueness at install time), `PublishingIntervalMs` (forwarded as `buffered_update_interval_ms` on subscribe), `WriteUserId` for ArchestrA secured-write, `EventPumpChannelCapacity` (default 50_000 — one second of headroom at 50k tags / 1Hz; tune via the `galaxy.events.dropped` metric).
 - **`Repository`** — `DiscoverPageSize`, `WatchDeployEvents`.
- **`Reconnect`** — `InitialBackoffMs`, `MaxBackoffMs`, `ReplayOnSessionLost` (calls the gateway's `ReplaySubscriptions` RPC after reconnect rather than re-issuing subscribe-bulk for every tag).
+- **`Reconnect`** — `InitialBackoffMs`, `MaxBackoffMs`, `ReplayOnSessionLost` (default `true`: after a gateway session is recreated, re-issue `SubscribeBulk` for every tracked subscription; `false` leaves them stranded — Bad + Degraded, see [Reconnect + Replay](#reconnect--replay)).
 - **`AlarmAck`** (optional, lmxopcua#749) — `FallbackUser`, `FallbackDomain`, `FallbackPasswordSecretRef`. See [Alarm acknowledge credential](#alarm-acknowledge-credential).

 ### Alarm acknowledge credential
@@ -147,7 +147,16 @@ Full per-field descriptions live in `Config/GalaxyDriverOptions.cs`. The full JS

 ## Reconnect + Replay

-`ReconnectSupervisor` owns an exponential-backoff loop bounded by `Reconnect.InitialBackoffMs` / `MaxBackoffMs`. On session loss it calls `GalaxyDriver.ReopenAsync`, which invokes `GalaxyMxSession.RecreateAsync` to dispose the stale/faulted session and client before rebuilding (`OpenSessionAsync` + `RegisterAsync`). Previously `ConnectAsync` was a no-op when a stale session handle was still present, so the reopen supervisor looped forever without recovering. After a successful reopen — when `ReplayOnSessionLost = true` — the supervisor calls the gateway's `ReplaySubscriptions` RPC with the cached subscription set from `SubscriptionRegistry` instead of re-subscribing tag-by-tag. The gateway's worker then re-issues `AdviseSupervisory` server-side under the apartment lock.
+`ReconnectSupervisor` owns an exponential-backoff loop bounded by `Reconnect.InitialBackoffMs` / `MaxBackoffMs`. On session loss it calls `GalaxyDriver.ReopenAsync`, which invokes `GalaxyMxSession.RecreateAsync` to dispose the stale/faulted session and client before rebuilding (`OpenSessionAsync` + `RegisterAsync`). Previously `ConnectAsync` was a no-op when a stale session handle was still present, so the reopen supervisor looped forever without recovering. After a successful reopen — when `ReplayOnSessionLost = true` (the default) — `ReplayAsync` re-issues `SubscribeBulk` for every subscription still holding the previous session's handles and rebinds the registry with the new ones (the gateway's batched `ReplaySubscriptions` RPC is not used).
+
+**`ReplayOnSessionLost = false` strands the subscriptions, visibly (#862).** A recreated gateway session carries none of the old session's advises, so with replay disabled every subscription that existed before the reconnect stays registered but receives nothing. That used to be silent — each node kept its last value as Good and health returned to `Healthy`; the only trace was a Warning. Now, following the [quality contract](../DriverLifecycle.md):
+
+- every tag of every stranded subscription is published **`BadNoCommunication`** once per reconnect, under that subscription's own handle (value null — the driver keeps no value cache, and a Bad sample's value is not client-observable; the next Good value comes from MXAccess's initial value when the tag is advised again);
+- `GetHealth()` reports **`Degraded`** with a message naming `Reconnect.ReplayOnSessionLost=false` and the stranded count, after the supervisor's own reconnect Degraded has ended;
+- `LastSuccessfulRead` is **not** advanced by those publishes. If no other data flows (no other live subscription), the host's [comms-loss watchdog](../DriverLifecycle.md) sees Degraded + a frozen `LastSuccessfulRead` and re-initializes the driver after 90 s — which re-subscribes everything, i.e. a delayed replay. Set `ReplayOnSessionLost = true` to recover immediately instead;
+- **Degraded clears** when the last stranded subscription is unsubscribed (the host unsubscribes the old handle on every data re-subscribe, e.g. a deploy, and on re-init) or the runtime is torn down. A `SubscribeAsync` issued after the reconnect is an ordinary new-session subscription: it receives data normally and does not count as stranded.
+
+The per-platform `ScanState` probe watcher is not replayed when `ReplayOnSessionLost = false` either (unchanged), so `/hosts` per-platform status also stops updating until a re-init.

 The session-less alarm feed (`GatewayGalaxyAlarmFeed`) and alarm acknowledger (`GatewayGalaxyAlarmAcknowledger`) run on a separate always-on `_ownedMxClient` that is intentionally **not** recreated when `ReconnectSupervisor` rebuilds the worker session. The feed has its own re-invoke loop (~5 s backoff) that reopens `StreamAlarms` after any stream fault, and gRPC.NET's channel auto-reconnect recovers the underlying HTTP/2 connection after a gateway restart — so both the alarm stream and individual ack calls recover without client replacement. The acknowledger is completely stateless between calls: each `AcknowledgeAsync` issues a single unary RPC and returns; no dead-client state is latched on failure. Channel-level keepalive hardening (TCP keep-alive intervals, gRPC ping frames) would require exposing additional knobs on `MxGatewayClientOptions` in the sibling `mxaccessgw` repo — a future option if flaky long-lived connections are observed in production.

@@ -107,6 +107,13 @@ public sealed class GalaxyDriver
    // GatewayGalaxySubscriber over a connected GalaxyMxSession.
    private IGalaxySubscriber? _subscriber;
    private readonly SubscriptionRegistry _subscriptions = new();
+
+    // #862: subscriptions a reconnect left without advises because Reconnect.ReplayOnSessionLost=false.
+    // Their tags were published BadNoCommunication and the driver reports Degraded while any of them is
+    // still registered. Cleared by UnsubscribeAsync (the host re-subscribes on every deploy / re-init) and
+    // by the runtime teardown; GetHealth also prunes ids the registry no longer holds, so an unsubscribe
+    // racing the replay can never pin Degraded.
+    private readonly System.Collections.Concurrent.ConcurrentDictionary<long, byte> _strandedSubscriptions = new();
    private EventPump? _eventPump;
    private readonly Lock _pumpLock = new();

@@ -693,6 +700,7 @@ public sealed class GalaxyDriver
        lock (_registryGate)
        {
            _subscriptions.Clear();
+            _strandedSubscriptions.Clear();
            // A new session too: a late PurgeParked from a call the teardown orphaned (captured the old
            // session) must never reach the next runtime's parked values.
            _subscriptions.StartNewSession();
@@ -1055,15 +1063,11 @@ public sealed class GalaxyDriver
        // Honor ReplayOnSessionLost. When operators opt out (false) the per-tag SubscribeBulk fan-out
        // is skipped, and the recreated gateway session carries NONE of the old advises — those
        // subscriptions stay deaf (their handles are dead and no longer resolve) until re-subscribed.
+        // #862: deaf must not look healthy. Every tag of every stranded subscription is published
+        // BadNoCommunication and the driver reports Degraded until the stranded subscriptions are gone.
        if (!_options.Reconnect.ReplayOnSessionLost)
        {
-            if (entries.Length > 0)
-            {
-                _logger.LogWarning(
-                    "GalaxyDriver {InstanceId} reconnect replay skipped — ReplayOnSessionLost=false; {SubCount} subscription(s) " +
-                    "receive no further data until they are re-subscribed (the recreated gateway session has none of their advises).",
-                    _driverInstanceId, entries.Length);
-            }
+            if (entries.Length > 0) MarkSubscriptionsStranded(entries);
            return;
        }

@@ -1087,6 +1091,77 @@ public sealed class GalaxyDriver
            _driverInstanceId, entries.Length, tagCount);
    }

+    /// <summary>
+    ///     #862 — a reconnect with <see cref="GalaxyReconnectOptions.ReplayOnSessionLost"/> = false left
+    ///     <paramref name="entries"/> without advises on the recreated gateway session. Records them as
+    ///     stranded (so <see cref="GetHealth"/> reports Degraded) and publishes every one of their tags
+    ///     <c>BadNoCommunication</c> — the quality contract's comms-loss status — so no node keeps serving
+    ///     its last value as Good. The value is null: the driver keeps no value cache, and a Bad sample's
+    ///     value is not client-observable anyway (the SDK nulls it on read); the node is restated by the
+    ///     initial value MXAccess delivers when the tag is next advised (a re-subscribe).
+    ///     Published straight to <see cref="OnDataChange"/>, never through the pump path: that would
+    ///     stamp <c>LastSuccessfulRead</c> for data that never arrived.
+    /// </summary>
+    /// <param name="entries">The subscriptions the reconnect stranded.</param>
+    private void MarkSubscriptionsStranded(
+        IReadOnlyList<(long SubscriptionId, IReadOnlyList<TagBinding> Bindings)> entries)
+    {
+        var tagCount = 0;
+        foreach (var (subscriptionId, _) in entries) _strandedSubscriptions[subscriptionId] = 0;
+
+        _logger.LogWarning(
+            "GalaxyDriver {InstanceId} reconnect replay skipped — Reconnect.ReplayOnSessionLost=false; {SubCount} subscription(s) " +
+            "receive no further data until they are re-subscribed (the recreated gateway session has none of their advises). " +
+            "Their tags are published BadNoCommunication and the driver reports Degraded until then.",
+            _driverInstanceId, entries.Count);
+
+        var now = DateTime.UtcNow;
+        var snapshot = new DataValueSnapshot(
+            Value: null, StatusCode: StatusCodeMap.BadNoCommunication, SourceTimestampUtc: now, ServerTimestampUtc: now);
+        foreach (var (subscriptionId, bindings) in entries)
+        {
+            // Unsubscribed while the reconnect ran: the host no longer holds this handle.
+            if (!_subscriptions.Contains(subscriptionId)) continue;
+            var handle = new GalaxySubscriptionHandle(subscriptionId);
+            foreach (var binding in bindings)
+            {
+                tagCount++;
+                try
+                {
+                    OnDataChange?.Invoke(this, new DataChangeEventArgs(handle, ToRawPath(binding.FullReference), snapshot));
+                }
+                catch (Exception ex)
+                {
+                    _logger.LogWarning(ex,
+                        "GalaxyDriver {InstanceId} OnDataChange handler threw publishing a stranded tag {FullRef} — continuing.",
+                        _driverInstanceId, binding.FullReference);
+                }
+            }
+        }
+        _logger.LogDebug(
+            "GalaxyDriver {InstanceId} published BadNoCommunication for {TagCount} stranded tag(s).", _driverInstanceId, tagCount);
+    }
+
+    /// <summary>
+    ///     #862 — the number of subscriptions a reconnect stranded (replay disabled) that are still registered.
+    ///     Prunes ids the registry no longer holds.
+    /// </summary>
+    /// <returns>The live stranded-subscription count.</returns>
+    private int CountStrandedSubscriptions()
+    {
+        if (_strandedSubscriptions.IsEmpty) return 0;
+        var count = 0;
+        foreach (var id in _strandedSubscriptions.Keys)
+        {
+            if (_subscriptions.Contains(id)) count++;
+            else _strandedSubscriptions.TryRemove(id, out _);
+        }
+        return count;
+    }
+
+    /// <summary>Test seam — the live stranded-subscription count (#862).</summary>
+    internal int StrandedSubscriptionCountForTests => CountStrandedSubscriptions();
+
    /// <summary>
    ///     Re-issues <c>SubscribeBulk</c> for every tracked subscription and rebinds the registry
    ///     with the item handles the gateway returned. Shared by the reconnect replay path and
@@ -1460,6 +1535,21 @@ public sealed class GalaxyDriver
        {
            return new DriverHealth(DriverState.Degraded, _health.LastSuccessfulRead, _supervisor.LastError);
        }
+
+        // #862: the transport is back but a reconnect with ReplayOnSessionLost=false left subscriptions
+        // without advises. LastSuccessfulRead is NOT advanced here — it only moves on real stream data, so
+        // if nothing else flows the host's comms-loss watchdog re-initializes the driver (which
+        // re-subscribes everything) exactly as for any other sustained loss.
+        var stranded = CountStrandedSubscriptions();
+        if (stranded > 0)
+        {
+            return new DriverHealth(
+                DriverState.Degraded,
+                _health.LastSuccessfulRead,
+                $"{stranded} Galaxy subscription(s) stranded by a gateway reconnect: Reconnect.ReplayOnSessionLost=false, " +
+                "so the recreated gateway session carries none of their advises — their tags are BadNoCommunication " +
+                "until they are re-subscribed. Set ReplayOnSessionLost=true to replay them on reconnect.");
+        }
        return _health;
    }

@@ -1957,6 +2047,7 @@ public sealed class GalaxyDriver
        }

        var bindings = _subscriptions.Remove(gsh.SubscriptionId, out var liveInCurrentSession);
+        _strandedSubscriptions.TryRemove(gsh.SubscriptionId, out _); // #862: gone ⇒ no longer degrades health
        if (bindings is null) return; // already removed or never registered

        // Not yet rebound after a reconnect: its handles died with the old session and the new one may
@@ -49,6 +49,11 @@ internal static class StatusCodeMap
    public const uint BadDeviceFailure                = 0x808B0000u;
    public const uint BadSensorFailure                = 0x808C0000u;
    public const uint BadCommunicationError           = 0x80050000u;
+    /// <summary>
+    ///     The quality contract's comms-loss status (<c>docs/DriverLifecycle.md</c>) — published for the tags
+    ///     of a subscription a reconnect left without advises (#862).
+    /// </summary>
+    public const uint BadNoCommunication              = 0x80310000u;
    public const uint BadOutOfService                 = 0x808D0000u;
    public const uint BadWaitingForInitialData        = 0x80320000u;
    public const uint BadInternalError                = 0x80020000u;
@@ -63,8 +63,9 @@ public sealed class GalaxyDriverInfrastructureTests
    {
        // ReplayOnSessionLost was a dangling option — defined + documented but never
        // read. After the fix, setting it to false makes the reconnect replay path
-        // skip SubscribeBulk (operator opts out of replay; the gateway's session-level
-        // ReplaySubscriptions handles state restoration).
+        // skip SubscribeBulk (operator opts out of replay). The recreated gateway session
+        // carries none of the old advises, so those subscriptions are stranded — published
+        // BadNoCommunication + driver Degraded (#862, StrandedSubscriptionTests).
        var sub = new ReplayCountingSubscriber();
        var opts = new GalaxyDriverOptions(
            new GalaxyGatewayOptions("https://mxgw.test:5001", "key"),
@@ -0,0 +1,248 @@
+using System.Collections.Concurrent;
+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.OtOpcUa.Core.Abstractions;
+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;
+
+/// <summary>
+///     #862 — with <c>Reconnect.ReplayOnSessionLost=false</c> a gateway reconnect keeps the driver's
+///     subscriptions but the recreated session carries none of their advises. That used to be invisible:
+///     every node kept its last value with Good quality and health went back to Healthy. Now each tag of
+///     each stranded subscription is published <c>BadNoCommunication</c> and the driver reports Degraded
+///     until those subscriptions are unsubscribed (or the runtime is torn down).
+/// </summary>
+public sealed class StrandedSubscriptionTests
+{
+    private const uint BadNoCommunication = 0x80310000u;
+
+    private static GalaxyDriverOptions Opts(bool replay) => new(
+        new GalaxyGatewayOptions("https://mxgw.test:5001", "key"),
+        new GalaxyMxAccessOptions("OtOpcUa-A"),
+        new GalaxyRepositoryOptions(WatchDeployEvents: false),
+        new GalaxyReconnectOptions(ReplayOnSessionLost: replay));
+
+    private static MxEvent DataChange(int handle, double value) => new()
+    {
+        Family = MxEventFamily.OnDataChange,
+        ItemHandle = handle,
+        Value = new MxValue { DoubleValue = value },
+        Quality = 192,
+        SourceTimestamp = Timestamp.FromDateTime(DateTime.UtcNow),
+    };
+
+    private static GalaxyDriver NewDriver(bool replay, HandleSubscriber subscriber) =>
+        new("g", Opts(replay), hierarchySource: null, dataReader: null, dataWriter: null, subscriber: subscriber)
+        {
+            RecreateOverrideForTests = _ => Task.CompletedTask,
+        };
+
+    private static async Task ReconnectAsync(GalaxyDriver driver)
+    {
+        await driver.InvokeReopenForTestAsync(TestContext.Current.CancellationToken);
+        await driver.InvokeReplayForTestAsync(TestContext.Current.CancellationToken);
+    }
+
+    /// <summary>
+    ///     The reconnect publishes BadNoCommunication for every tag of every stranded subscription, under
+    ///     that subscription's own handle, and degrades health with a message naming the setting.
+    /// </summary>
+    [Fact]
+    public async Task Replay_disabled_reconnect_publishes_BadNoCommunication_and_degrades()
+    {
+        var subscriber = new HandleSubscriber();
+        using var driver = NewDriver(replay: false, subscriber);
+        var events = new ConcurrentQueue<DataChangeEventArgs>();
+        driver.OnDataChange += (_, e) => events.Enqueue(e);
+
+        var ab = (GalaxySubscriptionHandle)await driver.SubscribeAsync(["Tank.A", "Tank.B"], TimeSpan.Zero, TestContext.Current.CancellationToken);
+        var c = (GalaxySubscriptionHandle)await driver.SubscribeAsync(["Tank.C"], TimeSpan.Zero, TestContext.Current.CancellationToken);
+        driver.GetHealth().State.ShouldNotBe(DriverState.Degraded);
+        var callsBefore = Volatile.Read(ref subscriber.SubscribeCalls);
+        var lastReadBefore = driver.GetHealth().LastSuccessfulRead;
+
+        await ReconnectAsync(driver);
+
+        Volatile.Read(ref subscriber.SubscribeCalls).ShouldBe(callsBefore, "replay is disabled — nothing is re-advised");
+        var published = events.ToArray();
+        published.Select(e => (e.FullReference, ((GalaxySubscriptionHandle)e.SubscriptionHandle).SubscriptionId))
+            .OrderBy(x => x.FullReference)
+            .ShouldBe([("Tank.A", ab.SubscriptionId), ("Tank.B", ab.SubscriptionId), ("Tank.C", c.SubscriptionId)]);
+        published.ShouldAllBe(e => e.Snapshot.StatusCode == BadNoCommunication && e.Snapshot.Value == null);
+
+        var health = driver.GetHealth();
+        health.State.ShouldBe(DriverState.Degraded);
+        health.LastError.ShouldNotBeNull().ShouldContain("ReplayOnSessionLost");
+        health.LastError.ShouldContain("2 Galaxy subscription(s)");
+        health.LastSuccessfulRead.ShouldBe(lastReadBefore, "a synthetic Bad publish is not evidence of a live gateway");
+        driver.StrandedSubscriptionCountForTests.ShouldBe(2);
+    }
+
+    /// <summary>
+    ///     Degraded clears exactly when the last stranded subscription is unsubscribed (what the host does on
+    ///     every re-subscribe), and the stranded handles' dead numbers are never sent to the new session.
+    /// </summary>
+    [Fact]
+    public async Task Degraded_clears_once_every_stranded_subscription_is_unsubscribed()
+    {
+        var subscriber = new HandleSubscriber();
+        using var driver = NewDriver(replay: false, subscriber);
+        var a = await driver.SubscribeAsync(["Tank.A"], TimeSpan.Zero, TestContext.Current.CancellationToken);
+        var b = await driver.SubscribeAsync(["Tank.B"], TimeSpan.Zero, TestContext.Current.CancellationToken);
+
+        await ReconnectAsync(driver);
+        driver.GetHealth().State.ShouldBe(DriverState.Degraded);
+
+        await driver.UnsubscribeAsync(a, TestContext.Current.CancellationToken);
+        driver.GetHealth().State.ShouldBe(DriverState.Degraded, "Tank.B is still stranded");
+        driver.GetHealth().LastError.ShouldNotBeNull().ShouldContain("1 Galaxy subscription(s)");
+
+        await driver.UnsubscribeAsync(b, TestContext.Current.CancellationToken);
+        driver.GetHealth().State.ShouldNotBe(DriverState.Degraded);
+        driver.StrandedSubscriptionCountForTests.ShouldBe(0);
+        subscriber.Unsubscribed.ShouldBeEmpty("old-session handle numbers must never reach the recreated session");
+    }
+
+    /// <summary>
+    ///     A subscribe after the reconnect is an ordinary new-session subscription: it receives data, is
+    ///     not counted as stranded, and keeps working once the stranded ones are gone.
+    /// </summary>
+    [Fact]
+    public async Task A_new_subscribe_after_the_reconnect_works_and_is_not_stranded()
+    {
+        var subscriber = new HandleSubscriber();
+        using var driver = NewDriver(replay: false, subscriber);
+        var delivered = new ConcurrentQueue<DataChangeEventArgs>();
+        driver.OnDataChange += (_, e) => delivered.Enqueue(e);
+        var old = await driver.SubscribeAsync(["Tank.A"], TimeSpan.Zero, TestContext.Current.CancellationToken);
+
+        await ReconnectAsync(driver);
+        driver.StrandedSubscriptionCountForTests.ShouldBe(1);
+
+        var fresh = (GalaxySubscriptionHandle)await driver.SubscribeAsync(["Tank.A"], TimeSpan.Zero, TestContext.Current.CancellationToken);
+        driver.StrandedSubscriptionCountForTests.ShouldBe(1, "the new subscription holds live new-session handles");
+        var freshHandle = subscriber.LastHandle;
+
+        subscriber.Emit(DataChange(freshHandle, 7.5));
+        await TestWait.UntilAsync(
+            () => delivered.Any(e => e.Snapshot.StatusCode == 0 && (double?)e.Snapshot.Value == 7.5),
+            "a subscribe after the reconnect must receive data");
+        delivered.Last().SubscriptionHandle.ShouldBe(fresh);
+
+        await driver.UnsubscribeAsync(old, TestContext.Current.CancellationToken);
+        driver.GetHealth().State.ShouldNotBe(DriverState.Degraded);
+        subscriber.Unsubscribed.ShouldBeEmpty();
+        await driver.UnsubscribeAsync(fresh, TestContext.Current.CancellationToken);
+        subscriber.Unsubscribed.ToArray().ShouldBe(new[] { freshHandle }, "the live subscription unadvises normally");
+    }
+
+    /// <summary>Default replay re-advises everything: no Bad publish, no Degraded.</summary>
+    [Fact]
+    public async Task Replay_enabled_reconnect_publishes_no_Bad_and_stays_undegraded()
+    {
+        var subscriber = new HandleSubscriber();
+        using var driver = NewDriver(replay: true, subscriber);
+        var events = new ConcurrentQueue<DataChangeEventArgs>();
+        driver.OnDataChange += (_, e) => events.Enqueue(e);
+        await driver.SubscribeAsync(["Tank.A"], TimeSpan.Zero, TestContext.Current.CancellationToken);
+
+        await ReconnectAsync(driver);
+
+        // Positive control: the replayed handle delivers, so the publish path was live while we looked.
+        subscriber.Emit(DataChange(subscriber.LastHandle, 1.0));
+        await TestWait.UntilAsync(() => !events.IsEmpty, "the replayed subscription must deliver");
+        events.ShouldAllBe(e => e.Snapshot.StatusCode != BadNoCommunication);
+        driver.GetHealth().State.ShouldNotBe(DriverState.Degraded);
+        driver.StrandedSubscriptionCountForTests.ShouldBe(0);
+    }
+
+    /// <summary>A second reconnect re-asserts the stranded state (still one entry per subscription).</summary>
+    [Fact]
+    public async Task A_second_reconnect_restates_the_same_stranded_subscriptions()
+    {
+        var subscriber = new HandleSubscriber();
+        using var driver = NewDriver(replay: false, subscriber);
+        var events = new ConcurrentQueue<DataChangeEventArgs>();
+        driver.OnDataChange += (_, e) => events.Enqueue(e);
+        await driver.SubscribeAsync(["Tank.A"], TimeSpan.Zero, TestContext.Current.CancellationToken);
+
+        await ReconnectAsync(driver);
+        await ReconnectAsync(driver);
+
+        events.Count.ShouldBe(2);
+        events.ShouldAllBe(e => e.Snapshot.StatusCode == BadNoCommunication);
+        driver.StrandedSubscriptionCountForTests.ShouldBe(1);
+    }
+
+    /// <summary>The runtime teardown forgets the stranded set along with the registry.</summary>
+    [Fact]
+    public async Task Shutdown_clears_the_stranded_state()
+    {
+        var subscriber = new HandleSubscriber();
+        using var driver = NewDriver(replay: false, subscriber);
+        await driver.SubscribeAsync(["Tank.A"], TimeSpan.Zero, TestContext.Current.CancellationToken);
+        await ReconnectAsync(driver);
+        driver.StrandedSubscriptionCountForTests.ShouldBe(1);
+
+        await driver.ShutdownAsync(TestContext.Current.CancellationToken);
+
+        driver.StrandedSubscriptionCountForTests.ShouldBe(0);
+        driver.GetHealth().State.ShouldNotBe(DriverState.Degraded);
+    }
+
+    /// <summary>Hands out monotonic item handles, records unsubscribes, and streams emitted events.</summary>
+    private sealed class HandleSubscriber : IGalaxySubscriber
+    {
+        private readonly Channel<MxEvent> _events = Channel.CreateUnbounded<MxEvent>();
+        private int _nextHandle = 100;
+
+        /// <summary>Number of SubscribeBulk calls.</summary>
+        public int SubscribeCalls;
+
+        /// <summary>Gets the most recently issued item handle.</summary>
+        public int LastHandle => Volatile.Read(ref _nextHandle);
+
+        /// <summary>Gets every item handle passed to UnsubscribeBulk.</summary>
+        public ConcurrentQueue<int> Unsubscribed { get; } = new();
+
+        /// <summary>Pushes an event onto the stream.</summary>
+        /// <param name="ev">The event.</param>
+        public void Emit(MxEvent ev) => _events.Writer.TryWrite(ev).ShouldBeTrue();
+
+        /// <inheritdoc />
+        public Task<IReadOnlyList<SubscribeResult>> SubscribeBulkAsync(
+            IReadOnlyList<string> fullReferences, int bufferedUpdateIntervalMs, CancellationToken cancellationToken)
+        {
+            Interlocked.Increment(ref SubscribeCalls);
+            IReadOnlyList<SubscribeResult> results = [.. fullReferences.Select(r => new SubscribeResult
+            {
+                TagAddress = r,
+                ItemHandle = Interlocked.Increment(ref _nextHandle),
+                WasSuccessful = true,
+            })];
+            return Task.FromResult(results);
+        }
+
+        /// <inheritdoc />
+        public Task UnsubscribeBulkAsync(IReadOnlyList<int> itemHandles, CancellationToken cancellationToken)
+        {
+            foreach (var h in itemHandles) Unsubscribed.Enqueue(h);
+            return Task.CompletedTask;
+        }
+
+        /// <inheritdoc />
+        public async IAsyncEnumerable<MxEvent> StreamEventsAsync([EnumeratorCancellation] CancellationToken cancellationToken)
+        {
+            await foreach (var ev in _events.Reader.ReadAllAsync(cancellationToken).ConfigureAwait(false))
+            {
+                yield return ev;
+            }
+        }
+    }
+}