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()