fix(galaxy): #865 review round 4 - replay fails on a dead stream, backoff-aware wait

- ReplayAsync fails when its restarted pump's stream has already ended, so
  the supervisor backs off and stays Degraded instead of going Healthy over
  a dead stream (whose next advise re-report restarted at attempt 1 with no
  backoff - a tight reconnect loop).
- ReconnectSupervisor.AttemptInProgress (reopen through replay); advises
  wait for a stale pump's replacement only while an attempt is running, not
  for a whole backoff (the state still reads Reopening/Replaying there).
- ReconnectSupervisor: _disposed is set under the state lock and re-checked
  inside ReportTransportFailure, so a late report cannot start a recovery
  loop that StopAsync already missed.
- EventPump.HasEnded is a flag set before StreamRequested is released, and
  a stream that ends/faults on its first MoveNext is marked ended first.
- SubscriptionRegistry fast path: a miss now falls through to the locked
  path (lookup + session + window check atomically) instead of a lock-free
  re-lookup.
- Tests: replay over an immediately-ending stream fails; AttemptInProgress
  across attempt/backoff; ReportTransportFailure after stop; the
  reopen-window test uses a generous wait bound.

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

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

@@ -1056,7 +1056,19 @@ public sealed class GalaxyDriver
         // #865: every re-advise below pushes its initial value once, to whatever stream is attached.
         EventPump? replayPump;
         lock (_pumpLock) replayPump = _eventPump;
-        if (replayPump is not null) await AwaitCurrentEventStreamAsync(cancellationToken).ConfigureAwait(false);
+        if (replayPump is not null)
+        {
+            await AwaitCurrentEventStreamAsync(cancellationToken).ConfigureAwait(false);
+            // A replay whose fresh pump already died must FAIL, so the supervisor backs off and stays
+            // Degraded — succeeding would go Healthy with a dead stream, and the next advise's re-report
+            // would start a new cycle at attempt 1 (no backoff): a tight reconnect loop.
+            lock (_pumpLock) replayPump = _eventPump;
+            if (replayPump is { HasEnded: true })
+            {
+                throw new InvalidOperationException(
+                    "The Galaxy event stream ended during the reconnect replay — retrying after backoff.");
+            }
+        }
 
         // Only subscriptions still holding the previous session's handles: one registered after the
         // reopen is already live on the recreated session (re-subscribing it would double-advise).
@@ -2279,7 +2291,7 @@ public sealed class GalaxyDriver
     }
 
     /// <summary>
-    ///     True when a stale pump will be replaced shortly: the supervisor is reopening/replaying, or there is
+    ///     True when a stale pump will be replaced shortly: a reopen/replay attempt is running, or there is
     ///     no supervisor (the injected-seam path, where the replay is driven directly), or a restart is underway.
     /// </summary>
     private bool ReplayExpectedSoon()
@@ -2287,8 +2299,9 @@ public sealed class GalaxyDriver
         if (Volatile.Read(ref _pumpRestarting)) return true;
         var supervisor = _supervisor;
         if (supervisor is null) return true;
-        var state = supervisor.CurrentState;
-        return state is ReconnectSupervisor.State.Reopening or ReconnectSupervisor.State.Replaying;
+        // Not the state: between attempts (the backoff delay, up to MaxBackoff) it still reads
+        // Reopening/Replaying while no replay is coming any time soon.
+        return supervisor.AttemptInProgress;
     }
 
     /// <summary>A current-session pump's loop ended while the supervisor was not listening (see AwaitCurrentEventStreamAsync).</summary>

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

@@ -155,7 +155,11 @@ internal sealed class EventPump : IAsyncDisposable
     internal int Session => _session;
 
     /// <summary>True once the stream loop has exited (faulted, completed by the gateway, or disposed).</summary>
-    internal bool HasEnded => _loop is { IsCompleted: true };
+    internal bool HasEnded => Volatile.Read(ref _ended);
+
+    // Set at the very start of the stream loop's finally — before StreamRequested is completed there — so a
+    // waiter released by a dying loop always sees it ended (the loop task itself completes only later).
+    private bool _ended;
 
     private readonly TaskCompletionSource _streamRequested = new(TaskCreationOptions.RunContinuationsAsynchronously);
 
