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 e548a138a..fb301e855 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/GalaxyDriver.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/GalaxyDriver.cs @@ -2445,6 +2445,11 @@ public sealed class GalaxyDriver /// replay; without this hand-off a transient transport drop permanently kills /// the event stream and GetHealth() keeps reporting Healthy. /// + private int _escalatedStreamFaults; + + /// Test seam (#865): stream faults that passed the superseded-pump filter (escalated toward the supervisor). + internal int EscalatedStreamFaultsForTests => Volatile.Read(ref _escalatedStreamFaults); + private void OnEventPumpStreamFault(EventPump? source, Exception cause) { if (_runtimeStopping) @@ -2461,13 +2466,16 @@ public sealed class GalaxyDriver 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). + // A pump that is no longer the live one: the identity check does the work (the replay + // replaces/replaced it). The session part only matters between a reopen and its replay, + // where the supervisor ignores faults anyway — ⚠️ a future session bump while Healthy that + // keeps the same pump would make its real faults look stale here. Its fault says nothing + // about the current transport, and reported while Replaying it would fail a good replay. _logger.LogDebug(cause, "GalaxyDriver {InstanceId} ignored a stream fault from a superseded event pump.", _driverInstanceId); return; } } + Interlocked.Increment(ref _escalatedStreamFaults); var supervisor = _supervisor; if (supervisor is null) diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/ProbeInitialValueTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/ProbeInitialValueTests.cs index 7eabd47ab..e496d7fae 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/ProbeInitialValueTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/ProbeInitialValueTests.cs @@ -116,6 +116,12 @@ public sealed class ProbeInitialValueTests } } + /// Fault the currently attached stream (the gateway dropping it). + public void FaultAttachedStream(Exception cause) + { + lock (_gate) _attached.ShouldNotBeNull("a stream must be attached").Writer.TryComplete(cause); + } + public string[] CallsSnapshot() { lock (_gate) return [.. Calls]; @@ -349,6 +355,45 @@ public sealed class ProbeInitialValueTests "a replay must not report success over a dead event stream"); } + /// + /// #865 review: a stream fault is escalated only from the LIVE pump. A fault from a pump superseded + /// by a reopen (its session is no longer current) is ignored — it would fail a good replay — while + /// the live pump's faults (before the reopen, and the replay's new pump after it) still escalate. + /// + [Fact] + public async Task Only_the_live_pumps_stream_faults_are_escalated() + { + var sub = new GatewayLikeSubscriber(); + await using var driver = new GalaxyDriver("drv", Opts(), hierarchySource: null, subscriber: sub) + { + RecreateOverrideForTests = _ => Task.CompletedTask, + }; + await driver.InitializeAsync("{}", CancellationToken.None); + await driver.SubscribeAsync(["Tank01.Level"], TimeSpan.Zero, CancellationToken.None); // pump 1 + + // Positive control: the live pump's fault escalates. + sub.FaultAttachedStream(new IOException("drop 1")); + await TestWait.UntilAsync(() => driver.EscalatedStreamFaultsForTests == 1, "the live pump's fault must escalate"); + + // Recover onto pump 2 (the replay restarts the faulted pump), then reopen: pump 2 is now superseded. + await driver.InvokeReplayForTestAsync(CancellationToken.None); + var pump2 = driver.EventPumpInstanceForTests.ShouldNotBeNull(); + await driver.InvokeReopenForTestAsync(CancellationToken.None); + sub.FaultAttachedStream(new IOException("late fault from the superseded pump")); + await TestWait.UntilAsync(() => ((EventPump)pump2).HasEnded, "the superseded pump's stream must have faulted"); + + // Positive control after the absence: the replay's new pump faults and that DOES escalate — so the + // count can only be 2 if the superseded pump's fault was ignored. + await driver.InvokeReplayForTestAsync(CancellationToken.None); + var pump3 = (EventPump)driver.EventPumpInstanceForTests.ShouldNotBeNull(); + pump3.ShouldNotBeSameAs(pump2); + sub.FaultAttachedStream(new IOException("drop 3")); + // HasEnded is set only after the fault callback ran, so once both pumps have ended every escalation + // that will ever happen has happened. + await TestWait.UntilAsync(() => pump3.HasEnded, "the new live pump's stream must have faulted"); + driver.EscalatedStreamFaultsForTests.ShouldBe(2, "the new live pump's fault escalates; the superseded pump's did not"); + } + /// /// A probe sync on a driver whose pump was never started (no platform source, e.g. the /// DiscoverAsync path) still starts it before advising — the watcher's pre-subscribe hook, not just