diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Browse/GatewayGalaxyPlatformSource.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Browse/GatewayGalaxyPlatformSource.cs index dfed16597..875bbddc9 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Browse/GatewayGalaxyPlatformSource.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Browse/GatewayGalaxyPlatformSource.cs @@ -5,11 +5,19 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Browse; /// /// Default : a platform-only DiscoverHierarchy sweep -/// over the gateway's — the request filters on -/// template_chain_contains ($WinPlatform, $AppEngine) and asks for no -/// attributes, so it is a few rows per galaxy rather than the full object + attribute dump the -/// address-space discovery pulls. Paged at the configured Repository.DiscoverPageSize. +/// over the gateway's — one sweep per platform template +/// ($WinPlatform, then $AppEngine), each filtered on that single +/// template_chain_contains token with no attributes, so it is a few rows per galaxy rather +/// than the full object + attribute dump the address-space discovery pulls. Paged at the +/// configured Repository.DiscoverPageSize. /// +/// +/// ⚠️ One template per request (#864). The gateway's shared +/// GalaxyHierarchyProjector.MatchesFilters ANDs the template_chain_contains entries — +/// an object must match every token — so a single request carrying both templates matches no +/// platform (a $WinPlatform's chain has no $AppEngine, and vice versa) and the probe +/// watcher seeded 0 platforms in production. The sweeps are unioned and de-duplicated by tag name. +/// public sealed class GatewayGalaxyPlatformSource : IGalaxyPlatformSource { /// Template-chain tokens that identify a platform / engine object. @@ -32,15 +40,16 @@ public sealed class GatewayGalaxyPlatformSource : IGalaxyPlatformSource => QueryAsync(_client.DiscoverHierarchyRawAsync, _pageSize, cancellationToken); /// - /// The platform-only sweep behind a delegate so it is unit-testable without a channel. Every - /// request carries the template filter and include_attributes = false; the reply is + /// The platform-only sweeps behind a delegate so it is unit-testable without a channel. One + /// paged sweep per entry of ; every request carries exactly that + /// one template token and include_attributes = false; the reply is /// re-filtered on the returned template chain (a gateway that ignores the filter still /// yields only platforms) and reduced to non-empty tag names. /// /// The raw DiscoverHierarchy RPC. /// Objects per page. /// Cancellation token for the walk. - /// Distinct platform tag names in gateway order. + /// Distinct platform tag names: $WinPlatform sweep first, then $AppEngine, each in gateway order. internal static async Task> QueryAsync( Func> discover, int pageSize, @@ -51,33 +60,37 @@ public sealed class GatewayGalaxyPlatformSource : IGalaxyPlatformSource var names = new List(); var seen = new HashSet(StringComparer.OrdinalIgnoreCase); - var seenPageTokens = new HashSet(StringComparer.Ordinal); - var pageToken = string.Empty; - do + foreach (var template in PlatformTemplates) { - var request = new DiscoverHierarchyRequest + var seenPageTokens = new HashSet(StringComparer.Ordinal); + var pageToken = string.Empty; + do { - PageSize = effective, - PageToken = pageToken, - IncludeAttributes = false, - }; - request.TemplateChainContains.Add(PlatformTemplates); + var request = new DiscoverHierarchyRequest + { + PageSize = effective, + PageToken = pageToken, + IncludeAttributes = false, + }; + // Exactly ONE token per request: the gateway ANDs template filters (#864). + request.TemplateChainContains.Add(template); - var reply = await discover(request, cancellationToken).ConfigureAwait(false); - foreach (var obj in reply.Objects) - { - if (string.IsNullOrEmpty(obj.TagName) || !IsPlatform(obj)) continue; - if (seen.Add(obj.TagName)) names.Add(obj.TagName); - } + var reply = await discover(request, cancellationToken).ConfigureAwait(false); + foreach (var obj in reply.Objects) + { + if (string.IsNullOrEmpty(obj.TagName) || !IsPlatform(obj)) continue; + if (seen.Add(obj.TagName)) names.Add(obj.TagName); + } - pageToken = reply.NextPageToken ?? string.Empty; - if (!string.IsNullOrWhiteSpace(pageToken) && !seenPageTokens.Add(pageToken)) - { - throw new InvalidOperationException( - $"Galaxy DiscoverHierarchy (platform query) returned a repeated page token '{pageToken}' — refusing to loop."); + pageToken = reply.NextPageToken ?? string.Empty; + if (!string.IsNullOrWhiteSpace(pageToken) && !seenPageTokens.Add(pageToken)) + { + throw new InvalidOperationException( + $"Galaxy DiscoverHierarchy (platform query, {template}) returned a repeated page token '{pageToken}' — refusing to loop."); + } } + while (!string.IsNullOrWhiteSpace(pageToken)); } - while (!string.IsNullOrWhiteSpace(pageToken)); return names; } 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 fd117b4f7..fb301e855 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/GalaxyDriver.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/GalaxyDriver.cs @@ -540,6 +540,9 @@ public sealed class GalaxyDriver // a cold gateway cache's first-load wait — longer than Galaxy's Tier-A Initialize budget. It // runs in the background: InitializeAsync neither waits on it nor fails because of it. await PrepareProbeSeedAsync(cancellationToken).ConfigureAwait(false); + // #865: the seed's probe advises start the event pump themselves (BuildProbeWatcher's + // pre-subscribe hook) — a driver with no authored tags never Reads/Subscribes, and the pump + // used to start only there, so every initial ScanState value went to no stream. RequestProbeSeed("initialize"); _logger.LogInformation( "GalaxyDriver {InstanceId} initialized — endpoint={Endpoint} clientName={ClientName}", @@ -730,9 +733,12 @@ public sealed class GalaxyDriver // The watcher publishes its probe handles into the driver's registry (reserved id) so the // EventPump's fan-out reaches OnPumpDataChange for ScanState events — without that every // probe event was dropped before it got anywhere near the watcher. + // #865: the watcher also starts the pump before every probe SubscribeBulk — the initial ScanState + // value only reaches a stream that is already attached (the gateway does not replay a fresh one). private PerPlatformProbeWatcher BuildProbeWatcher(IGalaxySubscriber subscriber) => new(subscriber, _hostStatuses, _subscriptions, _logger, - bufferedUpdateIntervalMs: _options.MxAccess.PublishingIntervalMs); + bufferedUpdateIntervalMs: _options.MxAccess.PublishingIntervalMs, + ensureEventStream: EnsureProbeEventStreamAsync); /// /// Build the production gw client + session + per-capability runtime components @@ -1047,6 +1053,26 @@ 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 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 }) + { + // A teardown cancelling this replay can end the pump too: that is a clean cancellation. + cancellationToken.ThrowIfCancellationRequested(); + // The synchronous case; a stream that ends a round trip later reports a transport failure + // while Replaying, which the supervisor itself turns into a failed attempt. + 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). @@ -1302,27 +1328,45 @@ public sealed class GalaxyDriver { old = _eventPump; if (old is null) return; // pump never started — nothing to restart + // #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. @@ -1675,7 +1719,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. - 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. + 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 @@ -1712,7 +1768,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 @@ -1920,7 +1976,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) @@ -1953,6 +2009,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 @@ -1961,7 +2021,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); @@ -1970,7 +2030,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: @@ -2069,11 +2128,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() { @@ -2089,18 +2162,206 @@ public sealed class GalaxyDriver "GalaxyDriver runtime is shutting down — no subscription can be served until the next InitializeAsync."); } if (_eventPump is not null) return _eventPump; - _eventPump = new EventPump( + // The fault callback carries the pump's identity, so a late fault from a pump that is no + // longer the live one (a stale pump of a reopened session) is not escalated (#865 review). + EventPump? created = null; + created = new EventPump( _subscriber!, _subscriptions, _logger, channelCapacity: _options.MxAccess.EventPumpChannelCapacity, clientName: _options.MxAccess.ClientName, - onStreamFault: OnEventPumpStreamFault, + onStreamFault: cause => OnEventPumpStreamFault(created, cause), onReplayGap: OnEventPumpReplayGap); + _eventPump = created; _eventPump.OnDataChange += OnPumpDataChange; _eventPump.Start(); return _eventPump; } } + /// + /// 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() + { + lock (_pumpLock) + { + if (_runtimeStopping || _disposed || _subscriber is null) return null; + 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 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 + /// the registry session to issue (and register) it under. + /// + /// + /// + /// 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: 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 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 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. + /// The registry session to issue the advise under. + private async Task AwaitCurrentEventStreamAsync(CancellationToken cancellationToken) + { + 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 (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 log + } + } + else if (pump.StreamRequested.IsCompleted) + { + return session; + } + } + 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: 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() + { + if (Volatile.Read(ref _pumpRestarting)) return true; + var supervisor = _supervisor; + if (supervisor is null) return true; + // 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). + 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; + } + _logger.LogDebug( + "GalaxyDriver {InstanceId} advising without a started current-session event stream (supervisor not Healthy or warning rate-limited).", + _driverInstanceId); + } + + /// + /// #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 Task EnsureProbeEventStreamAsync(CancellationToken cancellationToken) + { + // A seed pass a teardown already cancelled must not start a pump after the teardown finished. + cancellationToken.ThrowIfCancellationRequested(); + return AwaitCurrentEventStreamAsync(cancellationToken); + } + /// /// 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. @@ -2184,7 +2445,12 @@ public sealed class GalaxyDriver /// replay; without this hand-off a transient transport drop permanently kills /// the event stream and GetHealth() keeps reporting Healthy. /// - private void OnEventPumpStreamFault(Exception cause) + 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) { @@ -2194,6 +2460,23 @@ public sealed class GalaxyDriver return; } + if (source is not null) + { + bool stale; + lock (_pumpLock) stale = !ReferenceEquals(_eventPump, source) || source.Session != _subscriptions.CurrentSession; + if (stale) + { + // 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/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Health/PerPlatformProbeWatcher.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Health/PerPlatformProbeWatcher.cs index fe887b084..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 @@ -44,6 +44,11 @@ public sealed class PerPlatformProbeWatcher : IDisposable // driven directly (unit tests of the decoding table). private readonly SubscriptionRegistry? _registry; + // #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; + // 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. // Each handle carries the registry session that issued it (#859 I-A): item handles are per gateway @@ -79,16 +84,24 @@ public sealed class PerPlatformProbeWatcher : IDisposable /// The driver's subscription registry; null to keep the handles private. /// Optional logger for diagnostic messages. /// 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 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, HostStatusAggregator aggregator, SubscriptionRegistry? registry, ILogger? logger = null, - int bufferedUpdateIntervalMs = 0) + int bufferedUpdateIntervalMs = 0, + Func>? ensureEventStream = null) { _subscriber = subscriber ?? throw new ArgumentNullException(nameof(subscriber)); _aggregator = aggregator ?? throw new ArgumentNullException(nameof(aggregator)); _registry = registry; + _ensureEventStream = ensureEventStream; _logger = logger ?? NullLogger.Instance; if (bufferedUpdateIntervalMs < 0) { @@ -196,10 +209,18 @@ public sealed class PerPlatformProbeWatcher : IDisposable { var probeAddresses = platforms.Select(p => p + ProbeSuffix).ToArray(); + // #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. + // 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 e2be7600e..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 @@ -148,6 +148,36 @@ 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; + + /// True once the stream loop has exited (faulted, completed by the gateway, or disposed). + 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); + + /// + /// 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; + /// /// Marker written into the channel to wake the dispatch loop after a registration promoted /// parked early events (#859). Never dispatched itself; compared by reference. @@ -179,10 +209,29 @@ internal sealed class EventPump : IAsyncDisposable private async Task RunAsync(CancellationToken ct) { + IAsyncEnumerator? enumerator = null; try { - await foreach (var ev in _subscriber.StreamEventsAsync(ct).WithCancellation(ct).ConfigureAwait(false)) + enumerator = _subscriber.StreamEventsAsync(ct).GetAsyncEnumerator(ct); + // Start the first MoveNext BEFORE awaiting it: that synchronous step issues the RPC, which is + // the moment StreamRequested reports (#865). + var next = enumerator.MoveNextAsync(); + 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; if (ct.IsCancellationRequested) break; EventsReceived.Add(1, _clientTag); @@ -241,6 +290,21 @@ 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) + { + try { await enumerator.DisposeAsync().ConfigureAwait(false); } + catch (Exception ex) when (ct.IsCancellationRequested || ex is OperationCanceledException) + { + // Disposing a cancelled stream may surface its cancellation — shutdown noise. + } + catch (Exception ex) + { + _logger.LogDebug(ex, "Galaxy EventPump stream enumerator dispose failed."); + } + } // Tell the dispatch loop the producer is done so it drains and exits. _channel.Writer.TryComplete(); } 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..d6ca1318f 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,18 @@ public sealed class ReconnectSupervisor : IDisposable get { lock (_stateLock) return _state; } } + private bool _attemptInProgress; + + // Set (under _stateLock) by a ReportTransportFailure that arrives while Replaying; cleared on entering Replaying. + private bool _faultDuringReplay; + + /// + /// 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,10 +148,16 @@ 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) { - // Already recovering — nothing else to do. + // Already recovering — nothing else to do, except: a failure DURING the replay means the + // replay ran onto a transport that is already gone again (#865 — e.g. the restarted event + // stream refused or ended); the replay's success must not then be reported as Healthy. + if (_state == State.Replaying) _faultDuringReplay = true; _logger.LogDebug("Transport failure reported during {State}: {Message}", _state, cause.Message); return; } @@ -208,20 +226,32 @@ 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 } // === Replaying phase === - lock (_stateLock) TransitionLocked(State.Replaying, _lastError); + lock (_stateLock) + { + _faultDuringReplay = false; + TransitionLocked(State.Replaying, _lastError); + } try { @@ -234,12 +264,32 @@ public sealed class ReconnectSupervisor : IDisposable lock (_stateLock) { _lastError = ex.Message; } continue; // back to backoff + retry } + finally + { + Volatile.Write(ref _attemptInProgress, false); + } - // === Done === + // === Done — unless the transport failed again while the replay ran === + bool faultedDuringReplay; + string? replayFault; lock (_stateLock) { - _lastError = null; - TransitionLocked(State.Healthy, null); + faultedDuringReplay = _faultDuringReplay; + replayFault = _lastError; + if (!faultedDuringReplay) + { + _lastError = null; + TransitionLocked(State.Healthy, null); + } + } + if (faultedDuringReplay) + { + // Going Healthy here would hide a dead transport (the fault was swallowed above), and the + // next report would restart at attempt 1 with no backoff. Retry this cycle with backoff. + _logger.LogWarning( + "Galaxy replay completed but the transport failed again while it ran (attempt {Attempt}: {Fault}) — retrying", + attempt, replayFault); + continue; } _logger.LogInformation("Galaxy reconnect succeeded after {Attempt} attempt(s)", attempt); return; @@ -282,12 +332,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 d7a76d37c..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 @@ -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,12 @@ 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(); + // #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/Browse/GatewayGalaxyPlatformSourceTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Browse/GatewayGalaxyPlatformSourceTests.cs index a385ecfb8..130d98de3 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Browse/GatewayGalaxyPlatformSourceTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Browse/GatewayGalaxyPlatformSourceTests.cs @@ -15,33 +15,99 @@ public sealed class GatewayGalaxyPlatformSourceTests return o; } - /// Every request carries the platform template filter, no attributes, and the page size; replies are re-filtered. - [Fact] - public async Task Requests_filter_on_platform_templates_without_attributes_and_replies_are_refiltered() + /// + /// A fake gateway that applies the request filters with the SAME semantics as the shared + /// GalaxyHierarchyProjector.MatchesFilters: every template_chain_contains token must + /// match (AND, case-insensitive substring over the chain), category_ids is an OR set. Pages + /// the filtered result at the request's page size. + /// + private sealed class FilteringGateway(params GalaxyObject[] galaxy) { - var requests = new List(); - Task Discover(DiscoverHierarchyRequest req, CancellationToken ct) + public List Requests { get; } = []; + + public Task Discover(DiscoverHierarchyRequest req, CancellationToken ct) { - requests.Add(req); + Requests.Add(req.Clone()); + var matched = galaxy.Where(o => Matches(o, req)).ToList(); + var offset = req.PageToken == "" ? 0 : int.Parse(req.PageToken.Split(':')[1]); var reply = new DiscoverHierarchyReply(); - if (req.PageToken == "") + reply.Objects.Add(matched.Skip(offset).Take(req.PageSize)); + if (offset + req.PageSize < matched.Count) { - reply.Objects.Add([Obj("Plat01", "$WinPlatform"), Obj("Tank01", "$UserDefined")]); - reply.NextPageToken = "p2"; - } - else - { - reply.Objects.Add([Obj("Eng01", "$AppEngine", "$WinPlatform"), Obj("Plat01", "$WinPlatform"), Obj("", "$WinPlatform")]); + reply.NextPageToken = $"{string.Join(",", req.TemplateChainContains)}:{offset + req.PageSize}"; } return Task.FromResult(reply); } - var names = await GatewayGalaxyPlatformSource.QueryAsync(Discover, pageSize: 50, CancellationToken.None); + private static bool Matches(GalaxyObject obj, DiscoverHierarchyRequest req) + { + if (req.CategoryIds.Count > 0 && !req.CategoryIds.Contains(obj.CategoryId)) return false; + foreach (var filter in req.TemplateChainContains) + { + if (!obj.TemplateChain.Any(t => t.Contains(filter, StringComparison.OrdinalIgnoreCase))) return false; + } + return true; + } + } - names.ShouldBe(["Plat01", "Eng01"]); - requests.Count.ShouldBe(2); - requests.ShouldAllBe(r => r.PageSize == 50 && r.IncludeAttributes == false); - requests.ShouldAllBe(r => r.TemplateChainContains.Contains("$WinPlatform") && r.TemplateChainContains.Contains("$AppEngine")); + /// + /// #864: against a gateway that ANDs template filters (as the real one does), every deployed + /// $WinPlatform AND every $AppEngine is found — the old single request carrying both + /// tokens matched none. Shape of the VD03 DEV galaxy: 3 platforms, 2 engines. + /// + [Fact] + public async Task Finds_every_platform_and_engine_against_an_AND_filtering_gateway() + { + var gw = new FilteringGateway( + Obj("CTRX_DEV", "$WinPlatform"), + Obj("DevWinPlatform", "$WinPlatform"), + Obj("JoePlatform", "$JoePlatformTemplate", "$WinPlatform"), + Obj("DevAppEngine", "$AppEngine"), + Obj("JoeAppEngine", "$AppEngine"), + Obj("Tank01", "$UserDefined"), + Obj("Area01", "$Area")); + + var names = await GatewayGalaxyPlatformSource.QueryAsync(gw.Discover, pageSize: 50, CancellationToken.None); + + names.ShouldBe(["CTRX_DEV", "DevWinPlatform", "JoePlatform", "DevAppEngine", "JoeAppEngine"]); + gw.Requests.ShouldAllBe(r => r.TemplateChainContains.Count == 1 && r.PageSize == 50 && r.IncludeAttributes == false); + gw.Requests.Select(r => r.TemplateChainContains[0]).Distinct().ShouldBe(["$WinPlatform", "$AppEngine"]); + } + + /// Each per-template sweep pages to completion independently; tag names are de-duplicated across sweeps. + [Fact] + public async Task Each_template_sweep_pages_independently_and_names_are_deduplicated() + { + var gw = new FilteringGateway( + Obj("Plat01", "$WinPlatform"), + Obj("Plat02", "$WinPlatform"), + Obj("Plat03", "$WinPlatform"), + // An object whose chain carries both templates is reported once. + Obj("Hybrid01", "$AppEngine", "$WinPlatform"), + Obj("Eng01", "$AppEngine"), + Obj("Eng02", "$AppEngine"), + Obj("", "$WinPlatform")); + + var names = await GatewayGalaxyPlatformSource.QueryAsync(gw.Discover, pageSize: 2, CancellationToken.None); + + names.ShouldBe(["Plat01", "Plat02", "Plat03", "Hybrid01", "Eng01", "Eng02"]); + // $WinPlatform: 5 matches (incl. the empty tag) => 3 pages; $AppEngine: 3 matches => 2 pages. + gw.Requests.Count(r => r.TemplateChainContains[0] == "$WinPlatform").ShouldBe(3); + gw.Requests.Count(r => r.TemplateChainContains[0] == "$AppEngine").ShouldBe(2); + } + + /// A gateway that ignores the filter still yields only platforms (replies are re-filtered on the chain). + [Fact] + public async Task Replies_are_refiltered_on_the_template_chain() + { + Task Discover(DiscoverHierarchyRequest req, CancellationToken ct) + { + var reply = new DiscoverHierarchyReply(); + reply.Objects.Add([Obj("Plat01", "$WinPlatform"), Obj("Tank01", "$UserDefined"), Obj("Eng01", "$AppEngine")]); + return Task.FromResult(reply); + } + + (await GatewayGalaxyPlatformSource.QueryAsync(Discover, pageSize: 50, CancellationToken.None)).ShouldBe(["Plat01", "Eng01"]); } /// A gateway that strips the template chain is trusted to have filtered server-side. 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..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 @@ -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!)); @@ -897,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() @@ -1027,6 +1053,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/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 b22f8747c..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, @@ -303,6 +306,7 @@ public sealed class GalaxyDriverFixRound1Tests await driver.SubscribeAsync(["Tank01.Level"], TimeSpan.Zero, CancellationToken.None); driver.HasEventPumpForTests.ShouldBeTrue("positive control: subscribe works on a live runtime"); + var oldPump = driver.EventPumpInstanceForTests; sub.ParkUnsubscribe = true; var shutdown = driver.ShutdownAsync(CancellationToken.None); @@ -330,7 +334,12 @@ public sealed class GalaxyDriverFixRound1Tests driver.TrackedSubscriptionCountForTests.ShouldBe(0); await driver.InitializeAsync("{}", CancellationToken.None); - driver.HasEventPumpForTests.ShouldBeFalse("the next runtime starts with no inherited pump"); + // #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. + 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 new file mode 100644 index 000000000..e496d7fae --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/ProbeInitialValueTests.cs @@ -0,0 +1,419 @@ +using System.Runtime.CompilerServices; +using System.Threading.Channels; +using Google.Protobuf.WellKnownTypes; +using Shouldly; +using Xunit; +using ZB.MOM.WW.MxGateway.Contracts.Proto; +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; +using ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Browse; +using ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Config; +using ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Runtime; + +namespace ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests.Runtime; + +/// +/// #865: the platform ScanState probes' initial value. The gateway pushes an advise's current value +/// once, as its first event, and a FRESH StreamEvents (after_worker_sequence = 0) never +/// replays — so a probe advised before the driver's event stream is attached never gets a value, and +/// ScanState rarely changes again. The event pump used to start only on the first Read/Subscribe, +/// so a driver with no authored tags (or any driver before its first read) had no host statuses. +/// +public sealed class ProbeInitialValueTests +{ + private static readonly string[] Platforms = ["CTRX_DEV", "DevWinPlatform", "DevAppEngine"]; + + private static GalaxyDriverOptions Opts() => new( + new GalaxyGatewayOptions("https://mxgw.test:5001", "key"), + new GalaxyMxAccessOptions("ProbeInitialValue"), + new GalaxyRepositoryOptions(WatchDeployEvents: false), + new GalaxyReconnectOptions()); + + private sealed class FixedPlatformSource(params string[] platforms) : IGalaxyPlatformSource + { + public Task> GetPlatformTagNamesAsync(CancellationToken cancellationToken) + => Task.FromResult>(platforms); + } + + /// + /// Models the gateway's event semantics: every advise pushes its current value exactly once, and + /// only to a stream that is attached at that moment (no replay for a fresh stream). Records the + /// order in which the stream was requested and the advises were made. + /// + private sealed class GatewayLikeSubscriber : IGalaxySubscriber + { + private readonly Lock _gate = new(); + private Channel? _attached; + private int _nextHandle; + + /// "stream" per StreamEvents call, "subscribe:<refs>" per SubscribeBulk, in call order. + public List Calls { get; } = []; + + /// 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; + + public Task> SubscribeBulkAsync( + IReadOnlyList fullReferences, int bufferedUpdateIntervalMs, CancellationToken cancellationToken) + { + var results = new List(); + lock (_gate) + { + Calls.Add("subscribe:" + string.Join(",", fullReferences)); + foreach (var r in fullReferences) + { + var h = ++_nextHandle; + results.Add(new SubscribeResult { TagAddress = r, ItemHandle = h, WasSuccessful = true }); + var initial = new MxEvent + { + Family = MxEventFamily.OnDataChange, + ItemHandle = h, + Value = new MxValue { BoolValue = ScanStateRunning }, + Quality = 192, + SourceTimestamp = Timestamp.FromDateTime(DateTime.UtcNow), + }; + if (_attached is null || !_attached.Writer.TryWrite(initial)) Interlocked.Increment(ref DroppedInitialValues); + } + } + return Task.FromResult>(results); + } + + public Task UnsubscribeBulkAsync(IReadOnlyList itemHandles, CancellationToken cancellationToken) => Task.CompletedTask; + + public async IAsyncEnumerable StreamEventsAsync([EnumeratorCancellation] CancellationToken cancellationToken) + { + // Attached synchronously on the first MoveNext — the point a real gRPC call is issued. + var mine = Channel.CreateUnbounded(); + lock (_gate) + { + 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)) + { + yield return ev; + } + } + finally + { + lock (_gate) + { + if (ReferenceEquals(_attached, mine)) _attached = null; + } + } + } + + /// 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]; + } + } + + private static HostState? StateOf(GalaxyDriver driver, string host) => + driver.GetHostStatuses().FirstOrDefault(s => s.HostName == host)?.State; + + private static string Describe(GalaxyDriver driver, GatewayLikeSubscriber sub) => + $"hosts=[{string.Join(", ", driver.GetHostStatuses().Select(s => $"{s.HostName}={s.State}"))}] " + + $"calls=[{string.Join(" | ", sub.CallsSnapshot())}] dropped={sub.DroppedInitialValues} pump={driver.HasEventPumpForTests}"; + + /// + /// A driver with NO authored tags — nothing ever Reads or Subscribes — still reports every seeded + /// platform with its initial ScanState. Before #865 the pump never started, every initial value + /// was pushed to no stream, and only the transport entry ever appeared. + /// + [Fact] + public async Task Driver_with_no_authored_tags_reports_every_platform_from_its_initial_ScanState() + { + var sub = new GatewayLikeSubscriber(); + await using var driver = new GalaxyDriver("drv", Opts(), hierarchySource: null, subscriber: sub, + platformSource: new FixedPlatformSource(Platforms)); + + await driver.InitializeAsync("{}", CancellationToken.None); + + await TestWait.UntilAsync( + () => Platforms.All(p => StateOf(driver, p) == HostState.Running), + "every seeded platform must surface Running from its initial ScanState, with no Read/Subscribe", + () => Describe(driver, sub)); + sub.DroppedInitialValues.ShouldBe(0, "no probe's initial value may be pushed before the stream is attached"); + driver.TrackedSubscriptionCountForTests.ShouldBe(1, "only the probe entry — no data subscription was ever made"); + } + + /// + /// Ordering: the event stream is requested BEFORE the first probe SubscribeBulk. On the old code + /// the probe advise went out with no stream at all. + /// + [Fact] + public async Task Probe_subscribe_is_issued_only_after_the_event_stream_is_requested() + { + var sub = new GatewayLikeSubscriber(); + await using var driver = new GalaxyDriver("drv", Opts(), hierarchySource: null, subscriber: sub, + platformSource: new FixedPlatformSource(Platforms)); + + await driver.InitializeAsync("{}", CancellationToken.None); + await TestWait.UntilAsync( + () => driver.ProbeWatcherForTests?.WatchedPlatforms.Count == Platforms.Length, + "positive control: the background seed must advise every platform", + () => Describe(driver, sub)); + + var calls = sub.CallsSnapshot(); + var firstSubscribe = Array.FindIndex(calls, c => c.StartsWith("subscribe:", StringComparison.Ordinal)); + var firstStream = Array.IndexOf(calls, "stream"); + firstSubscribe.ShouldBeGreaterThanOrEqualTo(0); + firstStream.ShouldBeGreaterThanOrEqualTo(0, $"the probe seed must start the event stream; calls=[{string.Join(" | ", calls)}]"); + firstStream.ShouldBeLessThan(firstSubscribe, $"the stream must be requested before the probe advise; calls=[{string.Join(" | ", calls)}]"); + } + + /// + /// A reconnect replay re-advises the probes on the recreated session; their new initial values must + /// arrive too (the replay restarts the pump first). The value flips to Stopped so the presence wait + /// can only pass on a post-replay event. + /// + [Fact] + public async Task Reconnect_replay_delivers_the_re_advised_probes_initial_values() + { + var sub = new GatewayLikeSubscriber(); + await using var driver = new GalaxyDriver("drv", Opts(), hierarchySource: null, subscriber: sub, + platformSource: new FixedPlatformSource(Platforms)); + 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)); + + sub.ScanStateRunning = false; + await driver.InvokeReplayForTestAsync(CancellationToken.None); + + await TestWait.UntilAsync( + () => Platforms.All(p => StateOf(driver, p) == HostState.Stopped), + "the re-advised probes' initial values (Stopped) must reach the host statuses after a replay", + () => Describe(driver, sub)); + 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 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() + { + var sub = new GatewayLikeSubscriber(); + await using var driver = new GalaxyDriver("drv", Opts(), hierarchySource: null, subscriber: sub, + 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( + () => 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; + 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 driver.InvokeReplayForTestAsync(CancellationToken.None); + await TestWait.ForAsync(sync, "the window advise completes once the replay restarted the pump"); + + await TestWait.UntilAsync( + () => 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)); + 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"); + + 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 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"); + } + + /// + /// #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 + /// the init path. + /// + [Fact] + public async Task Probe_sync_without_a_platform_source_still_starts_the_stream_first() + { + var sub = new GatewayLikeSubscriber(); + // No platform source: init starts no pump; the probe set arrives through SyncPlatformsAsync. + await using var driver = new GalaxyDriver("drv", Opts(), hierarchySource: null, subscriber: sub); + await driver.InitializeAsync("{}", CancellationToken.None); + driver.HasEventPumpForTests.ShouldBeFalse("precondition: nothing has started the pump yet"); + + await driver.ProbeWatcherForTests.ShouldNotBeNull().SyncPlatformsAsync(Platforms, CancellationToken.None); + + await TestWait.UntilAsync( + () => Platforms.All(p => StateOf(driver, p) == HostState.Running), + "the watcher must start the stream before advising, so the initial values arrive", + () => Describe(driver, sub)); + sub.DroppedInitialValues.ShouldBe(0); + } +} 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..10ecb016a 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,85 @@ 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 review: a transport failure reported WHILE the replay runs (e.g. the restarted event stream is + /// refused a round trip later) must fail that attempt — not go Healthy over a dead transport. The + /// supervisor retries (with backoff) and goes Healthy only after a clean replay. Without the fix the + /// first replay's success went Healthy and replay ran once. + /// + [Fact] + public async Task A_failure_reported_during_the_replay_fails_that_attempt() + { + ReconnectSupervisor? self = null; + var replays = 0; + using var sup = new ReconnectSupervisor( + reopen: _ => Task.CompletedTask, + replay: _ => + { + if (Interlocked.Increment(ref replays) == 1) + { + self!.ReportTransportFailure(new IOException("stream refused during replay")); + } + return Task.CompletedTask; + }, + options: FastBackoff()); + self = sup; + + sup.ReportTransportFailure(new IOException("drop")); + await TestWait.HealthyAsync(sup, "the second, clean replay must recover"); + Volatile.Read(ref replays).ShouldBe(2, "the faulted replay must not count as recovered"); + } + + /// #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()