@@ -212,7 +216,19 @@ internal sealed class EventPump : IAsyncDisposable
             // Start the first MoveNext BEFORE awaiting it: that synchronous step issues the RPC, which is
             // the moment StreamRequested reports (#865).
             var next = enumerator.MoveNextAsync();
-            _streamRequested.TrySetResult();
+            if (next.IsCompleted)
+            {
+                // Already decided (an item, an immediate end, or a fault): consume it now, so a stream that
+                // ended or faulted at once is seen as ended (finally) BEFORE StreamRequested releases anyone.
+                var hasFirst = await next.ConfigureAwait(false);
+                if (!hasFirst) Volatile.Write(ref _ended, true);
+                _streamRequested.TrySetResult();
+                next = new ValueTask<bool>(hasFirst);
+            }
+            else
+            {
+                _streamRequested.TrySetResult();
+            }
             for (; await next.ConfigureAwait(false); next = enumerator.MoveNextAsync())
             {
                 var ev = enumerator.Current;
@@ -274,6 +290,7 @@ internal sealed class EventPump : IAsyncDisposable
         }
         finally
         {
+            Volatile.Write(ref _ended, true);
             // A loop that died before issuing the RPC must not strand a StreamRequested waiter.
             _streamRequested.TrySetResult();
             if (enumerator is not null)

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

@@ -104,6 +104,15 @@ public sealed class ReconnectSupervisor : IDisposable
         get { lock (_stateLock) return _state; }
     }
 
