dohertj2/lmxopcua
1
0
Fork 0
Code Issues 5 Pull Requests Actions Packages Projects Releases Wiki Activity

fix(galaxy): #865 review round 6 - ignore stream faults from a superseded pump

Browse Source 
Each EventPump's fault callback now carries the pump's identity; a fault
from a pump that is no longer the live one (or belongs to an older registry
session) is logged at Debug and not escalated, so a stale pump's late fault
cannot fail a good replay now that a failure during Replaying fails the
attempt. The replay-retry warning includes the fault.

Claude-Session: https://claude.ai/code/session_01GT67AD99zQQqpLCVm3p7xk
This commit is contained in:
Joseph Doherty
2026-11-11 13:16:01 -05:00
parent d80bdf6369
commit 13cc7b8353
2 changed files with 25 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/GalaxyDriver.cs
+21 -3
View File
@@ -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 <c>GetHealth()</c> keeps reporting Healthy.
/// </summary>
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)
{
src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/ReconnectSupervisor.cs
+4 -1
View File
@@ -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);