From e35f2a4e93a605fa5963d1f886cc84b5359eef94 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Wed, 11 Nov 2026 12:24:08 -0500 Subject: [PATCH] fix(galaxy): probe advises start the event stream first so initial ScanState arrives (#865) The gateway's StreamEvents with after_worker_sequence = 0 is a fresh stream and never replays, and the event pump started only on the first Read/Subscribe, so platform ScanState probes advised before then lost their initial value (and a driver with no authored tags never got any host status). PerPlatformProbeWatcher now awaits an ensure-event-stream hook before every probe SubscribeBulk (seed, discover sync, reconnect ResubscribeStaleAsync). The driver's hook starts the pump (refused while a teardown is in progress) and waits, bounded, on the new EventPump.StreamRequested signal - completed once the stream loop's first MoveNext has issued the StreamEvents RPC, or when the loop ends - so the advise never precedes the stream request. Claude-Session: https://claude.ai/code/session_01GT67AD99zQQqpLCVm3p7xk --- .../GalaxyDriver.cs | 49 +++- .../Health/PerPlatformProbeWatcher.cs | 18 +- .../Runtime/EventPump.cs | 34 ++- .../Runtime/GalaxyDriverFixRound1Tests.cs | 5 +- .../Runtime/ProbeInitialValueTests.cs | 218 ++++++++++++++++++ 5 files changed, 320 insertions(+), 4 deletions(-) create mode 100644 tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/ProbeInitialValueTests.cs 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..3102724f1 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 @@ -2101,6 +2107,47 @@ public sealed class GalaxyDriver } } + /// + /// for infrastructure callers (the probe watcher) that + /// must not fail because the runtime is going away: false — and no pump — while a teardown is in + /// 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; + return EnsureEventPumpStarted(); // re-enters _pumpLock (Lock is reentrant); cannot throw here + } + } + + /// Upper bound on how long a probe subscribe waits for the pump's StreamEvents call to be issued. + internal TimeSpan ProbeStreamRequestWait { get; set; } = TimeSpan.FromSeconds(5); + + /// + /// #865 — the probe watcher's pre-subscribe hook: start the pump (if this runtime may) and wait, + /// bounded, until its StreamEvents call has been issued, so the gateway has a subscriber to push + /// the probe's initial ScanState value to. A fresh stream (after_worker_sequence = 0) is never + /// replayed, so an initial value pushed before it attaches is lost — and ScanState rarely changes + /// again. On timeout the subscribe proceeds anyway (logged): a late stream is no worse than before. + /// + private async Task EnsureProbeEventStreamAsync(CancellationToken cancellationToken) + { + var pump = TryEnsureEventPumpStarted(); + if (pump is null) return; + try + { + await pump.StreamRequested.WaitAsync(ProbeStreamRequestWait, cancellationToken).ConfigureAwait(false); + } + catch (TimeoutException) + { + _logger.LogWarning( + "GalaxyDriver {InstanceId} event stream was not requested within {Wait} — subscribing the platform probes anyway; their initial ScanState may be missed.", + _driverInstanceId, ProbeStreamRequestWait); + } + } + /// /// Replay-gap callback for the . The gateway told us it evicted /// events we can never receive, so every cached value may be stale by an unknown amount. diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Health/PerPlatformProbeWatcher.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Health/PerPlatformProbeWatcher.cs index fe887b084..08d2dd395 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,23 @@ 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 (the driver starts its + /// and waits for its StreamEvents call to be issued). + /// 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,6 +208,10 @@ 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. + if (_ensureEventStream is not 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(); 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..c3640ab12 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,17 @@ internal sealed class EventPump : IAsyncDisposable private readonly int _session; + private readonly TaskCompletionSource _streamRequested = new(TaskCreationOptions.RunContinuationsAsynchronously); + + /// + /// Completes once the stream loop has issued its StreamEvents call (the first + /// MoveNextAsync runs the subscriber's iterator synchronously up to the RPC) — or once the + /// loop has ended without getting that far, so a waiter never hangs on a dead pump. #865: the + /// gateway does not replay a fresh stream, so a caller whose only expected event is the initial + /// value of an advise it is about to make waits on this first. + /// + 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 +190,17 @@ 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(); + _streamRequested.TrySetResult(); + for (; await next.ConfigureAwait(false); next = enumerator.MoveNextAsync()) { + var ev = enumerator.Current; if (ct.IsCancellationRequested) break; EventsReceived.Add(1, _clientTag); @@ -241,6 +259,20 @@ internal sealed class EventPump : IAsyncDisposable } finally { + // 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/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..2ca4698a7 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 @@ -303,6 +303,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 +331,9 @@ 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. + driver.EventPumpInstanceForTests.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..2144c8f7f --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/ProbeInitialValueTests.cs @@ -0,0 +1,218 @@ +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; + + /// 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; + } + try + { + await foreach (var ev in mine.Reader.ReadAllAsync(cancellationToken).ConfigureAwait(false)) + { + yield return ev; + } + } + finally + { + lock (_gate) + { + if (ReferenceEquals(_attached, mine)) _attached = null; + } + } + } + + 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); + } + + /// + /// 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); + } +}