From d80bdf63690d13c7858ac3f7b2c9eb91929e290b Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Wed, 11 Nov 2026 13:12:33 -0500 Subject: [PATCH] fix(galaxy): #865 review round 5 - a failure during the replay fails the attempt ReconnectSupervisor swallowed a transport failure reported while Replaying, so a restarted event stream refused/ended a round trip after the replay began still let the replay go Healthy over a dead stream (and the next advise's re-report restarted at attempt 1 with no backoff). A failure reported during Replaying is now remembered and turns that attempt's success into a retry with backoff. ReplayAsync's synchronous dead-pump throw yields to a teardown's cancellation. Claude-Session: https://claude.ai/code/session_01GT67AD99zQQqpLCVm3p7xk --- .../GalaxyDriver.cs | 4 +++ .../Runtime/ReconnectSupervisor.cs | 33 ++++++++++++++++--- .../Runtime/ReconnectSupervisorTests.cs | 29 ++++++++++++++++ 3 files changed, 61 insertions(+), 5 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 8800aa65f..2cfc682e4 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/GalaxyDriver.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/GalaxyDriver.cs @@ -1065,6 +1065,10 @@ public sealed class GalaxyDriver lock (_pumpLock) replayPump = _eventPump; if (replayPump is { HasEnded: true }) { + // A teardown cancelling this replay can end the pump too: that is a clean cancellation. + cancellationToken.ThrowIfCancellationRequested(); + // The synchronous case; a stream that ends a round trip later reports a transport failure + // while Replaying, which the supervisor itself turns into a failed attempt. throw new InvalidOperationException( "The Galaxy event stream ended during the reconnect replay — retrying after backoff."); } 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 41c1e5a52..43ab2d920 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 @@ -106,6 +106,9 @@ public sealed class ReconnectSupervisor : IDisposable private bool _attemptInProgress; + // Set (under _stateLock) by a ReportTransportFailure that arrives while Replaying; cleared on entering Replaying. + private bool _faultDuringReplay; + /// /// 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 @@ -151,7 +154,10 @@ public sealed class ReconnectSupervisor : IDisposable _lastError = cause.Message; if (_state != State.Healthy) { - // Already recovering — nothing else to do. + // Already recovering — nothing else to do, except: a failure DURING the replay means the + // replay ran onto a transport that is already gone again (#865 — e.g. the restarted event + // stream refused or ended); the replay's success must not then be reported as Healthy. + if (_state == State.Replaying) _faultDuringReplay = true; _logger.LogDebug("Transport failure reported during {State}: {Message}", _state, cause.Message); return; } @@ -241,7 +247,11 @@ public sealed class ReconnectSupervisor : IDisposable } // === Replaying phase === - lock (_stateLock) TransitionLocked(State.Replaying, _lastError); + lock (_stateLock) + { + _faultDuringReplay = false; + TransitionLocked(State.Replaying, _lastError); + } try { @@ -259,11 +269,24 @@ public sealed class ReconnectSupervisor : IDisposable Volatile.Write(ref _attemptInProgress, false); } - // === Done === + // === Done — unless the transport failed again while the replay ran === + bool faultedDuringReplay; lock (_stateLock) { - _lastError = null; - TransitionLocked(State.Healthy, null); + faultedDuringReplay = _faultDuringReplay; + if (!faultedDuringReplay) + { + _lastError = null; + TransitionLocked(State.Healthy, null); + } + } + if (faultedDuringReplay) + { + // Going Healthy here would hide a dead transport (the fault was swallowed above), and the + // next report would restart at attempt 1 with no backoff. Retry this cycle with backoff. + _logger.LogWarning( + "Galaxy replay completed but the transport failed again while it ran (attempt {Attempt}) — retrying", attempt); + continue; } _logger.LogInformation("Galaxy reconnect succeeded after {Attempt} attempt(s)", attempt); return; 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 b99f3b43c..10ecb016a 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 @@ -49,6 +49,35 @@ public sealed class ReconnectSupervisorTests sup.CurrentState.ShouldBe(ReconnectSupervisor.State.Reopening, "the backoff still reads Reopening — why the flag exists"); } + /// + /// #865 review: a transport failure reported WHILE the replay runs (e.g. the restarted event stream is + /// refused a round trip later) must fail that attempt — not go Healthy over a dead transport. The + /// supervisor retries (with backoff) and goes Healthy only after a clean replay. Without the fix the + /// first replay's success went Healthy and replay ran once. + /// + [Fact] + public async Task A_failure_reported_during_the_replay_fails_that_attempt() + { + ReconnectSupervisor? self = null; + var replays = 0; + using var sup = new ReconnectSupervisor( + reopen: _ => Task.CompletedTask, + replay: _ => + { + if (Interlocked.Increment(ref replays) == 1) + { + self!.ReportTransportFailure(new IOException("stream refused during replay")); + } + return Task.CompletedTask; + }, + options: FastBackoff()); + self = sup; + + sup.ReportTransportFailure(new IOException("drop")); + await TestWait.HealthyAsync(sup, "the second, clean replay must recover"); + Volatile.Read(ref replays).ShouldBe(2, "the faulted replay must not count as recovered"); + } + /// #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()