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
This commit is contained in:
Joseph Doherty
2026-11-11 13:12:33 -05:00
parent 8653960ea3
commit d80bdf6369
3 changed files with 61 additions and 5 deletions
@@ -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.");
}
@@ -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;
/// <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
@@ -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;
@@ -49,6 +49,35 @@ public sealed class ReconnectSupervisorTests
sup.CurrentState.ShouldBe(ReconnectSupervisor.State.Reopening, "the backoff still reads Reopening — why the flag exists");
}
/// <summary>
/// #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.
/// </summary>
[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");
}
/// <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()