From 59d33ab93805617305ebb5fcc838b64d350cf0c8 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Wed, 11 Nov 2026 13:20:49 -0500 Subject: [PATCH] test(galaxy): #865 review round 7 - pin the superseded-pump fault filter Adds an escalated-stream-fault test seam and a test: the live pump's fault escalates (positive controls before and after), a superseded pump's late fault after a reopen does not; the absence is asserted only once both pumps' callbacks are known to have run (HasEnded). Verified to fail with the filter disabled. The filter's comment now says the identity check does the work and warns about a future session bump while Healthy. Claude-Session: https://claude.ai/code/session_01GT67AD99zQQqpLCVm3p7xk --- .../GalaxyDriver.cs | 14 ++++-- .../Runtime/ProbeInitialValueTests.cs | 45 +++++++++++++++++++ 2 files changed, 56 insertions(+), 3 deletions(-) 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