+    private bool _attemptInProgress;
+
+    /// <summary>
+    ///     True while a reopen or replay is actually running — false between attempts (the backoff delay,
+    ///     during which the state still reads Reopening/Replaying) and while Healthy. #865: an advise waits
+    ///     for the replay's pump only while one is imminent.
+    /// </summary>
+    public bool AttemptInProgress => Volatile.Read(ref _attemptInProgress);
+
     /// <summary>True when CurrentState != Healthy. Drivers map this to DriverState.Degraded.</summary>
     public bool IsDegraded
     {
@@ -136,6 +145,9 @@ public sealed class ReconnectSupervisor : IDisposable
 
         lock (_stateLock)
         {
+            // Re-checked under the lock (StopAsync sets it under the same lock): a report that passed the
+            // check above while a stop ran must not start a loop nothing would ever cancel.
+            ObjectDisposedException.ThrowIf(_disposed, this);
             _lastError = cause.Message;
             if (_state != State.Healthy)
             {
@@ -208,13 +220,21 @@ public sealed class ReconnectSupervisor : IDisposable
             // === Reopening phase ===
             lock (_stateLock) TransitionLocked(State.Reopening, _lastError);
 
+            // In progress from the reopen through the replay (no gap between the phases); cleared on any
+            // exit from the attempt — failure (then the backoff), cancellation or success.
+            Volatile.Write(ref _attemptInProgress, true);
             try
             {
                 await _reopen(ct).ConfigureAwait(false);
             }
-            catch (OperationCanceledException) when (ct.IsCancellationRequested) { return; }
+            catch (OperationCanceledException) when (ct.IsCancellationRequested)
+            {
+                Volatile.Write(ref _attemptInProgress, false);
+                return;
+            }
             catch (Exception ex)
             {
+                Volatile.Write(ref _attemptInProgress, false);
                 _logger.LogWarning(ex, "Galaxy reopen failed (attempt {Attempt})", attempt);
                 lock (_stateLock) { _lastError = ex.Message; }
                 continue; // back to backoff + retry
@@ -234,6 +254,10 @@ public sealed class ReconnectSupervisor : IDisposable
                 lock (_stateLock) { _lastError = ex.Message; }
                 continue; // back to backoff + retry
             }
+            finally
+            {
+                Volatile.Write(ref _attemptInProgress, false);
+            }
 
             // === Done ===
             lock (_stateLock)
@@ -282,12 +306,14 @@ public sealed class ReconnectSupervisor : IDisposable
     /// <returns>A task that completes once the recovery loop has exited.</returns>
     public async Task StopAsync()
     {
-        if (_disposed) return;
-        _disposed = true;
-
         CancellationTokenSource? cts;
         Task? loop;
-        lock (_stateLock) { cts = _loopCts; loop = _recoveryLoop; _loopCts = null; _recoveryLoop = null; }
+        lock (_stateLock)
+        {
+            if (_disposed) return;
+            _disposed = true;
+            cts = _loopCts; loop = _recoveryLoop; _loopCts = null; _recoveryLoop = null;
+        }
 
         // 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.

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

@@ -503,17 +503,11 @@ internal sealed class SubscriptionRegistry
             subscribers = ResolveSubscribers(ev.ItemHandle);
             if (subscribers.Count > 0) return EarlyEventOutcome.Resolved;
             AfterFastPathMissForTests?.Invoke();
-            if (Volatile.Read(ref _pendingSubscribes) == 0)
-            {
-                // #865 (found by its stress run): the lookup above and this read are not atomic. A
-                // subscribe can register the handle AND close its window in between — then "no window
-                // open" does not mean "stale", and dropping here lost the initial value for good (the
-                // first event of a bulk advise, while the dispatch thread was slow). The window closes
-                // only after its registration, so once no window is open a re-lookup sees every
-                // registration that window made; nothing is held, so no older event can be overtaken.
-                subscribers = ResolveSubscribers(ev.ItemHandle);
-                return subscribers.Count > 0 ? EarlyEventOutcome.Resolved : EarlyEventOutcome.Stale;
-            }
+            // #865 (found by its stress run): a miss is NOT decided here. The lookup above and an
+            // open-window check are not atomic — a subscribe can register the handle AND close its window
+            // in between, and declaring the event stale then lost its initial value for good (the first
+            // event of a bulk advise, while the dispatch thread was slow). Misses are rare; the locked path
+            // below re-does the lookup, the session check and the window check atomically.
         }
 
         lock (_parkGate)

tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/ProbeInitialValueTests.cs

@@ -51,6 +51,11 @@ public sealed class ProbeInitialValueTests
         /// <summary>Initial values pushed while no stream was attached — lost, as on the real gateway.</summary>
         public int DroppedInitialValues;
 
+        private int _streams;
+
+        /// <summary>Streams beyond this count end immediately after attaching (a gateway refusing/ending them).</summary>
+        public int StreamsThatStayOpen = int.MaxValue;
+
         /// <summary>The ScanState value the next advise reports.</summary>
         public volatile bool ScanStateRunning = true;
 
@@ -90,6 +95,11 @@ public sealed class ProbeInitialValueTests
                 Calls.Add("stream");
                 _attached = mine;
             }
+            if (Interlocked.Increment(ref _streams) > StreamsThatStayOpen)
+            {
+                lock (_gate) { if (ReferenceEquals(_attached, mine)) _attached = null; }
+                yield break; // the gateway completed this stream at once
+            }
             try
             {
                 await foreach (var ev in mine.Reader.ReadAllAsync(cancellationToken).ConfigureAwait(false))
@@ -208,6 +218,9 @@ public sealed class ProbeInitialValueTests
             platformSource: new FixedPlatformSource(Platforms))
         {
             RecreateOverrideForTests = _ => Task.CompletedTask,
+            // Generous: the window advise must still be waiting when the replay below starts the pump, even
+            // on a stalled agent; the wait ends the moment that pump exists.
+            StreamRequestWait = TimeSpan.FromSeconds(20),
         };
         await driver.InitializeAsync("{}", CancellationToken.None);
         await TestWait.UntilAsync(
@@ -315,6 +328,27 @@ public sealed class ProbeInitialValueTests
         sub.DroppedInitialValues.ShouldBe(0);
     }
 
+    /// <summary>
+    ///     A replay whose restarted pump's stream ends at once must FAIL (the supervisor then backs off and
+    ///     stays Degraded) rather than succeed onto a dead stream — succeeding went Healthy, and the next
+    ///     advise's re-report started a new cycle with no backoff: a tight reconnect loop. Positive control:
+    ///     the first replay, whose stream stays open, succeeds.
+    /// </summary>
+    [Fact]
+    public async Task Replay_fails_when_the_restarted_stream_ends_at_once()
+    {
+        var sub = new GatewayLikeSubscriber { StreamsThatStayOpen = 2 };
+        await using var driver = new GalaxyDriver("drv", Opts(), hierarchySource: null, subscriber: sub);
+        await driver.InitializeAsync("{}", CancellationToken.None);
+        await driver.SubscribeAsync(["Tank01.Level"], TimeSpan.Zero, CancellationToken.None); // stream 1
+
+        await driver.InvokeReplayForTestAsync(CancellationToken.None); // stream 2 stays open — succeeds
+
+        await Should.ThrowAsync<InvalidOperationException>(
+            () => driver.InvokeReplayForTestAsync(CancellationToken.None), // stream 3 ends at once
+            "a replay must not report success over a dead event stream");
+    }
+
     /// <summary>
     ///     A probe sync on a driver whose pump was never started (no platform source, e.g. the
     ///     DiscoverAsync path) still starts it before advising — the watcher's pre-subscribe hook, not just

tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/ReconnectSupervisorTests.cs

@@ -15,6 +15,56 @@ public sealed class ReconnectSupervisorTests
         new(InitialBackoffOverride: TimeSpan.FromMilliseconds(5),
             MaxBackoffOverride: TimeSpan.FromMilliseconds(20));
 
+    /// <summary>
+    ///     #865: AttemptInProgress is true while a reopen (through the replay) runs and FALSE during the backoff
+    ///     between attempts — when the state still reads Reopening. Advises wait for a replay only while one is
+    ///     actually running, not for a whole backoff.
+    /// </summary>
+    [Fact]
+    public async Task AttemptInProgress_is_true_during_an_attempt_and_false_during_the_backoff()
+    {
+        var reopenEntered = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+        var releaseReopen = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+        var reopens = 0;
+        using var sup = new ReconnectSupervisor(
+            reopen: async _ =>
+            {
+                if (Interlocked.Increment(ref reopens) == 1)
+                {
+                    reopenEntered.TrySetResult();
+                    await releaseReopen.Task;
+                    throw new IOException("gateway still down");
+                }
+            },
+            replay: _ => Task.CompletedTask,
+            backoffDelay: (_, _, _) => TimeSpan.FromMinutes(10)); // park in the backoff
+
+        sup.AttemptInProgress.ShouldBeFalse("Healthy: no attempt");
+        sup.ReportTransportFailure(new IOException("drop"));
+        await TestWait.ForAsync(reopenEntered.Task, "the first reopen must start");
+        sup.AttemptInProgress.ShouldBeTrue("a reopen is running");
+
+        releaseReopen.SetResult(); // the reopen fails -> backoff of 10 min
+        await TestWait.UntilAsync(() => !sup.AttemptInProgress, "the failed attempt must clear the flag");
+        sup.CurrentState.ShouldBe(ReconnectSupervisor.State.Reopening, "the backoff still reads Reopening — why the flag exists");
+    }
+
+    /// <summary>#865: a transport failure reported after StopAsync never starts a recovery loop (it throws ODE).</summary>
+    [Fact]
+    public async Task ReportTransportFailure_after_stop_never_starts_a_loop()
+    {
+        var reopens = 0;
+        var sup = new ReconnectSupervisor(
+            reopen: _ => { Interlocked.Increment(ref reopens); return Task.CompletedTask; },
+            replay: _ => Task.CompletedTask,
+            options: FastBackoff());
+        await sup.StopAsync();
+
+        Should.Throw<ObjectDisposedException>(() => sup.ReportTransportFailure(new IOException("late")));
+        sup.CurrentState.ShouldBe(ReconnectSupervisor.State.Healthy);
+        Volatile.Read(ref reopens).ShouldBe(0);
+    }
+
     /// <summary>Verifies that the supervisor starts in a healthy state with no errors.</summary>
     [Fact]
     public void InitialState_IsHealthy()