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 2cfc682e4..e548a138a 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/GalaxyDriver.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/GalaxyDriver.cs @@ -2162,12 +2162,16 @@ public sealed class GalaxyDriver "GalaxyDriver runtime is shutting down — no subscription can be served until the next InitializeAsync."); } if (_eventPump is not null) return _eventPump; - _eventPump = new EventPump( + // The fault callback carries the pump's identity, so a late fault from a pump that is no + // longer the live one (a stale pump of a reopened session) is not escalated (#865 review). + EventPump? created = null; + created = new EventPump( _subscriber!, _subscriptions, _logger, channelCapacity: _options.MxAccess.EventPumpChannelCapacity, clientName: _options.MxAccess.ClientName, - onStreamFault: OnEventPumpStreamFault, + onStreamFault: cause => OnEventPumpStreamFault(created, cause), onReplayGap: OnEventPumpReplayGap); + _eventPump = created; _eventPump.OnDataChange += OnPumpDataChange; _eventPump.Start(); return _eventPump; @@ -2441,7 +2445,7 @@ public sealed class GalaxyDriver /// replay; without this hand-off a transient transport drop permanently kills /// the event stream and GetHealth() keeps reporting Healthy. /// - private void OnEventPumpStreamFault(Exception cause) + private void OnEventPumpStreamFault(EventPump? source, Exception cause) { if (_runtimeStopping) { @@ -2451,6 +2455,20 @@ public sealed class GalaxyDriver return; } + if (source is not null) + { + bool stale; + lock (_pumpLock) stale = !ReferenceEquals(_eventPump, source) || source.Session != _subscriptions.CurrentSession; + if (stale) + { + // A pump that is no longer the live one (its session was reopened; the replay replaces or + // already replaced it). Its fault says nothing about the current transport — and reported + // while Replaying it would fail a good replay (#865 review). + _logger.LogDebug(cause, "GalaxyDriver {InstanceId} ignored a stream fault from a superseded event pump.", _driverInstanceId); + return; + } + } + var supervisor = _supervisor; if (supervisor is 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 43ab2d920..d6ca1318f 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 @@ -271,9 +271,11 @@ public sealed class ReconnectSupervisor : IDisposable // === Done — unless the transport failed again while the replay ran === bool faultedDuringReplay; + string? replayFault; lock (_stateLock) { faultedDuringReplay = _faultDuringReplay; + replayFault = _lastError; if (!faultedDuringReplay) { _lastError = null; @@ -285,7 +287,8 @@ public sealed class ReconnectSupervisor : IDisposable // 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); + "Galaxy replay completed but the transport failed again while it ran (attempt {Attempt}: {Fault}) — retrying", + attempt, replayFault); continue; } _logger.LogInformation("Galaxy reconnect succeeded after {Attempt} attempt(s)", attempt);