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 c9dae1420..8800aa65f 100644
--- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/GalaxyDriver.cs
+++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/GalaxyDriver.cs
@@ -1056,7 +1056,19 @@ public sealed class GalaxyDriver
// #865: every re-advise below pushes its initial value once, to whatever stream is attached.
EventPump? replayPump;
lock (_pumpLock) replayPump = _eventPump;
- if (replayPump is not null) await AwaitCurrentEventStreamAsync(cancellationToken).ConfigureAwait(false);
+ if (replayPump is not null)
+ {
+ await AwaitCurrentEventStreamAsync(cancellationToken).ConfigureAwait(false);
+ // A replay whose fresh pump already died must FAIL, so the supervisor backs off and stays
+ // Degraded — succeeding would go Healthy with a dead stream, and the next advise's re-report
+ // would start a new cycle at attempt 1 (no backoff): a tight reconnect loop.
+ lock (_pumpLock) replayPump = _eventPump;
+ if (replayPump is { HasEnded: true })
+ {
+ throw new InvalidOperationException(
+ "The Galaxy event stream ended during the reconnect replay — retrying after backoff.");
+ }
+ }
// Only subscriptions still holding the previous session's handles: one registered after the
// reopen is already live on the recreated session (re-subscribing it would double-advise).
@@ -2279,7 +2291,7 @@ public sealed class GalaxyDriver
}
///
- /// True when a stale pump will be replaced shortly: the supervisor is reopening/replaying, or there is
+ /// True when a stale pump will be replaced shortly: a reopen/replay attempt is running, or there is
/// no supervisor (the injected-seam path, where the replay is driven directly), or a restart is underway.
///
private bool ReplayExpectedSoon()
@@ -2287,8 +2299,9 @@ public sealed class GalaxyDriver
if (Volatile.Read(ref _pumpRestarting)) return true;
var supervisor = _supervisor;
if (supervisor is null) return true;
- var state = supervisor.CurrentState;
- return state is ReconnectSupervisor.State.Reopening or ReconnectSupervisor.State.Replaying;
+ // Not the state: between attempts (the backoff delay, up to MaxBackoff) it still reads
+ // Reopening/Replaying while no replay is coming any time soon.
+ return supervisor.AttemptInProgress;
}
/// A current-session pump's loop ended while the supervisor was not listening (see AwaitCurrentEventStreamAsync).
diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/EventPump.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/EventPump.cs
index dc8905f85..cdb36a264 100644
--- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/EventPump.cs
+++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/EventPump.cs
@@ -155,7 +155,11 @@ internal sealed class EventPump : IAsyncDisposable
internal int Session => _session;
/// True once the stream loop has exited (faulted, completed by the gateway, or disposed).
- internal bool HasEnded => _loop is { IsCompleted: true };
+ internal bool HasEnded => Volatile.Read(ref _ended);
+
+ // Set at the very start of the stream loop's finally — before StreamRequested is completed there — so a
+ // waiter released by a dying loop always sees it ended (the loop task itself completes only later).
+ private bool _ended;
private readonly TaskCompletionSource _streamRequested = new(TaskCreationOptions.RunContinuationsAsynchronously);
@@ -212,7 +216,19 @@ internal sealed class EventPump : IAsyncDisposable
// Start the first MoveNext BEFORE awaiting it: that synchronous step issues the RPC, which is
// the moment StreamRequested reports (#865).
var next = enumerator.MoveNextAsync();
- _streamRequested.TrySetResult();
+ if (next.IsCompleted)
+ {
+ // Already decided (an item, an immediate end, or a fault): consume it now, so a stream that
+ // ended or faulted at once is seen as ended (finally) BEFORE StreamRequested releases anyone.
+ var hasFirst = await next.ConfigureAwait(false);
+ if (!hasFirst) Volatile.Write(ref _ended, true);
+ _streamRequested.TrySetResult();
+ next = new ValueTask(hasFirst);
+ }
+ else
+ {
+ _streamRequested.TrySetResult();
+ }
for (; await next.ConfigureAwait(false); next = enumerator.MoveNextAsync())
{
var ev = enumerator.Current;
@@ -274,6 +290,7 @@ internal sealed class EventPump : IAsyncDisposable
}
finally
{
+ Volatile.Write(ref _ended, true);
// A loop that died before issuing the RPC must not strand a StreamRequested waiter.
_streamRequested.TrySetResult();
if (enumerator is not 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 7b6c64101..41c1e5a52 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
@@ -104,6 +104,15 @@ public sealed class ReconnectSupervisor : IDisposable
get { lock (_stateLock) return _state; }
}
+ private bool _attemptInProgress;
+
+ ///
+ /// 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
+ /// for the replay's pump only while one is imminent.
+ ///
+ public bool AttemptInProgress => Volatile.Read(ref _attemptInProgress);
+
/// True when CurrentState != Healthy. Drivers map this to DriverState.Degraded.
public bool IsDegraded
{
@@ -136,6 +145,9 @@ public sealed class ReconnectSupervisor : IDisposable
lock (_stateLock)
{
+ // Re-checked under the lock (StopAsync sets it under the same lock): a report that passed the
+ // check above while a stop ran must not start a loop nothing would ever cancel.
+ ObjectDisposedException.ThrowIf(_disposed, this);
_lastError = cause.Message;
if (_state != State.Healthy)
{
@@ -208,13 +220,21 @@ public sealed class ReconnectSupervisor : IDisposable
// === Reopening phase ===
lock (_stateLock) TransitionLocked(State.Reopening, _lastError);
+ // In progress from the reopen through the replay (no gap between the phases); cleared on any
+ // exit from the attempt — failure (then the backoff), cancellation or success.
+ Volatile.Write(ref _attemptInProgress, true);
try
{
await _reopen(ct).ConfigureAwait(false);
}
- catch (OperationCanceledException) when (ct.IsCancellationRequested) { return; }
+ catch (OperationCanceledException) when (ct.IsCancellationRequested)
+ {
+ Volatile.Write(ref _attemptInProgress, false);
+ return;
+ }
catch (Exception ex)
{
+ Volatile.Write(ref _attemptInProgress, false);
_logger.LogWarning(ex, "Galaxy reopen failed (attempt {Attempt})", attempt);
lock (_stateLock) { _lastError = ex.Message; }
continue; // back to backoff + retry
@@ -234,6 +254,10 @@ public sealed class ReconnectSupervisor : IDisposable
lock (_stateLock) { _lastError = ex.Message; }
continue; // back to backoff + retry
}
+ finally
+ {
+ Volatile.Write(ref _attemptInProgress, false);
+ }
// === Done ===
lock (_stateLock)
@@ -282,12 +306,14 @@ public sealed class ReconnectSupervisor : IDisposable
/// A task that completes once the recovery loop has exited.
public async Task StopAsync()
{
- if (_disposed) return;
- _disposed = true;
-
CancellationTokenSource? cts;
Task? loop;
- lock (_stateLock) { cts = _loopCts; loop = _recoveryLoop; _loopCts = null; _recoveryLoop = null; }
+ lock (_stateLock)
+ {
+ if (_disposed) return;
+ _disposed = true;
+ cts = _loopCts; loop = _recoveryLoop; _loopCts = null; _recoveryLoop = null;
+ }
// The loop's own finally may have released the CTS already — a cancelled-then-disposed CTS
// throws on Cancel, and a disposed one is a no-op on Dispose.
diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/SubscriptionRegistry.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/SubscriptionRegistry.cs
index 95e6f83d9..5f4aaf82f 100644
--- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/SubscriptionRegistry.cs
+++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/SubscriptionRegistry.cs
@@ -503,17 +503,11 @@ internal sealed class SubscriptionRegistry
subscribers = ResolveSubscribers(ev.ItemHandle);
if (subscribers.Count > 0) return EarlyEventOutcome.Resolved;
AfterFastPathMissForTests?.Invoke();
- if (Volatile.Read(ref _pendingSubscribes) == 0)
- {
- // #865 (found by its stress run): the lookup above and this read are not atomic. A
- // subscribe can register the handle AND close its window in between — then "no window
- // open" does not mean "stale", and dropping here lost the initial value for good (the
- // first event of a bulk advise, while the dispatch thread was slow). The window closes
- // only after its registration, so once no window is open a re-lookup sees every
- // registration that window made; nothing is held, so no older event can be overtaken.
- subscribers = ResolveSubscribers(ev.ItemHandle);
- return subscribers.Count > 0 ? EarlyEventOutcome.Resolved : EarlyEventOutcome.Stale;
- }
+ // #865 (found by its stress run): a miss is NOT decided here. The lookup above and an
+ // open-window check are not atomic — a subscribe can register the handle AND close its window
+ // in between, and declaring the event stale then lost its initial value for good (the first
+ // event of a bulk advise, while the dispatch thread was slow). Misses are rare; the locked path
+ // below re-does the lookup, the session check and the window check atomically.
}
lock (_parkGate)
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 36f152888..7eabd47ab 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
@@ -51,6 +51,11 @@ public sealed class ProbeInitialValueTests
/// Initial values pushed while no stream was attached — lost, as on the real gateway.
public int DroppedInitialValues;
+ private int _streams;
+
+ /// Streams beyond this count end immediately after attaching (a gateway refusing/ending them).
+ public int StreamsThatStayOpen = int.MaxValue;
+
/// The ScanState value the next advise reports.
public volatile bool ScanStateRunning = true;
@@ -90,6 +95,11 @@ public sealed class ProbeInitialValueTests
Calls.Add("stream");
_attached = mine;
}
+ if (Interlocked.Increment(ref _streams) > StreamsThatStayOpen)
+ {
+ lock (_gate) { if (ReferenceEquals(_attached, mine)) _attached = null; }
+ yield break; // the gateway completed this stream at once
+ }
try
{
await foreach (var ev in mine.Reader.ReadAllAsync(cancellationToken).ConfigureAwait(false))
@@ -208,6 +218,9 @@ public sealed class ProbeInitialValueTests
platformSource: new FixedPlatformSource(Platforms))
{
RecreateOverrideForTests = _ => Task.CompletedTask,
+ // Generous: the window advise must still be waiting when the replay below starts the pump, even
+ // on a stalled agent; the wait ends the moment that pump exists.
+ StreamRequestWait = TimeSpan.FromSeconds(20),
};
await driver.InitializeAsync("{}", CancellationToken.None);
await TestWait.UntilAsync(
@@ -315,6 +328,27 @@ public sealed class ProbeInitialValueTests
sub.DroppedInitialValues.ShouldBe(0);
}
+ ///
+ /// A replay whose restarted pump's stream ends at once must FAIL (the supervisor then backs off and
+ /// stays Degraded) rather than succeed onto a dead stream — succeeding went Healthy, and the next
+ /// advise's re-report started a new cycle with no backoff: a tight reconnect loop. Positive control:
+ /// the first replay, whose stream stays open, succeeds.
+ ///
+ [Fact]
+ public async Task Replay_fails_when_the_restarted_stream_ends_at_once()
+ {
+ var sub = new GatewayLikeSubscriber { StreamsThatStayOpen = 2 };
+ await using var driver = new GalaxyDriver("drv", Opts(), hierarchySource: null, subscriber: sub);
+ await driver.InitializeAsync("{}", CancellationToken.None);
+ await driver.SubscribeAsync(["Tank01.Level"], TimeSpan.Zero, CancellationToken.None); // stream 1
+
+ await driver.InvokeReplayForTestAsync(CancellationToken.None); // stream 2 stays open — succeeds
+
+ await Should.ThrowAsync(
+ () => driver.InvokeReplayForTestAsync(CancellationToken.None), // stream 3 ends at once
+ "a replay must not report success over a dead event stream");
+ }
+
///
/// 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
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 accc819cf..b99f3b43c 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
@@ -15,6 +15,56 @@ public sealed class ReconnectSupervisorTests
new(InitialBackoffOverride: TimeSpan.FromMilliseconds(5),
MaxBackoffOverride: TimeSpan.FromMilliseconds(20));
+ ///
+ /// #865: AttemptInProgress is true while a reopen (through the replay) runs and FALSE during the backoff
+ /// between attempts — when the state still reads Reopening. Advises wait for a replay only while one is
+ /// actually running, not for a whole backoff.
+ ///
+ [Fact]
+ public async Task AttemptInProgress_is_true_during_an_attempt_and_false_during_the_backoff()
+ {
+ var reopenEntered = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+ var releaseReopen = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+ var reopens = 0;
+ using var sup = new ReconnectSupervisor(
+ reopen: async _ =>
+ {
+ if (Interlocked.Increment(ref reopens) == 1)
+ {
+ reopenEntered.TrySetResult();
+ await releaseReopen.Task;
+ throw new IOException("gateway still down");
+ }
+ },
+ replay: _ => Task.CompletedTask,
+ backoffDelay: (_, _, _) => TimeSpan.FromMinutes(10)); // park in the backoff
+
+ sup.AttemptInProgress.ShouldBeFalse("Healthy: no attempt");
+ sup.ReportTransportFailure(new IOException("drop"));
+ await TestWait.ForAsync(reopenEntered.Task, "the first reopen must start");
+ sup.AttemptInProgress.ShouldBeTrue("a reopen is running");
+
+ releaseReopen.SetResult(); // the reopen fails -> backoff of 10 min
+ await TestWait.UntilAsync(() => !sup.AttemptInProgress, "the failed attempt must clear the flag");
+ sup.CurrentState.ShouldBe(ReconnectSupervisor.State.Reopening, "the backoff still reads Reopening — why the flag exists");
+ }
+
+ /// #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()
+ {
+ var reopens = 0;
+ var sup = new ReconnectSupervisor(
+ reopen: _ => { Interlocked.Increment(ref reopens); return Task.CompletedTask; },
+ replay: _ => Task.CompletedTask,
+ options: FastBackoff());
+ await sup.StopAsync();
+
+ Should.Throw(() => sup.ReportTransportFailure(new IOException("late")));
+ sup.CurrentState.ShouldBe(ReconnectSupervisor.State.Healthy);
+ Volatile.Read(ref reopens).ShouldBe(0);
+ }
+
/// Verifies that the supervisor starts in a healthy state with no errors.
[Fact]
public void InitialState_IsHealthy()