From 9060ff55617530747703f611f98f52a9751e8c10 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Wed, 11 Nov 2026 12:46:15 -0500 Subject: [PATCH] fix(galaxy): #865 review round 2 - wait for the replay's pump, never race a second stream Round 1 replaced a stale-session pump from any advise path and disposed the old one in the background; that could start a second stream while the old one still held the gateway's single-subscriber slot, and a kept current-session pump could be dead with no fault re-reported. - EnsureEventPumpStarted reuses again; only ReplayAsync restarts the pump. RestartEventPumpForReplay keeps a current-session pump only while its loop is running (EventPump.HasEnded), and flags the restart so no advise path starts a competing pump in the dispose gap. - AwaitCurrentEventStreamAsync (bounded by StreamRequestWait) starts a pump only when none exists, WAITS for the replay while the pump is stale or restarting, waits for StreamRequested, and returns the session it verified. SubscribeAsync (per #861 attempt), read-via-subscribe and the probe hook issue and fence the advise under that session, so a reconnect after the check is caught as a straddle and retried. - Read: a deadline (not caller) cancel during the wait falls through to per-ref results. - Tests: reopen-window probe test now asserts the advise waits for the replay's single new stream; first Subscribe/Read wait for the stream; the fencing tests that advise in the window use a short wait bound; the production-path seed test uses an empty galaxy (a real probe advise now starts a stream against the fake endpoint). Claude-Session: https://claude.ai/code/session_01GT67AD99zQQqpLCVm3p7xk --- .../GalaxyDriver.cs | 210 +++++++++++------- .../Health/PerPlatformProbeWatcher.cs | 17 +- .../Runtime/EventPump.cs | 3 + .../Runtime/EarlyEventWindowTests.cs | 4 + .../Runtime/GalaxyDriverFixRound1Tests.cs | 5 +- .../Runtime/ProbeInitialValueTests.cs | 64 ++++-- 6 files changed, 199 insertions(+), 104 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 97a2d32cb..7dbb6506d 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,7 @@ 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 AwaitStreamRequestedAsync(replayPump, cancellationToken).ConfigureAwait(false); + if (replayPump is not null) await AwaitCurrentEventStreamAsync(cancellationToken).ConfigureAwait(false); // 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). @@ -1312,31 +1312,45 @@ public sealed class GalaxyDriver { old = _eventPump; if (old is null) return; // pump never started — nothing to restart - // #865: a pump created after the reopen's fence (an advise in the reopen → replay window - // started it via EnsureEventPumpStarted) already streams the recreated session, and advises - // registered under it may be waiting for their initial values on it — keep it. - if (old.Session == _subscriptions.CurrentSession) return; + // #865: a pump first started after the reopen's fence (no pump existed before the reconnect) + // already streams the recreated session, and advises registered under it may be waiting for + // their initial values on it — keep it, unless its loop has ended (a dead pump is always replaced, + // #859 P-1: a stream fault during Replaying is not re-reported by the supervisor). + if (old.Session == _subscriptions.CurrentSession && !old.HasEnded) return; _eventPump = null; + // #865: until the replacement exists, AwaitCurrentEventStreamAsync must not start one of its + // own — two streams would race for the gateway's single-subscriber slot while the old one detaches. + _pumpRestarting = true; } - // Detach + dispose the faulted pump outside the lock so a slow shutdown doesn't - // block a concurrent EnsureEventPumpStarted; the old loop already exited on fault. - old.OnDataChange -= OnPumpDataChange; - try { old.DisposeAsync().AsTask().GetAwaiter().GetResult(); } - catch (Exception ex) + try { - _logger.LogWarning(ex, - "GalaxyDriver {InstanceId} faulted EventPump dispose failed during replay — continuing.", - _driverInstanceId); - } + // Detach + dispose the faulted pump outside the lock so a slow shutdown doesn't + // block a concurrent EnsureEventPumpStarted; the old loop already exited on fault. + old.OnDataChange -= OnPumpDataChange; + try { old.DisposeAsync().AsTask().GetAwaiter().GetResult(); } + catch (Exception ex) + { + _logger.LogWarning(ex, + "GalaxyDriver {InstanceId} faulted EventPump dispose failed during replay — continuing.", + _driverInstanceId); + } - // EnsureEventPumpStarted creates + starts a fresh pump under the lock. Skip the - // recreate if the driver is being disposed or its runtime torn down — the teardown - // already released the pump and a fresh one here would leak past it. - if (_disposed || _runtimeStopping) return; - EnsureEventPumpStarted(); + // EnsureEventPumpStarted creates + starts a fresh pump under the lock. Skip the + // recreate if the driver is being disposed or its runtime torn down — the teardown + // already released the pump and a fresh one here would leak past it. + if (_disposed || _runtimeStopping) return; + EnsureEventPumpStarted(); + } + finally + { + lock (_pumpLock) _pumpRestarting = false; + } } + /// True while is between dropping the old pump and starting its replacement. + private bool _pumpRestarting; + private void OnSupervisorStateChanged(object? sender, StateTransition transition) { // Reflect supervisor state in DriverHealth + transport forwarder. @@ -1689,10 +1703,19 @@ public sealed class GalaxyDriver var readSubscriber = _subscriber!; // A live pump must exist (values arrive through it); the read listens on ReadDataChange, not // on this instance, so it survives a reconnect's pump restart. - var readPump = EnsureEventPumpStarted(); - // #865: a read waits ONLY for the initial value, which reaches only an already-requested stream. - // Not the caller's token: a cancel here must still yield per-ref Bad snapshots, and the wait is bounded. - await AwaitStreamRequestedAsync(readPump, CancellationToken.None).ConfigureAwait(false); + EnsureEventPumpStarted(); + // #865: a read waits ONLY for the initial value, which reaches only a current-session stream that is + // already requested. The internal deadline bounds this too; a deadline (not caller) cancel falls + // through so the read still answers per-ref, as it does for a deadline during the RPC. + int issueSession; + try + { + issueSession = await AwaitCurrentEventStreamAsync(cancellationToken).ConfigureAwait(false); + } + catch (OperationCanceledException) when (!callerToken.IsCancellationRequested) + { + issueSession = _subscriptions.CurrentSession; + } var subscriptionId = _subscriptions.NextSubscriptionId(); // Pre-allocate one TaskCompletionSource per full-reference so the OnDataChange @@ -1729,7 +1752,7 @@ public sealed class GalaxyDriver // window before the RPC and close it right after registration (the finally below), not after the // value wait: a window held for the whole wait would park every unrelated unknown-handle event. var earlyEventWindow = _subscriptions.BeginPendingSubscribe(); - var session = _subscriptions.CurrentSession; // #861 — the session this SubscribeBulk is issued on + var session = issueSession; // #861/#865 — the session this SubscribeBulk is issued on (its stream verified) try { // The RPC gets only the CALLER's token: the internal deadline bounds our wait, not the RPC, so a slow @@ -1938,8 +1961,6 @@ public sealed class GalaxyDriver var generation = Volatile.Read(ref _runtimeGeneration); var subscriber = _subscriber; var pump = EnsureEventPumpStarted(); - // #865: the advise's initial value reaches only an already-requested stream (no replay). - await AwaitStreamRequestedAsync(pump, cancellationToken).ConfigureAwait(false); var subscriptionId = _subscriptions.NextSubscriptionId(); if (fullReferences.Count == 0) @@ -1972,6 +1993,10 @@ public sealed class GalaxyDriver for (var attempt = 1; ; attempt++) { + // #865: per attempt (a #861 retry is a new session): the advise's initial value reaches only a + // current-session stream that is already requested — the gateway does not replay a fresh stream. + var issueSession = await AwaitCurrentEventStreamAsync(cancellationToken).ConfigureAwait(false); + // #859: MXAccess's initial value for a new handle can reach the event pump before this // reply is registered (the gateway gives no reply-before-event ordering, and the reply and // the event ride separate RPCs). Open the early-event window BEFORE the RPC so such events @@ -1980,7 +2005,7 @@ public sealed class GalaxyDriver var earlyEventWindow = _subscriptions.BeginPendingSubscribe(); try { - var session = _subscriptions.CurrentSession; // #861 — the session this SubscribeBulk is issued on + var session = issueSession; // #861/#865 — the session this SubscribeBulk is issued on (its stream verified) var results = await subscriber .SubscribeBulkAsync(dialledRefs, bufferedIntervalMs, cancellationToken) .ConfigureAwait(false); @@ -2094,18 +2119,8 @@ public sealed class GalaxyDriver /// reverse map, not by spinning a pump per /// subscription. /// - /// - /// #865: a pump created for an OLDER registry session is replaced, not reused. Between - /// 's session swap and 's pump restart the - /// current pump still streams the dead gateway session; an advise made in that window goes to the - /// recreated session, so its initial value would reach no stream — and, registered under the - /// current session, the replay would not re-advise it either. A pump whose session is current - /// was created after the reopen's second fence, so its stream is on the recreated session. - /// private EventPump EnsureEventPumpStarted() { - EventPump? stale = null; - EventPump pump; lock (_pumpLock) { // Checked BEFORE the reuse: the flag is set before the teardown takes the pump under this @@ -2117,12 +2132,7 @@ public sealed class GalaxyDriver throw new InvalidOperationException( "GalaxyDriver runtime is shutting down — no subscription can be served until the next InitializeAsync."); } - if (_eventPump is not null && _eventPump.Session == _subscriptions.CurrentSession) return _eventPump; - if (_eventPump is not null) - { - stale = _eventPump; - stale.OnDataChange -= OnPumpDataChange; - } + if (_eventPump is not null) return _eventPump; _eventPump = new EventPump( _subscriber!, _subscriptions, _logger, channelCapacity: _options.MxAccess.EventPumpChannelCapacity, @@ -2131,27 +2141,13 @@ public sealed class GalaxyDriver onReplayGap: OnEventPumpReplayGap); _eventPump.OnDataChange += OnPumpDataChange; _eventPump.Start(); - pump = _eventPump; + return _eventPump; } - - if (stale is not null) DisposeStalePumpInBackground(stale); - return pump; } - /// Dispose a replaced pump off the caller's path (its stream is dead; the dispose only joins its loops). - private void DisposeStalePumpInBackground(EventPump stale) => _ = Task.Run(async () => - { - try { await stale.DisposeAsync().ConfigureAwait(false); } - catch (Exception ex) - { - _logger.LogWarning(ex, "GalaxyDriver {InstanceId} stale-session EventPump dispose failed.", _driverInstanceId); - } - }); - /// - /// for infrastructure callers (the probe watcher) that - /// must not fail because the runtime is going away: null — and no pump — while a teardown is in - /// progress or there is no subscriber. + /// for infrastructure callers that must not fail because the + /// runtime is going away: null — and no pump — while a teardown is in progress or there is no subscriber. /// /// The running (or just started) pump; null when none may be started. private EventPump? TryEnsureEventPumpStarted() @@ -2172,43 +2168,95 @@ public sealed class GalaxyDriver } } - /// Upper bound on how long an advise waits for the pump's StreamEvents call to be started. + /// Upper bound on how long an advise waits for a current-session event stream (#865). internal TimeSpan StreamRequestWait { get; set; } = TimeSpan.FromSeconds(5); /// - /// #865 — wait, bounded, until has started its StreamEvents call, so an - /// advise made next has a stream for its initial value. A fresh stream (after_worker_sequence = 0) - /// is never replayed, so an initial value pushed before it attaches is lost — and a slow-changing - /// attribute (ScanState, a setpoint) may not publish again for a long time. On timeout the caller - /// proceeds (logged): a late stream is no worse than the pre-#865 behaviour. + /// #865 — make sure an advise about to be issued has an event stream for its initial value, and return + /// the registry session to issue (and register) it under. /// - private async Task AwaitStreamRequestedAsync(EventPump pump, CancellationToken cancellationToken) + /// + /// + /// A fresh StreamEvents (after_worker_sequence = 0) is never replayed, so an initial value pushed before + /// the stream attaches is lost — and a slow-changing attribute (ScanState, a setpoint) may not publish + /// again for a long time. So: start the pump if there is none, and wait until its stream call has started + /// ( — this narrows, it cannot close, the attach race). + /// + /// + /// A pump bound to an OLDER session belongs to a gateway session a reconnect replaced (the reopen → + /// replay window): an advise now would go to the recreated session with no stream for it, and — + /// registered under the current session — the replay would not re-advise it. The pump is not replaced + /// here (only restarts it, so two streams never race for the single-subscriber + /// slot); the advise WAITS for the replay's pump instead. The returned session is the one the pump was + /// verified against: a reconnect after this check makes the caller's registration fail as a session + /// straddle (#861), whose retry comes back through here. + /// + /// + /// Bounded by ; on timeout the caller proceeds with the current session + /// (logged) — no worse than the pre-#865 behaviour. + /// + /// + /// Cancels the wait. + /// The registry session to issue the advise under. + private async Task AwaitCurrentEventStreamAsync(CancellationToken cancellationToken) { - try - { - await pump.StreamRequested.WaitAsync(StreamRequestWait, cancellationToken).ConfigureAwait(false); - } - catch (TimeoutException) + var elapsed = System.Diagnostics.Stopwatch.StartNew(); + while (true) { + var session = _subscriptions.CurrentSession; + EventPump? pump; + bool restarting; + lock (_pumpLock) + { + restarting = _pumpRestarting; + pump = restarting ? null : TryEnsureEventPumpStarted(); + } + if (pump is null && !restarting) return session; // tearing down — the caller's own fences decide + + var remaining = StreamRequestWait - elapsed.Elapsed; + if (pump is not null && pump.Session == session) + { + if (remaining > TimeSpan.Zero) + { + try + { + await pump.StreamRequested.WaitAsync(remaining, cancellationToken).ConfigureAwait(false); + return session; + } + catch (TimeoutException) + { + // fall through to the warning + } + } + else if (pump.StreamRequested.IsCompleted) + { + return session; + } + } + else if (remaining > TimeSpan.Zero) + { + // Reopen → replay window: the replay restarts the pump for the current session. + await Task.Delay(TimeSpan.FromMilliseconds(20), cancellationToken).ConfigureAwait(false); + continue; + } + _logger.LogWarning( - "GalaxyDriver {InstanceId} event stream was not started within {Wait} — advising anyway; initial values may be missed.", + "GalaxyDriver {InstanceId} had no started event stream for the current gateway session within {Wait} — advising anyway; initial values may be missed.", _driverInstanceId, StreamRequestWait); + return _subscriptions.CurrentSession; } } /// - /// #865 — the probe watcher's pre-subscribe hook: start the pump (if this runtime may) and wait for - /// its stream (). A probe's initial ScanState value is usually - /// its only event, and a driver with no authored tags never Reads/Subscribes, so without this its - /// pump — and every host status — never existed. + /// #865 — the probe watcher's pre-subscribe hook (). A probe's + /// initial ScanState value is usually its only event, and a driver with no authored tags never + /// Reads/Subscribes, so without this its pump — and every host status — never existed. /// - private async Task EnsureProbeEventStreamAsync(CancellationToken cancellationToken) + private Task EnsureProbeEventStreamAsync(CancellationToken cancellationToken) { // A seed pass a teardown already cancelled must not start a pump after the teardown finished. cancellationToken.ThrowIfCancellationRequested(); - var pump = TryEnsureEventPumpStarted(); - if (pump is null) return; - await AwaitStreamRequestedAsync(pump, cancellationToken).ConfigureAwait(false); + return AwaitCurrentEventStreamAsync(cancellationToken); } /// diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Health/PerPlatformProbeWatcher.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Health/PerPlatformProbeWatcher.cs index 08d2dd395..fc84e59b1 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Health/PerPlatformProbeWatcher.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Health/PerPlatformProbeWatcher.cs @@ -47,7 +47,7 @@ public sealed class PerPlatformProbeWatcher : IDisposable // #865: invoked before every probe SubscribeBulk so the driver's event stream is attached when the // gateway pushes the initial ScanState value (a fresh StreamEvents does not replay earlier events, // and ScanState rarely changes — a lost initial value leaves the host missing indefinitely). - private readonly Func? _ensureEventStream; + private readonly Func>? _ensureEventStream; // Tracked platform → gw item handle. Item handle 0 means the gw rejected the subscribe; // we keep the entry so SyncPlatformsAsync doesn't try to subscribe it again on every call. @@ -86,8 +86,9 @@ public sealed class PerPlatformProbeWatcher : IDisposable /// Buffered update interval in milliseconds; must be >= 0. /// /// Optional: awaited before every probe SubscribeBulk to make sure the event stream that carries - /// the initial ScanState value is running and requested (the driver starts its - /// and waits for its StreamEvents call to be issued). + /// the initial ScanState value is running and requested for the current gateway session (the + /// driver starts its and waits for its StreamEvents call). Returns the + /// registry session the stream was verified against; the advise is issued and fenced under it. /// internal PerPlatformProbeWatcher( IGalaxySubscriber subscriber, @@ -95,7 +96,7 @@ public sealed class PerPlatformProbeWatcher : IDisposable SubscriptionRegistry? registry, ILogger? logger = null, int bufferedUpdateIntervalMs = 0, - Func? ensureEventStream = null) + Func>? ensureEventStream = null) { _subscriber = subscriber ?? throw new ArgumentNullException(nameof(subscriber)); _aggregator = aggregator ?? throw new ArgumentNullException(nameof(aggregator)); @@ -210,12 +211,16 @@ public sealed class PerPlatformProbeWatcher : IDisposable // #865: the stream must be live BEFORE the advise — the initial value is the probe's only // event until the platform's scan state changes, and an unattached stream never sees it. - if (_ensureEventStream is not null) await _ensureEventStream(cancellationToken).ConfigureAwait(false); + // The returned session is the one the stream was verified for: a reconnect after the check makes + // the session test below refuse the reply as a straddle, and the retry comes back through here. + int? streamSession = _ensureEventStream is null + ? null + : await _ensureEventStream(cancellationToken).ConfigureAwait(false); // #859: the initial ScanState value can overtake the SubscribeBulk reply; park it until // PublishBindingsLocked registers the handles below (window closes at method exit). using var earlyEventWindow = _registry?.BeginPendingSubscribe(); - var session = CurrentSession; // the session this SubscribeBulk is issued on + var session = streamSession ?? CurrentSession; // the session this SubscribeBulk is issued on // PR 6.3 — use the configured bufferedUpdateIntervalMs (defaults to 0 = gw cadence // when the driver hasn't overridden MxAccess.PublishingIntervalMs). Probe ScanState // changes are rare so a coarser interval is usually fine; deployments that need 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 e56d8d57a..dc8905f85 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 @@ -154,6 +154,9 @@ 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 }; + private readonly TaskCompletionSource _streamRequested = new(TaskCreationOptions.RunContinuationsAsynchronously); /// diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/EarlyEventWindowTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/EarlyEventWindowTests.cs index 1d0ade328..474841437 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/EarlyEventWindowTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/EarlyEventWindowTests.cs @@ -844,6 +844,7 @@ public sealed class EarlyEventWindowTests "g", Opts(), hierarchySource: null, dataReader: null, dataWriter: null, subscriber: subscriber) { RecreateOverrideForTests = _ => Task.CompletedTask, + StreamRequestWait = TimeSpan.FromMilliseconds(100), // advises in the reopen window on purpose (#865 bound) }; var delivered = new ConcurrentQueue<(string Ref, double Value)>(); driver.OnDataChange += (_, e) => delivered.Enqueue((e.FullReference, (double)e.Snapshot.Value!)); @@ -1027,6 +1028,9 @@ public sealed class EarlyEventWindowTests "g", Opts(), hierarchySource: null, dataReader: null, dataWriter: null, subscriber: subscriber) { RecreateOverrideForTests = _ => Task.CompletedTask, + // These tests advise in the reopen → replay window on purpose (fencing, not values). #865 makes + // such an advise wait for the replay's pump, bounded; keep the bound short so they don't idle. + StreamRequestWait = TimeSpan.FromMilliseconds(100), }; await driver.InitializeAsync("{}", TestContext.Current.CancellationToken); return driver; diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/GalaxyDriverFixRound1Tests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/GalaxyDriverFixRound1Tests.cs index e2339a751..11c5a095a 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/GalaxyDriverFixRound1Tests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/GalaxyDriverFixRound1Tests.cs @@ -272,7 +272,10 @@ public sealed class GalaxyDriverFixRound1Tests [Fact] public async Task Production_path_queries_the_platform_source_at_init() { - var platforms = new ScriptedPlatformSource(); + // An empty galaxy: since #865 a probe advise starts the event stream first, and this owned runtime's + // gRPC channel points at no real gateway — its stream would fault and (correctly) degrade health. + // The query itself is what this test pins. + var platforms = new ScriptedPlatformSource { Script = _ => [] }; await using var driver = new GalaxyDriver("drv", Opts(), hierarchySource: null, platformSource: platforms) { SessionOpenOverrideForTests = _ => Task.CompletedTask, 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 81c986ab2..774aa762e 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 @@ -195,10 +195,10 @@ public sealed class ProbeInitialValueTests /// /// Review finding: a probe advised in the reopen → replay window (the session was swapped, the old - /// pump still streams the dead session) must still get its initial value — the stale-session pump - /// is replaced before the advise, and the replay keeps the replacement (it already streams the - /// recreated session). Reusing the old pump bound the value to the old session, so it never routed, - /// and the replay skipped the platform (its handle is current-session). + /// pump still streams the dead session) must still get its initial value. The advise waits for the + /// replay's restarted pump instead of advising onto the stale one (its value would route under the + /// old session and be lost, and the replay skips a current-session handle) or starting a competing + /// stream (the gateway's default single-subscriber mode refuses a second attach). /// [Fact] public async Task Probe_advised_between_reopen_and_replay_still_gets_its_initial_value() @@ -218,23 +218,55 @@ public sealed class ProbeInitialValueTests await driver.InvokeReopenForTestAsync(CancellationToken.None); // session swapped; no replay yet sub.ScanStateRunning = false; - await driver.ProbeWatcherForTests.ShouldNotBeNull() + var callsBeforeWindow = sub.CallsSnapshot().Length; + // The window advise must WAIT for the replay's current-session pump rather than advise onto the + // stale one (or start a competing stream). + var sync = driver.ProbeWatcherForTests.ShouldNotBeNull() .SyncPlatformsAsync([.. Platforms, "LatePlatform"], CancellationToken.None); - await TestWait.UntilAsync( - () => StateOf(driver, "LatePlatform") == HostState.Stopped, - "a probe advised in the reopen window must get its initial value", - () => Describe(driver, sub)); - var windowPump = driver.EventPumpInstanceForTests; - windowPump.ShouldNotBeSameAs(oldPump, "the stale-session pump must have been replaced before the advise"); - await driver.InvokeReplayForTestAsync(CancellationToken.None); - driver.EventPumpInstanceForTests.ShouldBeSameAs(windowPump, "the replay keeps a pump already on the current session"); + await TestWait.ForAsync(sync, "the window advise completes once the replay restarted the pump"); + await TestWait.UntilAsync( - () => Platforms.All(p => StateOf(driver, p) == HostState.Stopped), - "the replay re-advises the stale probes and their initial values arrive on the kept pump", + () => StateOf(driver, "LatePlatform") == HostState.Stopped + && Platforms.All(p => StateOf(driver, p) == HostState.Stopped), + "the window advise and the replay's re-advises all get their initial values", () => Describe(driver, sub)); - StateOf(driver, "LatePlatform").ShouldBe(HostState.Stopped); + driver.EventPumpInstanceForTests.ShouldNotBeSameAs(oldPump, "the replay restarted the stale pump"); + var windowCalls = sub.CallsSnapshot()[callsBeforeWindow..]; + var lateAdvise = Array.FindIndex(windowCalls, c => c.Contains("LatePlatform", StringComparison.Ordinal)); + var newStream = Array.IndexOf(windowCalls, "stream"); + newStream.ShouldBeGreaterThanOrEqualTo(0, $"the replay must start a new stream; calls=[{string.Join(" | ", windowCalls)}]"); + lateAdvise.ShouldBeGreaterThan(newStream, $"the window advise must follow the new stream; calls=[{string.Join(" | ", windowCalls)}]"); + windowCalls.Count(c => c == "stream").ShouldBe(1, "no competing stream may be started for the window advise"); + sub.DroppedInitialValues.ShouldBe(0); + } + + /// + /// The data paths too: a first SubscribeAsync and a first ReadAsync on a driver whose pump was never + /// started request the stream BEFORE advising, so their initial values arrive (a read of a value that + /// never changes again would otherwise time out). + /// + [Fact] + public async Task First_subscribe_and_read_advise_only_after_the_stream_is_requested() + { + var sub = new GatewayLikeSubscriber(); + await using var driver = new GalaxyDriver("drv", Opts(), hierarchySource: null, subscriber: sub) + { + ReadViaSubscribeDeadline = TimeSpan.FromSeconds(10), + }; + await driver.InitializeAsync("{}", CancellationToken.None); + var delivered = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + driver.OnDataChange += (_, e) => delivered.TrySetResult(e); + + await driver.SubscribeAsync(["Tank01.Level"], TimeSpan.Zero, CancellationToken.None); + var calls = sub.CallsSnapshot(); + Array.IndexOf(calls, "stream").ShouldBe(0, $"the stream must be requested before the advise; calls=[{string.Join(" | ", calls)}]"); + (await TestWait.ForAsync(delivered.Task, "the subscribe's initial value must arrive", () => Describe(driver, sub))) + .FullReference.ShouldBe("Tank01.Level"); + + var read = await driver.ReadAsync(["Tank02.Level"], CancellationToken.None); + read[0].StatusCode.ShouldBe(0u, "the read's initial value must arrive (not BadTimeout)"); sub.DroppedInitialValues.ShouldBe(0); }