From 8653960ea3c841787606947d4475e9f751eeb443 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Wed, 11 Nov 2026 13:07:26 -0500 Subject: [PATCH] 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 --- .../GalaxyDriver.cs | 21 ++++++-- .../Runtime/EventPump.cs | 21 +++++++- .../Runtime/ReconnectSupervisor.cs | 36 +++++++++++-- .../Runtime/SubscriptionRegistry.cs | 16 ++---- .../Runtime/ProbeInitialValueTests.cs | 34 +++++++++++++ .../Runtime/ReconnectSupervisorTests.cs | 50 +++++++++++++++++++ 6 files changed, 156 insertions(+), 22 deletions(-) diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/GalaxyDriver.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/GalaxyDriver.cs index c9dae1420..8800aa65f 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/GalaxyDriver.cs +++ b/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 } /// - /// 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. /// 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; } /// A current-session pump's loop ended while the supervisor was not listening (see AwaitCurrentEventStreamAsync). diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/EventPump.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/EventPump.cs index dc8905f85..cdb36a264 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/EventPump.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/EventPump.cs @@ -155,7 +155,11 @@ internal sealed class EventPump : IAsyncDisposable internal int Session => _session; /// True once the stream loop has exited (faulted, completed by the gateway, or disposed). - 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(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) diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/ReconnectSupervisor.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/ReconnectSupervisor.cs index 7b6c64101..41c1e5a52 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/ReconnectSupervisor.cs +++ b/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; + + /// + /// 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. + /// + public bool AttemptInProgress => Volatile.Read(ref _attemptInProgress); + /// True when CurrentState != Healthy. Drivers map this to DriverState.Degraded. 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 /// A task that completes once the recovery loop has exited. 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. diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/SubscriptionRegistry.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/SubscriptionRegistry.cs index 95e6f83d9..5f4aaf82f 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/SubscriptionRegistry.cs +++ b/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) diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/ProbeInitialValueTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/ProbeInitialValueTests.cs index 36f152888..7eabd47ab 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/ProbeInitialValueTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/ProbeInitialValueTests.cs @@ -51,6 +51,11 @@ public sealed class ProbeInitialValueTests /// Initial values pushed while no stream was attached — lost, as on the real gateway. public int DroppedInitialValues; + private int _streams; + + /// Streams beyond this count end immediately after attaching (a gateway refusing/ending them). + public int StreamsThatStayOpen = int.MaxValue; + /// The ScanState value the next advise reports. 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); } + /// + /// 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. + /// + [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( + () => driver.InvokeReplayForTestAsync(CancellationToken.None), // stream 3 ends at once + "a replay must not report success over a dead event stream"); + } + /// /// 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 diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/ReconnectSupervisorTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/ReconnectSupervisorTests.cs index accc819cf..b99f3b43c 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/ReconnectSupervisorTests.cs +++ b/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)); + /// + /// #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. + /// + [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"); + } + + /// #865: a transport failure reported after StopAsync never starts a recovery loop (it throws ODE). + [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(() => sup.ReportTransportFailure(new IOException("late"))); + sup.CurrentState.ShouldBe(ReconnectSupervisor.State.Healthy); + Volatile.Read(ref reopens).ShouldBe(0); + } + /// Verifies that the supervisor starts in a healthy state with no errors. [Fact] public void InitialState_IsHealthy()