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 3102724f1..97a2d32cb 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/GalaxyDriver.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/GalaxyDriver.cs @@ -1053,6 +1053,10 @@ public sealed class GalaxyDriver // subscriptions to replay the dead pump used to stay in _eventPump, so every later // Subscribe/Read reused it and was silently deaf while health stayed Healthy. RestartEventPumpForReplay(); + // #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); // 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). @@ -1308,6 +1312,10 @@ 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; _eventPump = null; } @@ -1681,7 +1689,10 @@ 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. - EnsureEventPumpStarted(); + 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); var subscriptionId = _subscriptions.NextSubscriptionId(); // Pre-allocate one TaskCompletionSource per full-reference so the OnDataChange @@ -1927,6 +1938,8 @@ 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) @@ -2081,8 +2094,18 @@ 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 @@ -2094,7 +2117,12 @@ 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) return _eventPump; + if (_eventPump is not null && _eventPump.Session == _subscriptions.CurrentSession) return _eventPump; + if (_eventPump is not null) + { + stale = _eventPump; + stale.OnDataChange -= OnPumpDataChange; + } _eventPump = new EventPump( _subscriber!, _subscriptions, _logger, channelCapacity: _options.MxAccess.EventPumpChannelCapacity, @@ -2103,13 +2131,26 @@ public sealed class GalaxyDriver onReplayGap: OnEventPumpReplayGap); _eventPump.OnDataChange += OnPumpDataChange; _eventPump.Start(); - return _eventPump; + pump = _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: false — and no pump — while a teardown is in + /// 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. @@ -2118,36 +2159,58 @@ public sealed class GalaxyDriver lock (_pumpLock) { if (_runtimeStopping || _disposed || _subscriber is null) return null; - return EnsureEventPumpStarted(); // re-enters _pumpLock (Lock is reentrant); cannot throw here + try + { + return EnsureEventPumpStarted(); // re-enters _pumpLock (Lock is reentrant) + } + catch (InvalidOperationException) + { + // _runtimeStopping is written without _pumpLock; a teardown that flipped it between the + // check above and EnsureEventPumpStarted's own check is the same "no pump" answer. + return null; + } } } - /// Upper bound on how long a probe subscribe waits for the pump's StreamEvents call to be issued. - internal TimeSpan ProbeStreamRequestWait { get; set; } = TimeSpan.FromSeconds(5); + /// Upper bound on how long an advise waits for the pump's StreamEvents call to be started. + internal TimeSpan StreamRequestWait { get; set; } = TimeSpan.FromSeconds(5); /// - /// #865 — the probe watcher's pre-subscribe hook: start the pump (if this runtime may) and wait, - /// bounded, until its StreamEvents call has been issued, so the gateway has a subscriber to push - /// the probe's initial ScanState value to. A fresh stream (after_worker_sequence = 0) is never - /// replayed, so an initial value pushed before it attaches is lost — and ScanState rarely changes - /// again. On timeout the subscribe proceeds anyway (logged): a late stream is no worse than before. + /// #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. /// - private async Task EnsureProbeEventStreamAsync(CancellationToken cancellationToken) + private async Task AwaitStreamRequestedAsync(EventPump pump, CancellationToken cancellationToken) { - var pump = TryEnsureEventPumpStarted(); - if (pump is null) return; try { - await pump.StreamRequested.WaitAsync(ProbeStreamRequestWait, cancellationToken).ConfigureAwait(false); + await pump.StreamRequested.WaitAsync(StreamRequestWait, cancellationToken).ConfigureAwait(false); } catch (TimeoutException) { _logger.LogWarning( - "GalaxyDriver {InstanceId} event stream was not requested within {Wait} — subscribing the platform probes anyway; their initial ScanState may be missed.", - _driverInstanceId, ProbeStreamRequestWait); + "GalaxyDriver {InstanceId} event stream was not started within {Wait} — advising anyway; initial values may be missed.", + _driverInstanceId, StreamRequestWait); } } + /// + /// #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. + /// + private async 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); + } + /// /// Replay-gap callback for the . The gateway told us it evicted /// events we can never receive, so every cached value may be stale by an unknown amount. 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 c3640ab12..e56d8d57a 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 @@ -148,15 +148,27 @@ internal sealed class EventPump : IAsyncDisposable private readonly int _session; + /// + /// The registry session this pump was created for (#865). A pump of an older session was opened + /// against a gateway session a reconnect replaced; its stream will never carry the new session's events. + /// + internal int Session => _session; + private readonly TaskCompletionSource _streamRequested = new(TaskCreationOptions.RunContinuationsAsynchronously); /// - /// Completes once the stream loop has issued its StreamEvents call (the first - /// MoveNextAsync runs the subscriber's iterator synchronously up to the RPC) — or once the - /// loop has ended without getting that far, so a waiter never hangs on a dead pump. #865: the - /// gateway does not replay a fresh stream, so a caller whose only expected event is the initial - /// value of an advise it is about to make waits on this first. + /// Completes once the stream loop has STARTED its StreamEvents call (the first + /// MoveNextAsync runs the subscriber's iterator synchronously up to the gRPC call creation) — + /// or once the loop has ended without getting that far, so a waiter never hangs on a dead pump. + /// #865: the gateway does not replay a fresh stream, so a caller whose only expected event is the + /// initial value of an advise it is about to make waits on this first. /// + /// + /// ⚠️ This NARROWS the stream-attach race, it does not close it: the HTTP/2 request is sent + /// asynchronously and the gateway attaches the subscriber after auth, concurrently with any other + /// RPC. In practice the attach wins (an advise's initial value needs a worker IPC round trip plus an + /// MXAccess advise), but the gateway offers no attach acknowledgement to await. + /// public Task StreamRequested => _streamRequested.Task; /// diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/EventPumpStreamRequestedTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/EventPumpStreamRequestedTests.cs new file mode 100644 index 000000000..d1e28e1d6 --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/EventPumpStreamRequestedTests.cs @@ -0,0 +1,62 @@ +using System.Runtime.CompilerServices; +using Shouldly; +using Xunit; +using ZB.MOM.WW.MxGateway.Contracts.Proto; +using ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Runtime; + +namespace ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests.Runtime; + +/// +/// #865: — what advise paths wait on so an initial value is +/// never pushed before the stream is requested. It must complete once the RPC is started (before any +/// event), and must never strand a waiter on a pump whose loop died before getting that far. +/// +public sealed class EventPumpStreamRequestedTests +{ + private sealed class ScriptedSubscriber(Func> stream) : IGalaxySubscriber + { + public Task> SubscribeBulkAsync( + IReadOnlyList fullReferences, int bufferedUpdateIntervalMs, CancellationToken cancellationToken) + => Task.FromResult>([]); + + public Task UnsubscribeBulkAsync(IReadOnlyList itemHandles, CancellationToken cancellationToken) => Task.CompletedTask; + + public IAsyncEnumerable StreamEventsAsync(CancellationToken cancellationToken) => stream(cancellationToken); + } + + /// A stream that is requested and then never yields: StreamRequested completes; the RPC started first. + [Fact] + public async Task Completes_once_the_stream_is_requested_before_any_event() + { + var rpcStarted = 0; + async IAsyncEnumerable Never([EnumeratorCancellation] CancellationToken ct) + { + Interlocked.Exchange(ref rpcStarted, 1); // the synchronous first step — where a real call is created + await Task.Delay(Timeout.Infinite, ct).ConfigureAwait(false); + yield break; + } + + await using var pump = new EventPump(new ScriptedSubscriber(Never), new SubscriptionRegistry()); + pump.StreamRequested.IsCompleted.ShouldBeFalse("not before Start"); + pump.Start(); + + await TestWait.ForAsync(pump.StreamRequested, "StreamRequested must complete once the stream call is started"); + Volatile.Read(ref rpcStarted).ShouldBe(1, "StreamRequested must not precede the stream call"); + } + + /// A subscriber that throws before reaching the RPC still completes StreamRequested (no stranded waiter). + [Fact] + public async Task Completes_when_the_stream_fails_before_the_rpc() + { + var faulted = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + await using var pump = new EventPump( + new ScriptedSubscriber(_ => throw new InvalidOperationException("not connected")), + new SubscriptionRegistry(), + onStreamFault: ex => faulted.TrySetResult(ex)); + pump.Start(); + + await TestWait.ForAsync(pump.StreamRequested, "a dead pump must not strand a StreamRequested waiter"); + (await TestWait.ForAsync(faulted.Task, "positive control: the failure is still reported as a stream fault")) + .ShouldBeOfType(); + } +} 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 2ca4698a7..e2339a751 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 @@ -333,7 +333,10 @@ public sealed class GalaxyDriverFixRound1Tests await driver.InitializeAsync("{}", CancellationToken.None); // #865: the new runtime's background probe seed starts a pump of its own (its advises need a live // stream) — but it must be a FRESH one, never the dead subscriber's. - driver.EventPumpInstanceForTests.ShouldNotBeSameAs(oldPump, "the next runtime must not inherit the old pump"); + await WaitUntilAsync(() => driver.ProbeWatcherForTests?.WatchedPlatforms.Count == 1, + "positive control: the new runtime's background seed must advise its platform"); + driver.EventPumpInstanceForTests.ShouldNotBeNull("the seed's probe advise must have started a pump") + .ShouldNotBeSameAs(oldPump, "the next runtime must not inherit the old pump"); } // ───────────────────────────── IMPORTANT 3: bounded gateway close ───────────────────────────── 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 2144c8f7f..81c986ab2 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 @@ -193,6 +193,51 @@ public sealed class ProbeInitialValueTests sub.DroppedInitialValues.ShouldBe(0); } + /// + /// 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). + /// + [Fact] + public async Task Probe_advised_between_reopen_and_replay_still_gets_its_initial_value() + { + var sub = new GatewayLikeSubscriber(); + await using var driver = new GalaxyDriver("drv", Opts(), hierarchySource: null, subscriber: sub, + platformSource: new FixedPlatformSource(Platforms)) + { + RecreateOverrideForTests = _ => Task.CompletedTask, + }; + await driver.InitializeAsync("{}", CancellationToken.None); + await TestWait.UntilAsync( + () => Platforms.All(p => StateOf(driver, p) == HostState.Running), + "positive control: the initial seed reports Running", + () => Describe(driver, sub)); + var oldPump = driver.EventPumpInstanceForTests; + + await driver.InvokeReopenForTestAsync(CancellationToken.None); // session swapped; no replay yet + sub.ScanStateRunning = false; + await 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.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", + () => Describe(driver, sub)); + StateOf(driver, "LatePlatform").ShouldBe(HostState.Stopped); + sub.DroppedInitialValues.ShouldBe(0); + } + /// /// 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