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 7dbb6506d..c9dae1420 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/GalaxyDriver.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/GalaxyDriver.cs @@ -1703,7 +1703,7 @@ 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(); + ThrowIfRuntimeStopping(); // the pump itself is started by AwaitCurrentEventStreamAsync below // #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. @@ -1960,7 +1960,7 @@ public sealed class GalaxyDriver // generation, so the Register below can tell its SubscribeBulk straddled a teardown. var generation = Volatile.Read(ref _runtimeGeneration); var subscriber = _subscriber; - var pump = EnsureEventPumpStarted(); + ThrowIfRuntimeStopping(); // the pump is started per attempt by AwaitCurrentEventStreamAsync (#865) var subscriptionId = _subscriptions.NextSubscriptionId(); if (fullReferences.Count == 0) @@ -2014,7 +2014,6 @@ public sealed class GalaxyDriver switch (TryRegisterForGeneration(subscriptionId, bindings, generation, session)) { case RegisterOutcome.Registered: - _ = pump; // keep the pump alive for the subscription's lifetime return new GalaxySubscriptionHandle(subscriptionId); case RegisterOutcome.TornDown: @@ -2113,11 +2112,25 @@ public sealed class GalaxyDriver } } + /// A Subscribe/Read is refused while a teardown is in progress (the pump would belong to a dying subscriber). + private void ThrowIfRuntimeStopping() + { + lock (_pumpLock) + { + if (_runtimeStopping) + { + throw new InvalidOperationException( + "GalaxyDriver runtime is shutting down — no subscription can be served until the next InitializeAsync."); + } + } + } + /// - /// Lazily start the on the first subscribe. The pump is - /// shared across every subscription on this driver — fan-out happens through the - /// reverse map, not by spinning a pump per - /// subscription. + /// Lazily start the on the first advise. Since #865 the advise paths + /// (probe seed, Subscribe, Read) go through , and only the + /// replay replaces an existing pump. The pump is shared across every subscription on this driver — + /// fan-out happens through the reverse map, not by spinning a pump + /// per subscription. /// private EventPump EnsureEventPumpStarted() { @@ -2168,8 +2181,12 @@ public sealed class GalaxyDriver } } - /// Upper bound on how long an advise waits for a current-session event stream (#865). - internal TimeSpan StreamRequestWait { get; set; } = TimeSpan.FromSeconds(5); + /// + /// Upper bound on how long an advise waits for a current-session event stream (#865). Kept well under + /// Galaxy's Tier-A Read (2 s) / Subscribe (5 s) call budgets, so the "proceed anyway" fallback is + /// reachable before the resilience pipeline times the call out. + /// + internal TimeSpan StreamRequestWait { get; set; } = TimeSpan.FromSeconds(1); /// /// #865 — make sure an advise about to be issued has an event stream for its initial value, and return @@ -2180,20 +2197,30 @@ public sealed class GalaxyDriver /// 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). + /// ( — this narrows, it cannot close, the attach race: the gateway + /// offers no attach acknowledgement). /// /// /// 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. + /// here (only restarts it, so two streams never race for the gateway's + /// single-subscriber slot); while the supervisor is actually reopening/replaying the advise WAITS for the + /// replay's pump. During a reconnect BACKOFF (TransportLost) no replay is imminent, so it does not wait. + /// 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. + /// Bounded by per call (a #861 retry starts a new budget). On timeout the + /// caller proceeds with the current session — no worse than the pre-#865 behaviour, but note the cost: + /// such an advise registers as current, so the replay will not re-advise it, and its initial value is + /// lost until the attribute next changes. + /// + /// + /// A current-session pump whose loop has ENDED is not a stream (its + /// completes on exit). If the supervisor already reports Healthy, its fault was swallowed while it was + /// recovering (it ignores faults outside Healthy) — the driver would be deaf while Healthy — so it is + /// reported again to start a new recovery cycle. /// /// /// Cancels the wait. @@ -2216,16 +2243,22 @@ public sealed class GalaxyDriver var remaining = StreamRequestWait - elapsed.Elapsed; if (pump is not null && pump.Session == session) { + if (pump.HasEnded) + { + ReportEndedPump(pump); + return session; + } if (remaining > TimeSpan.Zero) { try { await pump.StreamRequested.WaitAsync(remaining, cancellationToken).ConfigureAwait(false); + if (pump.HasEnded) ReportEndedPump(pump); return session; } catch (TimeoutException) { - // fall through to the warning + // fall through to the log } } else if (pump.StreamRequested.IsCompleted) @@ -2233,18 +2266,67 @@ public sealed class GalaxyDriver return session; } } - else if (remaining > TimeSpan.Zero) + else if (remaining > TimeSpan.Zero && ReplayExpectedSoon()) { // Reopen → replay window: the replay restarts the pump for the current session. await Task.Delay(TimeSpan.FromMilliseconds(20), cancellationToken).ConfigureAwait(false); continue; } + LogNoCurrentStream(); + return _subscriptions.CurrentSession; + } + } + + /// + /// True when a stale pump will be replaced shortly: the supervisor is reopening/replaying, or there is + /// no supervisor (the injected-seam path, where the replay is driven directly), or a restart is underway. + /// + private bool ReplayExpectedSoon() + { + 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; + } + + /// A current-session pump's loop ended while the supervisor was not listening (see AwaitCurrentEventStreamAsync). + private void ReportEndedPump(EventPump pump) + { + var supervisor = _supervisor; + if (supervisor is null || _runtimeStopping || supervisor.CurrentState != ReconnectSupervisor.State.Healthy) return; + lock (_pumpLock) + { + if (!ReferenceEquals(_eventPump, pump)) return; // already replaced + } + _logger.LogWarning( + "GalaxyDriver {InstanceId} event stream has ended while the reconnect supervisor reports Healthy — re-reporting the transport failure.", + _driverInstanceId); + try { supervisor.ReportTransportFailure(new InvalidOperationException("The Galaxy event stream ended; no events can be received.")); } + catch (ObjectDisposedException) { /* driver disposing */ } + } + + private long _lastNoStreamWarningTicks; + + /// Warning at most once a minute while the supervisor is Healthy; Debug otherwise (an outage would flood). + private void LogNoCurrentStream() + { + var supervisor = _supervisor; + var now = Environment.TickCount64; + var last = Interlocked.Read(ref _lastNoStreamWarningTicks); + if ((supervisor is null || supervisor.CurrentState == ReconnectSupervisor.State.Healthy) + && (last == 0 || now - last >= 60_000) + && Interlocked.CompareExchange(ref _lastNoStreamWarningTicks, now, last) == last) + { _logger.LogWarning( "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; + return; } + _logger.LogDebug( + "GalaxyDriver {InstanceId} advising without a started current-session event stream (supervisor not Healthy or warning rate-limited).", + _driverInstanceId); } /// 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 d7a76d37c..95e6f83d9 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 @@ -314,6 +314,12 @@ internal sealed class SubscriptionRegistry /// internal Action? AfterMapWriteForTests { get; set; } + /// + /// Test seam (#865): invoked on 's lock-free fast path after the handle + /// lookup missed, before the open-window check. + /// + internal Action? AfterFastPathMissForTests { get; set; } + /// /// Test seam (M-1): parks unconditionally, producing the parked list for a /// resolved handle that the atomic registration otherwise makes impossible. @@ -496,7 +502,18 @@ internal sealed class SubscriptionRegistry { subscribers = ResolveSubscribers(ev.ItemHandle); if (subscribers.Count > 0) return EarlyEventOutcome.Resolved; - if (Volatile.Read(ref _pendingSubscribes) == 0) return EarlyEventOutcome.Stale; + 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; + } } lock (_parkGate) 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 474841437..84c62252c 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 @@ -898,6 +898,31 @@ public sealed class EarlyEventWindowTests driver.SubscriptionsForTests.ResolveSubscribers(77).ShouldBeEmpty(); } + /// + /// #865 stress finding: the dispatch fast path looked the handle up, then checked for an open window — + /// not atomically. A subscribe that registered the handle AND closed its window in between made the + /// event "stale" and its initial value was dropped for good. It must resolve. + /// + [Fact] + public void A_registration_landing_between_the_fast_path_lookup_and_the_window_check_still_resolves() + { + var registry = new SubscriptionRegistry(); + var session = registry.CurrentSession; + var window = registry.BeginPendingSubscribe(); + registry.AfterFastPathMissForTests = () => + { + registry.AfterFastPathMissForTests = null; + registry.TryRegister(registry.NextSubscriptionId(), [new TagBinding("Tank.A", 7)], session).ShouldBeTrue(); + window.Dispose(); // the subscribe's window closes right after its registration + }; + + registry.ResolveOrPark(DataChange(7, 1), session, out var subscribers, out _).ShouldBe(EarlyEventOutcome.Resolved); + subscribers.Single().FullReference.ShouldBe("Tank.A"); + + // Positive control for the Stale branch: an unknown handle with no window open is still dropped. + registry.ResolveOrPark(DataChange(8, 1), session, out _, out _).ShouldBe(EarlyEventOutcome.Stale); + } + /// I-B: a purge that completes after a reconnect does not delete the new session's parked values. [Fact] public void A_purge_for_an_older_session_is_a_no_op() 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 774aa762e..36f152888 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 @@ -265,11 +265,56 @@ public sealed class ProbeInitialValueTests (await TestWait.ForAsync(delivered.Task, "the subscribe's initial value must arrive", () => Describe(driver, sub))) .FullReference.ShouldBe("Tank01.Level"); + sub.DroppedInitialValues.ShouldBe(0); + } + + /// A first ReadAsync on a fresh driver (no pump yet) requests the stream before its advise. + [Fact] + public async Task First_read_advises_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); + driver.HasEventPumpForTests.ShouldBeFalse("precondition: the read is the first advise"); + var read = await driver.ReadAsync(["Tank02.Level"], CancellationToken.None); + + var calls = sub.CallsSnapshot(); + Array.IndexOf(calls, "stream").ShouldBe(0, $"the stream must be requested before the read's advise; calls=[{string.Join(" | ", calls)}]"); read[0].StatusCode.ShouldBe(0u, "the read's initial value must arrive (not BadTimeout)"); sub.DroppedInitialValues.ShouldBe(0); } + /// + /// The replay's re-advise of DATA subscriptions (which has no per-advise hook) waits for the restarted + /// pump's stream, so each re-advised tag's new initial value arrives. The value flips so only a + /// post-replay event can satisfy the wait. + /// + [Fact] + public async Task Reconnect_replay_delivers_re_advised_data_subscriptions_initial_values() + { + var sub = new GatewayLikeSubscriber(); + await using var driver = new GalaxyDriver("drv", Opts(), hierarchySource: null, subscriber: sub); + await driver.InitializeAsync("{}", CancellationToken.None); + var last = new System.Collections.Concurrent.ConcurrentDictionary(); + driver.OnDataChange += (_, e) => last[e.FullReference] = e.Snapshot.Value; + + await driver.SubscribeAsync(["Tank01.Level"], TimeSpan.Zero, CancellationToken.None); + await TestWait.UntilAsync(() => last.TryGetValue("Tank01.Level", out var v) && v is true, + "positive control: the subscribe's initial value arrives", () => Describe(driver, sub)); + + sub.ScanStateRunning = false; // the fake's next initial value + await driver.InvokeReplayForTestAsync(CancellationToken.None); + + await TestWait.UntilAsync(() => last.TryGetValue("Tank01.Level", out var v) && v is false, + "the replay's re-advise must deliver its new initial value", + () => Describe(driver, sub) + $" last=[{string.Join(",", last.Select(kv => $"{kv.Key}={kv.Value}"))}] parked={driver.ParkedEarlyEventCountForTests} h2=[{string.Join(",", driver.SubscriptionsForTests.ResolveSubscribers(2).Select(b => b.FullReference))}] h1=[{string.Join(",", driver.SubscriptionsForTests.ResolveSubscribers(1).Select(b => b.FullReference))}]"); + 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