From c4bc1ef3096cd4c0293d2c0f9dbd068bbf5e01cc Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Wed, 11 Nov 2026 12:24:03 -0500 Subject: [PATCH 1/9] fix(galaxy): seed platforms with one template filter per request (#864) The gateway's shared GalaxyHierarchyProjector ANDs template_chain_contains entries, so the single request carrying both $WinPlatform and $AppEngine matched no object and the probe watcher always seeded 0 platforms. GatewayGalaxyPlatformSource now runs one paged sweep per template and unions/dedupes by tag name. The tests use a fake gateway that applies the projector's AND semantics (they fail on the old single request). Claude-Session: https://claude.ai/code/session_01GT67AD99zQQqpLCVm3p7xk --- .../Browse/GatewayGalaxyPlatformSource.cs | 69 +++++++----- .../GatewayGalaxyPlatformSourceTests.cs | 102 ++++++++++++++---- 2 files changed, 125 insertions(+), 46 deletions(-) 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/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. From e35f2a4e93a605fa5963d1f886cc84b5359eef94 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Wed, 11 Nov 2026 12:24:08 -0500 Subject: [PATCH 2/9] 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); + } +} From cc1a7ec687f84e262d934d2fad756ac287c97359 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Wed, 11 Nov 2026 12:31:51 -0500 Subject: [PATCH 3/9] fix(galaxy): #865 review round 1 - stale-session pump, data-path stream wait - EnsureEventPumpStarted replaces a pump created for an older registry session (the reopen -> replay window still held the dead session's pump, so an advise there lost its initial value and the replay skipped it); RestartEventPumpForReplay keeps a pump already on the current session. - SubscribeAsync, read-via-subscribe and ReplayAsync also await the bounded StreamRequested wait before advising, not just the probes. - TryEnsureEventPumpStarted tolerates a teardown flipping _runtimeStopping mid-check; the probe hook honours an already-cancelled seed token. - StreamRequested is documented as narrowing, not closing, the attach race. - Tests: reopen-window probe test, EventPump.StreamRequested tests, and the FixRound1 fresh-pump assertion now waits on a positive control. Claude-Session: https://claude.ai/code/session_01GT67AD99zQQqpLCVm3p7xk --- .../GalaxyDriver.cs | 99 +++++++++++++++---- .../Runtime/EventPump.cs | 22 ++++- .../Runtime/EventPumpStreamRequestedTests.cs | 62 ++++++++++++ .../Runtime/GalaxyDriverFixRound1Tests.cs | 5 +- .../Runtime/ProbeInitialValueTests.cs | 45 +++++++++ 5 files changed, 209 insertions(+), 24 deletions(-) create mode 100644 tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/EventPumpStreamRequestedTests.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 3102724f1..97a2d32cb 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/GalaxyDriver.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/GalaxyDriver.cs @@ -1053,6 +1053,10 @@ public sealed class GalaxyDriver // subscriptions to replay the dead pump used to stay in _eventPump, so every later // Subscribe/Read reused it and was silently deaf while health stayed Healthy. RestartEventPumpForReplay(); + // #865: every re-advise below pushes its initial value once, to whatever stream is attached. + EventPump? replayPump; + lock (_pumpLock) replayPump = _eventPump; + if (replayPump is not null) await AwaitStreamRequestedAsync(replayPump, cancellationToken).ConfigureAwait(false); // Only subscriptions still holding the previous session's handles: one registered after the // reopen is already live on the recreated session (re-subscribing it would double-advise). @@ -1308,6 +1312,10 @@ public sealed class GalaxyDriver { old = _eventPump; if (old is null) return; // pump never started — nothing to restart + // #865: a pump created after the reopen's fence (an advise in the reopen → replay window + // started it via EnsureEventPumpStarted) already streams the recreated session, and advises + // registered under it may be waiting for their initial values on it — keep it. + if (old.Session == _subscriptions.CurrentSession) return; _eventPump = null; } @@ -1681,7 +1689,10 @@ public sealed class GalaxyDriver var readSubscriber = _subscriber!; // A live pump must exist (values arrive through it); the read listens on ReadDataChange, not // on this instance, so it survives a reconnect's pump restart. - EnsureEventPumpStarted(); + var readPump = EnsureEventPumpStarted(); + // #865: a read waits ONLY for the initial value, which reaches only an already-requested stream. + // Not the caller's token: a cancel here must still yield per-ref Bad snapshots, and the wait is bounded. + await AwaitStreamRequestedAsync(readPump, CancellationToken.None).ConfigureAwait(false); var subscriptionId = _subscriptions.NextSubscriptionId(); // Pre-allocate one TaskCompletionSource per full-reference so the OnDataChange @@ -1927,6 +1938,8 @@ public sealed class GalaxyDriver var generation = Volatile.Read(ref _runtimeGeneration); var subscriber = _subscriber; var pump = EnsureEventPumpStarted(); + // #865: the advise's initial value reaches only an already-requested stream (no replay). + await AwaitStreamRequestedAsync(pump, cancellationToken).ConfigureAwait(false); var subscriptionId = _subscriptions.NextSubscriptionId(); if (fullReferences.Count == 0) @@ -2081,8 +2094,18 @@ public sealed class GalaxyDriver /// reverse map, not by spinning a pump per /// subscription. /// + /// + /// #865: a pump created for an OLDER registry session is replaced, not reused. Between + /// 's session swap and 's pump restart the + /// current pump still streams the dead gateway session; an advise made in that window goes to the + /// recreated session, so its initial value would reach no stream — and, registered under the + /// current session, the replay would not re-advise it either. A pump whose session is current + /// was created after the reopen's second fence, so its stream is on the recreated session. + /// private EventPump EnsureEventPumpStarted() { + EventPump? stale = null; + EventPump pump; lock (_pumpLock) { // Checked BEFORE the reuse: the flag is set before the teardown takes the pump under this @@ -2094,7 +2117,12 @@ public sealed class GalaxyDriver throw new InvalidOperationException( "GalaxyDriver runtime is shutting down — no subscription can be served until the next InitializeAsync."); } - if (_eventPump is not null) return _eventPump; + if (_eventPump is not null && _eventPump.Session == _subscriptions.CurrentSession) return _eventPump; + if (_eventPump is not null) + { + stale = _eventPump; + stale.OnDataChange -= OnPumpDataChange; + } _eventPump = new EventPump( _subscriber!, _subscriptions, _logger, channelCapacity: _options.MxAccess.EventPumpChannelCapacity, @@ -2103,13 +2131,26 @@ public sealed class GalaxyDriver onReplayGap: OnEventPumpReplayGap); _eventPump.OnDataChange += OnPumpDataChange; _eventPump.Start(); - return _eventPump; + pump = _eventPump; } + + if (stale is not null) DisposeStalePumpInBackground(stale); + return pump; } + /// Dispose a replaced pump off the caller's path (its stream is dead; the dispose only joins its loops). + private void DisposeStalePumpInBackground(EventPump stale) => _ = Task.Run(async () => + { + try { await stale.DisposeAsync().ConfigureAwait(false); } + catch (Exception ex) + { + _logger.LogWarning(ex, "GalaxyDriver {InstanceId} stale-session EventPump dispose failed.", _driverInstanceId); + } + }); + /// /// for infrastructure callers (the probe watcher) that - /// must not fail because the runtime is going away: false — and no pump — while a teardown is in + /// must not fail because the runtime is going away: null — and no pump — while a teardown is in /// progress or there is no subscriber. /// /// The running (or just started) pump; null when none may be started. @@ -2118,36 +2159,58 @@ public sealed class GalaxyDriver lock (_pumpLock) { if (_runtimeStopping || _disposed || _subscriber is null) return null; - return EnsureEventPumpStarted(); // re-enters _pumpLock (Lock is reentrant); cannot throw here + try + { + return EnsureEventPumpStarted(); // re-enters _pumpLock (Lock is reentrant) + } + catch (InvalidOperationException) + { + // _runtimeStopping is written without _pumpLock; a teardown that flipped it between the + // check above and EnsureEventPumpStarted's own check is the same "no pump" answer. + return null; + } } } - /// Upper bound on how long a probe subscribe waits for the pump's StreamEvents call to be issued. - internal TimeSpan ProbeStreamRequestWait { get; set; } = TimeSpan.FromSeconds(5); + /// Upper bound on how long an advise waits for the pump's StreamEvents call to be started. + internal TimeSpan StreamRequestWait { get; set; } = TimeSpan.FromSeconds(5); /// - /// #865 — the probe watcher's pre-subscribe hook: start the pump (if this runtime may) and wait, - /// bounded, until its StreamEvents call has been issued, so the gateway has a subscriber to push - /// the probe's initial ScanState value to. A fresh stream (after_worker_sequence = 0) is never - /// replayed, so an initial value pushed before it attaches is lost — and ScanState rarely changes - /// again. On timeout the subscribe proceeds anyway (logged): a late stream is no worse than before. + /// #865 — wait, bounded, until has started its StreamEvents call, so an + /// advise made next has a stream for its initial value. A fresh stream (after_worker_sequence = 0) + /// is never replayed, so an initial value pushed before it attaches is lost — and a slow-changing + /// attribute (ScanState, a setpoint) may not publish again for a long time. On timeout the caller + /// proceeds (logged): a late stream is no worse than the pre-#865 behaviour. /// - private async Task EnsureProbeEventStreamAsync(CancellationToken cancellationToken) + private async Task AwaitStreamRequestedAsync(EventPump pump, CancellationToken cancellationToken) { - var pump = TryEnsureEventPumpStarted(); - if (pump is null) return; try { - await pump.StreamRequested.WaitAsync(ProbeStreamRequestWait, cancellationToken).ConfigureAwait(false); + await pump.StreamRequested.WaitAsync(StreamRequestWait, cancellationToken).ConfigureAwait(false); } catch (TimeoutException) { _logger.LogWarning( - "GalaxyDriver {InstanceId} event stream was not requested within {Wait} — subscribing the platform probes anyway; their initial ScanState may be missed.", - _driverInstanceId, ProbeStreamRequestWait); + "GalaxyDriver {InstanceId} event stream was not started within {Wait} — advising anyway; initial values may be missed.", + _driverInstanceId, StreamRequestWait); } } + /// + /// #865 — the probe watcher's pre-subscribe hook: start the pump (if this runtime may) and wait for + /// its stream (). A probe's initial ScanState value is usually + /// its only event, and a driver with no authored tags never Reads/Subscribes, so without this its + /// pump — and every host status — never existed. + /// + private async Task EnsureProbeEventStreamAsync(CancellationToken cancellationToken) + { + // A seed pass a teardown already cancelled must not start a pump after the teardown finished. + cancellationToken.ThrowIfCancellationRequested(); + var pump = TryEnsureEventPumpStarted(); + if (pump is null) return; + await AwaitStreamRequestedAsync(pump, cancellationToken).ConfigureAwait(false); + } + /// /// Replay-gap callback for the . The gateway told us it evicted /// events we can never receive, so every cached value may be stale by an unknown amount. diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/EventPump.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/EventPump.cs index c3640ab12..e56d8d57a 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/EventPump.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/EventPump.cs @@ -148,15 +148,27 @@ internal sealed class EventPump : IAsyncDisposable private readonly int _session; + /// + /// The registry session this pump was created for (#865). A pump of an older session was opened + /// against a gateway session a reconnect replaced; its stream will never carry the new session's events. + /// + internal int Session => _session; + private readonly TaskCompletionSource _streamRequested = new(TaskCreationOptions.RunContinuationsAsynchronously); /// - /// Completes once the stream loop has issued its StreamEvents call (the first - /// MoveNextAsync runs the subscriber's iterator synchronously up to the RPC) — or once the - /// loop has ended without getting that far, so a waiter never hangs on a dead pump. #865: the - /// gateway does not replay a fresh stream, so a caller whose only expected event is the initial - /// value of an advise it is about to make waits on this first. + /// Completes once the stream loop has STARTED its StreamEvents call (the first + /// MoveNextAsync runs the subscriber's iterator synchronously up to the gRPC call creation) — + /// or once the loop has ended without getting that far, so a waiter never hangs on a dead pump. + /// #865: the gateway does not replay a fresh stream, so a caller whose only expected event is the + /// initial value of an advise it is about to make waits on this first. /// + /// + /// ⚠️ This NARROWS the stream-attach race, it does not close it: the HTTP/2 request is sent + /// asynchronously and the gateway attaches the subscriber after auth, concurrently with any other + /// RPC. In practice the attach wins (an advise's initial value needs a worker IPC round trip plus an + /// MXAccess advise), but the gateway offers no attach acknowledgement to await. + /// public Task StreamRequested => _streamRequested.Task; /// diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/EventPumpStreamRequestedTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/EventPumpStreamRequestedTests.cs new file mode 100644 index 000000000..d1e28e1d6 --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/EventPumpStreamRequestedTests.cs @@ -0,0 +1,62 @@ +using System.Runtime.CompilerServices; +using Shouldly; +using Xunit; +using ZB.MOM.WW.MxGateway.Contracts.Proto; +using ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Runtime; + +namespace ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests.Runtime; + +/// +/// #865: — what advise paths wait on so an initial value is +/// never pushed before the stream is requested. It must complete once the RPC is started (before any +/// event), and must never strand a waiter on a pump whose loop died before getting that far. +/// +public sealed class EventPumpStreamRequestedTests +{ + private sealed class ScriptedSubscriber(Func> stream) : IGalaxySubscriber + { + public Task> SubscribeBulkAsync( + IReadOnlyList fullReferences, int bufferedUpdateIntervalMs, CancellationToken cancellationToken) + => Task.FromResult>([]); + + public Task UnsubscribeBulkAsync(IReadOnlyList itemHandles, CancellationToken cancellationToken) => Task.CompletedTask; + + public IAsyncEnumerable StreamEventsAsync(CancellationToken cancellationToken) => stream(cancellationToken); + } + + /// A stream that is requested and then never yields: StreamRequested completes; the RPC started first. + [Fact] + public async Task Completes_once_the_stream_is_requested_before_any_event() + { + var rpcStarted = 0; + async IAsyncEnumerable Never([EnumeratorCancellation] CancellationToken ct) + { + Interlocked.Exchange(ref rpcStarted, 1); // the synchronous first step — where a real call is created + await Task.Delay(Timeout.Infinite, ct).ConfigureAwait(false); + yield break; + } + + await using var pump = new EventPump(new ScriptedSubscriber(Never), new SubscriptionRegistry()); + pump.StreamRequested.IsCompleted.ShouldBeFalse("not before Start"); + pump.Start(); + + await TestWait.ForAsync(pump.StreamRequested, "StreamRequested must complete once the stream call is started"); + Volatile.Read(ref rpcStarted).ShouldBe(1, "StreamRequested must not precede the stream call"); + } + + /// A subscriber that throws before reaching the RPC still completes StreamRequested (no stranded waiter). + [Fact] + public async Task Completes_when_the_stream_fails_before_the_rpc() + { + var faulted = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + await using var pump = new EventPump( + new ScriptedSubscriber(_ => throw new InvalidOperationException("not connected")), + new SubscriptionRegistry(), + onStreamFault: ex => faulted.TrySetResult(ex)); + pump.Start(); + + await TestWait.ForAsync(pump.StreamRequested, "a dead pump must not strand a StreamRequested waiter"); + (await TestWait.ForAsync(faulted.Task, "positive control: the failure is still reported as a stream fault")) + .ShouldBeOfType(); + } +} diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/GalaxyDriverFixRound1Tests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/GalaxyDriverFixRound1Tests.cs index 2ca4698a7..e2339a751 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/GalaxyDriverFixRound1Tests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/GalaxyDriverFixRound1Tests.cs @@ -333,7 +333,10 @@ public sealed class GalaxyDriverFixRound1Tests await driver.InitializeAsync("{}", CancellationToken.None); // #865: the new runtime's background probe seed starts a pump of its own (its advises need a live // stream) — but it must be a FRESH one, never the dead subscriber's. - driver.EventPumpInstanceForTests.ShouldNotBeSameAs(oldPump, "the next runtime must not inherit the old pump"); + await WaitUntilAsync(() => driver.ProbeWatcherForTests?.WatchedPlatforms.Count == 1, + "positive control: the new runtime's background seed must advise its platform"); + driver.EventPumpInstanceForTests.ShouldNotBeNull("the seed's probe advise must have started a pump") + .ShouldNotBeSameAs(oldPump, "the next runtime must not inherit the old pump"); } // ───────────────────────────── IMPORTANT 3: bounded gateway close ───────────────────────────── diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/ProbeInitialValueTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/ProbeInitialValueTests.cs index 2144c8f7f..81c986ab2 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/ProbeInitialValueTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/ProbeInitialValueTests.cs @@ -193,6 +193,51 @@ public sealed class ProbeInitialValueTests sub.DroppedInitialValues.ShouldBe(0); } + /// + /// Review finding: a probe advised in the reopen → replay window (the session was swapped, the old + /// pump still streams the dead session) must still get its initial value — the stale-session pump + /// is replaced before the advise, and the replay keeps the replacement (it already streams the + /// recreated session). Reusing the old pump bound the value to the old session, so it never routed, + /// and the replay skipped the platform (its handle is current-session). + /// + [Fact] + public async Task Probe_advised_between_reopen_and_replay_still_gets_its_initial_value() + { + var sub = new GatewayLikeSubscriber(); + await using var driver = new GalaxyDriver("drv", Opts(), hierarchySource: null, subscriber: sub, + platformSource: new FixedPlatformSource(Platforms)) + { + RecreateOverrideForTests = _ => Task.CompletedTask, + }; + await driver.InitializeAsync("{}", CancellationToken.None); + await TestWait.UntilAsync( + () => Platforms.All(p => StateOf(driver, p) == HostState.Running), + "positive control: the initial seed reports Running", + () => Describe(driver, sub)); + var oldPump = driver.EventPumpInstanceForTests; + + await driver.InvokeReopenForTestAsync(CancellationToken.None); // session swapped; no replay yet + sub.ScanStateRunning = false; + await driver.ProbeWatcherForTests.ShouldNotBeNull() + .SyncPlatformsAsync([.. Platforms, "LatePlatform"], CancellationToken.None); + + await TestWait.UntilAsync( + () => StateOf(driver, "LatePlatform") == HostState.Stopped, + "a probe advised in the reopen window must get its initial value", + () => Describe(driver, sub)); + var windowPump = driver.EventPumpInstanceForTests; + windowPump.ShouldNotBeSameAs(oldPump, "the stale-session pump must have been replaced before the advise"); + + await driver.InvokeReplayForTestAsync(CancellationToken.None); + driver.EventPumpInstanceForTests.ShouldBeSameAs(windowPump, "the replay keeps a pump already on the current session"); + await TestWait.UntilAsync( + () => Platforms.All(p => StateOf(driver, p) == HostState.Stopped), + "the replay re-advises the stale probes and their initial values arrive on the kept pump", + () => Describe(driver, sub)); + StateOf(driver, "LatePlatform").ShouldBe(HostState.Stopped); + sub.DroppedInitialValues.ShouldBe(0); + } + /// /// A probe sync on a driver whose pump was never started (no platform source, e.g. the /// DiscoverAsync path) still starts it before advising — the watcher's pre-subscribe hook, not just From 9060ff55617530747703f611f98f52a9751e8c10 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Wed, 11 Nov 2026 12:46:15 -0500 Subject: [PATCH 4/9] fix(galaxy): #865 review round 2 - wait for the replay's pump, never race a second stream Round 1 replaced a stale-session pump from any advise path and disposed the old one in the background; that could start a second stream while the old one still held the gateway's single-subscriber slot, and a kept current-session pump could be dead with no fault re-reported. - EnsureEventPumpStarted reuses again; only ReplayAsync restarts the pump. RestartEventPumpForReplay keeps a current-session pump only while its loop is running (EventPump.HasEnded), and flags the restart so no advise path starts a competing pump in the dispose gap. - AwaitCurrentEventStreamAsync (bounded by StreamRequestWait) starts a pump only when none exists, WAITS for the replay while the pump is stale or restarting, waits for StreamRequested, and returns the session it verified. SubscribeAsync (per #861 attempt), read-via-subscribe and the probe hook issue and fence the advise under that session, so a reconnect after the check is caught as a straddle and retried. - Read: a deadline (not caller) cancel during the wait falls through to per-ref results. - Tests: reopen-window probe test now asserts the advise waits for the replay's single new stream; first Subscribe/Read wait for the stream; the fencing tests that advise in the window use a short wait bound; the production-path seed test uses an empty galaxy (a real probe advise now starts a stream against the fake endpoint). Claude-Session: https://claude.ai/code/session_01GT67AD99zQQqpLCVm3p7xk --- .../GalaxyDriver.cs | 210 +++++++++++------- .../Health/PerPlatformProbeWatcher.cs | 17 +- .../Runtime/EventPump.cs | 3 + .../Runtime/EarlyEventWindowTests.cs | 4 + .../Runtime/GalaxyDriverFixRound1Tests.cs | 5 +- .../Runtime/ProbeInitialValueTests.cs | 64 ++++-- 6 files changed, 199 insertions(+), 104 deletions(-) diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/GalaxyDriver.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/GalaxyDriver.cs index 97a2d32cb..7dbb6506d 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/GalaxyDriver.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/GalaxyDriver.cs @@ -1056,7 +1056,7 @@ public sealed class GalaxyDriver // #865: every re-advise below pushes its initial value once, to whatever stream is attached. EventPump? replayPump; lock (_pumpLock) replayPump = _eventPump; - if (replayPump is not null) await AwaitStreamRequestedAsync(replayPump, cancellationToken).ConfigureAwait(false); + if (replayPump is not null) await AwaitCurrentEventStreamAsync(cancellationToken).ConfigureAwait(false); // Only subscriptions still holding the previous session's handles: one registered after the // reopen is already live on the recreated session (re-subscribing it would double-advise). @@ -1312,31 +1312,45 @@ public sealed class GalaxyDriver { old = _eventPump; if (old is null) return; // pump never started — nothing to restart - // #865: a pump created after the reopen's fence (an advise in the reopen → replay window - // started it via EnsureEventPumpStarted) already streams the recreated session, and advises - // registered under it may be waiting for their initial values on it — keep it. - if (old.Session == _subscriptions.CurrentSession) return; + // #865: a pump first started after the reopen's fence (no pump existed before the reconnect) + // already streams the recreated session, and advises registered under it may be waiting for + // their initial values on it — keep it, unless its loop has ended (a dead pump is always replaced, + // #859 P-1: a stream fault during Replaying is not re-reported by the supervisor). + if (old.Session == _subscriptions.CurrentSession && !old.HasEnded) return; _eventPump = null; + // #865: until the replacement exists, AwaitCurrentEventStreamAsync must not start one of its + // own — two streams would race for the gateway's single-subscriber slot while the old one detaches. + _pumpRestarting = true; } - // Detach + dispose the faulted pump outside the lock so a slow shutdown doesn't - // block a concurrent EnsureEventPumpStarted; the old loop already exited on fault. - old.OnDataChange -= OnPumpDataChange; - try { old.DisposeAsync().AsTask().GetAwaiter().GetResult(); } - catch (Exception ex) + try { - _logger.LogWarning(ex, - "GalaxyDriver {InstanceId} faulted EventPump dispose failed during replay — continuing.", - _driverInstanceId); - } + // Detach + dispose the faulted pump outside the lock so a slow shutdown doesn't + // block a concurrent EnsureEventPumpStarted; the old loop already exited on fault. + old.OnDataChange -= OnPumpDataChange; + try { old.DisposeAsync().AsTask().GetAwaiter().GetResult(); } + catch (Exception ex) + { + _logger.LogWarning(ex, + "GalaxyDriver {InstanceId} faulted EventPump dispose failed during replay — continuing.", + _driverInstanceId); + } - // EnsureEventPumpStarted creates + starts a fresh pump under the lock. Skip the - // recreate if the driver is being disposed or its runtime torn down — the teardown - // already released the pump and a fresh one here would leak past it. - if (_disposed || _runtimeStopping) return; - EnsureEventPumpStarted(); + // EnsureEventPumpStarted creates + starts a fresh pump under the lock. Skip the + // recreate if the driver is being disposed or its runtime torn down — the teardown + // already released the pump and a fresh one here would leak past it. + if (_disposed || _runtimeStopping) return; + EnsureEventPumpStarted(); + } + finally + { + lock (_pumpLock) _pumpRestarting = false; + } } + /// True while is between dropping the old pump and starting its replacement. + private bool _pumpRestarting; + private void OnSupervisorStateChanged(object? sender, StateTransition transition) { // Reflect supervisor state in DriverHealth + transport forwarder. @@ -1689,10 +1703,19 @@ public sealed class GalaxyDriver var readSubscriber = _subscriber!; // A live pump must exist (values arrive through it); the read listens on ReadDataChange, not // on this instance, so it survives a reconnect's pump restart. - var readPump = EnsureEventPumpStarted(); - // #865: a read waits ONLY for the initial value, which reaches only an already-requested stream. - // Not the caller's token: a cancel here must still yield per-ref Bad snapshots, and the wait is bounded. - await AwaitStreamRequestedAsync(readPump, CancellationToken.None).ConfigureAwait(false); + EnsureEventPumpStarted(); + // #865: a read waits ONLY for the initial value, which reaches only a current-session stream that is + // already requested. The internal deadline bounds this too; a deadline (not caller) cancel falls + // through so the read still answers per-ref, as it does for a deadline during the RPC. + int issueSession; + try + { + issueSession = await AwaitCurrentEventStreamAsync(cancellationToken).ConfigureAwait(false); + } + catch (OperationCanceledException) when (!callerToken.IsCancellationRequested) + { + issueSession = _subscriptions.CurrentSession; + } var subscriptionId = _subscriptions.NextSubscriptionId(); // Pre-allocate one TaskCompletionSource per full-reference so the OnDataChange @@ -1729,7 +1752,7 @@ public sealed class GalaxyDriver // window before the RPC and close it right after registration (the finally below), not after the // value wait: a window held for the whole wait would park every unrelated unknown-handle event. var earlyEventWindow = _subscriptions.BeginPendingSubscribe(); - var session = _subscriptions.CurrentSession; // #861 — the session this SubscribeBulk is issued on + var session = issueSession; // #861/#865 — the session this SubscribeBulk is issued on (its stream verified) try { // The RPC gets only the CALLER's token: the internal deadline bounds our wait, not the RPC, so a slow @@ -1938,8 +1961,6 @@ public sealed class GalaxyDriver var generation = Volatile.Read(ref _runtimeGeneration); var subscriber = _subscriber; var pump = EnsureEventPumpStarted(); - // #865: the advise's initial value reaches only an already-requested stream (no replay). - await AwaitStreamRequestedAsync(pump, cancellationToken).ConfigureAwait(false); var subscriptionId = _subscriptions.NextSubscriptionId(); if (fullReferences.Count == 0) @@ -1972,6 +1993,10 @@ public sealed class GalaxyDriver for (var attempt = 1; ; attempt++) { + // #865: per attempt (a #861 retry is a new session): the advise's initial value reaches only a + // current-session stream that is already requested — the gateway does not replay a fresh stream. + var issueSession = await AwaitCurrentEventStreamAsync(cancellationToken).ConfigureAwait(false); + // #859: MXAccess's initial value for a new handle can reach the event pump before this // reply is registered (the gateway gives no reply-before-event ordering, and the reply and // the event ride separate RPCs). Open the early-event window BEFORE the RPC so such events @@ -1980,7 +2005,7 @@ public sealed class GalaxyDriver var earlyEventWindow = _subscriptions.BeginPendingSubscribe(); try { - var session = _subscriptions.CurrentSession; // #861 — the session this SubscribeBulk is issued on + var session = issueSession; // #861/#865 — the session this SubscribeBulk is issued on (its stream verified) var results = await subscriber .SubscribeBulkAsync(dialledRefs, bufferedIntervalMs, cancellationToken) .ConfigureAwait(false); @@ -2094,18 +2119,8 @@ public sealed class GalaxyDriver /// reverse map, not by spinning a pump per /// subscription. /// - /// - /// #865: a pump created for an OLDER registry session is replaced, not reused. Between - /// 's session swap and 's pump restart the - /// current pump still streams the dead gateway session; an advise made in that window goes to the - /// recreated session, so its initial value would reach no stream — and, registered under the - /// current session, the replay would not re-advise it either. A pump whose session is current - /// was created after the reopen's second fence, so its stream is on the recreated session. - /// private EventPump EnsureEventPumpStarted() { - EventPump? stale = null; - EventPump pump; lock (_pumpLock) { // Checked BEFORE the reuse: the flag is set before the teardown takes the pump under this @@ -2117,12 +2132,7 @@ public sealed class GalaxyDriver throw new InvalidOperationException( "GalaxyDriver runtime is shutting down — no subscription can be served until the next InitializeAsync."); } - if (_eventPump is not null && _eventPump.Session == _subscriptions.CurrentSession) return _eventPump; - if (_eventPump is not null) - { - stale = _eventPump; - stale.OnDataChange -= OnPumpDataChange; - } + if (_eventPump is not null) return _eventPump; _eventPump = new EventPump( _subscriber!, _subscriptions, _logger, channelCapacity: _options.MxAccess.EventPumpChannelCapacity, @@ -2131,27 +2141,13 @@ public sealed class GalaxyDriver onReplayGap: OnEventPumpReplayGap); _eventPump.OnDataChange += OnPumpDataChange; _eventPump.Start(); - pump = _eventPump; + return _eventPump; } - - if (stale is not null) DisposeStalePumpInBackground(stale); - return pump; } - /// Dispose a replaced pump off the caller's path (its stream is dead; the dispose only joins its loops). - private void DisposeStalePumpInBackground(EventPump stale) => _ = Task.Run(async () => - { - try { await stale.DisposeAsync().ConfigureAwait(false); } - catch (Exception ex) - { - _logger.LogWarning(ex, "GalaxyDriver {InstanceId} stale-session EventPump dispose failed.", _driverInstanceId); - } - }); - /// - /// for infrastructure callers (the probe watcher) that - /// must not fail because the runtime is going away: null — and no pump — while a teardown is in - /// progress or there is no subscriber. + /// for infrastructure callers that must not fail because the + /// runtime is going away: null — and no pump — while a teardown is in progress or there is no subscriber. /// /// The running (or just started) pump; null when none may be started. private EventPump? TryEnsureEventPumpStarted() @@ -2172,43 +2168,95 @@ public sealed class GalaxyDriver } } - /// Upper bound on how long an advise waits for the pump's StreamEvents call to be started. + /// Upper bound on how long an advise waits for a current-session event stream (#865). internal TimeSpan StreamRequestWait { get; set; } = TimeSpan.FromSeconds(5); /// - /// #865 — wait, bounded, until has started its StreamEvents call, so an - /// advise made next has a stream for its initial value. A fresh stream (after_worker_sequence = 0) - /// is never replayed, so an initial value pushed before it attaches is lost — and a slow-changing - /// attribute (ScanState, a setpoint) may not publish again for a long time. On timeout the caller - /// proceeds (logged): a late stream is no worse than the pre-#865 behaviour. + /// #865 — make sure an advise about to be issued has an event stream for its initial value, and return + /// the registry session to issue (and register) it under. /// - private async Task AwaitStreamRequestedAsync(EventPump pump, CancellationToken cancellationToken) + /// + /// + /// A fresh StreamEvents (after_worker_sequence = 0) is never replayed, so an initial value pushed before + /// the stream attaches is lost — and a slow-changing attribute (ScanState, a setpoint) may not publish + /// again for a long time. So: start the pump if there is none, and wait until its stream call has started + /// ( — this narrows, it cannot close, the attach race). + /// + /// + /// A pump bound to an OLDER session belongs to a gateway session a reconnect replaced (the reopen → + /// replay window): an advise now would go to the recreated session with no stream for it, and — + /// registered under the current session — the replay would not re-advise it. The pump is not replaced + /// here (only restarts it, so two streams never race for the single-subscriber + /// slot); the advise WAITS for the replay's pump instead. The returned session is the one the pump was + /// verified against: a reconnect after this check makes the caller's registration fail as a session + /// straddle (#861), whose retry comes back through here. + /// + /// + /// Bounded by ; on timeout the caller proceeds with the current session + /// (logged) — no worse than the pre-#865 behaviour. + /// + /// + /// Cancels the wait. + /// The registry session to issue the advise under. + private async Task AwaitCurrentEventStreamAsync(CancellationToken cancellationToken) { - try - { - await pump.StreamRequested.WaitAsync(StreamRequestWait, cancellationToken).ConfigureAwait(false); - } - catch (TimeoutException) + var elapsed = System.Diagnostics.Stopwatch.StartNew(); + while (true) { + var session = _subscriptions.CurrentSession; + EventPump? pump; + bool restarting; + lock (_pumpLock) + { + restarting = _pumpRestarting; + pump = restarting ? null : TryEnsureEventPumpStarted(); + } + if (pump is null && !restarting) return session; // tearing down — the caller's own fences decide + + var remaining = StreamRequestWait - elapsed.Elapsed; + if (pump is not null && pump.Session == session) + { + if (remaining > TimeSpan.Zero) + { + try + { + await pump.StreamRequested.WaitAsync(remaining, cancellationToken).ConfigureAwait(false); + return session; + } + catch (TimeoutException) + { + // fall through to the warning + } + } + else if (pump.StreamRequested.IsCompleted) + { + return session; + } + } + else if (remaining > TimeSpan.Zero) + { + // Reopen → replay window: the replay restarts the pump for the current session. + await Task.Delay(TimeSpan.FromMilliseconds(20), cancellationToken).ConfigureAwait(false); + continue; + } + _logger.LogWarning( - "GalaxyDriver {InstanceId} event stream was not started within {Wait} — advising anyway; initial values may be missed.", + "GalaxyDriver {InstanceId} had no started event stream for the current gateway session within {Wait} — advising anyway; initial values may be missed.", _driverInstanceId, StreamRequestWait); + return _subscriptions.CurrentSession; } } /// - /// #865 — the probe watcher's pre-subscribe hook: start the pump (if this runtime may) and wait for - /// its stream (). A probe's initial ScanState value is usually - /// its only event, and a driver with no authored tags never Reads/Subscribes, so without this its - /// pump — and every host status — never existed. + /// #865 — the probe watcher's pre-subscribe hook (). A probe's + /// initial ScanState value is usually its only event, and a driver with no authored tags never + /// Reads/Subscribes, so without this its pump — and every host status — never existed. /// - private async Task EnsureProbeEventStreamAsync(CancellationToken cancellationToken) + private Task EnsureProbeEventStreamAsync(CancellationToken cancellationToken) { // A seed pass a teardown already cancelled must not start a pump after the teardown finished. cancellationToken.ThrowIfCancellationRequested(); - var pump = TryEnsureEventPumpStarted(); - if (pump is null) return; - await AwaitStreamRequestedAsync(pump, cancellationToken).ConfigureAwait(false); + return AwaitCurrentEventStreamAsync(cancellationToken); } /// diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Health/PerPlatformProbeWatcher.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Health/PerPlatformProbeWatcher.cs index 08d2dd395..fc84e59b1 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Health/PerPlatformProbeWatcher.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Health/PerPlatformProbeWatcher.cs @@ -47,7 +47,7 @@ public sealed class PerPlatformProbeWatcher : IDisposable // #865: invoked before every probe SubscribeBulk so the driver's event stream is attached when the // gateway pushes the initial ScanState value (a fresh StreamEvents does not replay earlier events, // and ScanState rarely changes — a lost initial value leaves the host missing indefinitely). - private readonly Func? _ensureEventStream; + private readonly Func>? _ensureEventStream; // Tracked platform → gw item handle. Item handle 0 means the gw rejected the subscribe; // we keep the entry so SyncPlatformsAsync doesn't try to subscribe it again on every call. @@ -86,8 +86,9 @@ public sealed class PerPlatformProbeWatcher : IDisposable /// Buffered update interval in milliseconds; must be >= 0. /// /// Optional: awaited before every probe SubscribeBulk to make sure the event stream that carries - /// the initial ScanState value is running and requested (the driver starts its - /// and waits for its StreamEvents call to be issued). + /// the initial ScanState value is running and requested for the current gateway session (the + /// driver starts its and waits for its StreamEvents call). Returns the + /// registry session the stream was verified against; the advise is issued and fenced under it. /// internal PerPlatformProbeWatcher( IGalaxySubscriber subscriber, @@ -95,7 +96,7 @@ public sealed class PerPlatformProbeWatcher : IDisposable SubscriptionRegistry? registry, ILogger? logger = null, int bufferedUpdateIntervalMs = 0, - Func? ensureEventStream = null) + Func>? ensureEventStream = null) { _subscriber = subscriber ?? throw new ArgumentNullException(nameof(subscriber)); _aggregator = aggregator ?? throw new ArgumentNullException(nameof(aggregator)); @@ -210,12 +211,16 @@ public sealed class PerPlatformProbeWatcher : IDisposable // #865: the stream must be live BEFORE the advise — the initial value is the probe's only // event until the platform's scan state changes, and an unattached stream never sees it. - if (_ensureEventStream is not null) await _ensureEventStream(cancellationToken).ConfigureAwait(false); + // The returned session is the one the stream was verified for: a reconnect after the check makes + // the session test below refuse the reply as a straddle, and the retry comes back through here. + int? streamSession = _ensureEventStream is null + ? null + : await _ensureEventStream(cancellationToken).ConfigureAwait(false); // #859: the initial ScanState value can overtake the SubscribeBulk reply; park it until // PublishBindingsLocked registers the handles below (window closes at method exit). using var earlyEventWindow = _registry?.BeginPendingSubscribe(); - var session = CurrentSession; // the session this SubscribeBulk is issued on + var session = streamSession ?? CurrentSession; // the session this SubscribeBulk is issued on // PR 6.3 — use the configured bufferedUpdateIntervalMs (defaults to 0 = gw cadence // when the driver hasn't overridden MxAccess.PublishingIntervalMs). Probe ScanState // changes are rare so a coarser interval is usually fine; deployments that need diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/EventPump.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/EventPump.cs index e56d8d57a..dc8905f85 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/EventPump.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/EventPump.cs @@ -154,6 +154,9 @@ internal sealed class EventPump : IAsyncDisposable /// internal int Session => _session; + /// True once the stream loop has exited (faulted, completed by the gateway, or disposed). + internal bool HasEnded => _loop is { IsCompleted: true }; + private readonly TaskCompletionSource _streamRequested = new(TaskCreationOptions.RunContinuationsAsynchronously); /// diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/EarlyEventWindowTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/EarlyEventWindowTests.cs index 1d0ade328..474841437 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/EarlyEventWindowTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/EarlyEventWindowTests.cs @@ -844,6 +844,7 @@ public sealed class EarlyEventWindowTests "g", Opts(), hierarchySource: null, dataReader: null, dataWriter: null, subscriber: subscriber) { RecreateOverrideForTests = _ => Task.CompletedTask, + StreamRequestWait = TimeSpan.FromMilliseconds(100), // advises in the reopen window on purpose (#865 bound) }; var delivered = new ConcurrentQueue<(string Ref, double Value)>(); driver.OnDataChange += (_, e) => delivered.Enqueue((e.FullReference, (double)e.Snapshot.Value!)); @@ -1027,6 +1028,9 @@ public sealed class EarlyEventWindowTests "g", Opts(), hierarchySource: null, dataReader: null, dataWriter: null, subscriber: subscriber) { RecreateOverrideForTests = _ => Task.CompletedTask, + // These tests advise in the reopen → replay window on purpose (fencing, not values). #865 makes + // such an advise wait for the replay's pump, bounded; keep the bound short so they don't idle. + StreamRequestWait = TimeSpan.FromMilliseconds(100), }; await driver.InitializeAsync("{}", TestContext.Current.CancellationToken); return driver; diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/GalaxyDriverFixRound1Tests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/GalaxyDriverFixRound1Tests.cs index e2339a751..11c5a095a 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/GalaxyDriverFixRound1Tests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/GalaxyDriverFixRound1Tests.cs @@ -272,7 +272,10 @@ public sealed class GalaxyDriverFixRound1Tests [Fact] public async Task Production_path_queries_the_platform_source_at_init() { - var platforms = new ScriptedPlatformSource(); + // An empty galaxy: since #865 a probe advise starts the event stream first, and this owned runtime's + // gRPC channel points at no real gateway — its stream would fault and (correctly) degrade health. + // The query itself is what this test pins. + var platforms = new ScriptedPlatformSource { Script = _ => [] }; await using var driver = new GalaxyDriver("drv", Opts(), hierarchySource: null, platformSource: platforms) { SessionOpenOverrideForTests = _ => Task.CompletedTask, diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/ProbeInitialValueTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/ProbeInitialValueTests.cs index 81c986ab2..774aa762e 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/ProbeInitialValueTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/ProbeInitialValueTests.cs @@ -195,10 +195,10 @@ public sealed class ProbeInitialValueTests /// /// Review finding: a probe advised in the reopen → replay window (the session was swapped, the old - /// pump still streams the dead session) must still get its initial value — the stale-session pump - /// is replaced before the advise, and the replay keeps the replacement (it already streams the - /// recreated session). Reusing the old pump bound the value to the old session, so it never routed, - /// and the replay skipped the platform (its handle is current-session). + /// pump still streams the dead session) must still get its initial value. The advise waits for the + /// replay's restarted pump instead of advising onto the stale one (its value would route under the + /// old session and be lost, and the replay skips a current-session handle) or starting a competing + /// stream (the gateway's default single-subscriber mode refuses a second attach). /// [Fact] public async Task Probe_advised_between_reopen_and_replay_still_gets_its_initial_value() @@ -218,23 +218,55 @@ public sealed class ProbeInitialValueTests await driver.InvokeReopenForTestAsync(CancellationToken.None); // session swapped; no replay yet sub.ScanStateRunning = false; - await driver.ProbeWatcherForTests.ShouldNotBeNull() + var callsBeforeWindow = sub.CallsSnapshot().Length; + // The window advise must WAIT for the replay's current-session pump rather than advise onto the + // stale one (or start a competing stream). + var sync = driver.ProbeWatcherForTests.ShouldNotBeNull() .SyncPlatformsAsync([.. Platforms, "LatePlatform"], CancellationToken.None); - await TestWait.UntilAsync( - () => StateOf(driver, "LatePlatform") == HostState.Stopped, - "a probe advised in the reopen window must get its initial value", - () => Describe(driver, sub)); - var windowPump = driver.EventPumpInstanceForTests; - windowPump.ShouldNotBeSameAs(oldPump, "the stale-session pump must have been replaced before the advise"); - await driver.InvokeReplayForTestAsync(CancellationToken.None); - driver.EventPumpInstanceForTests.ShouldBeSameAs(windowPump, "the replay keeps a pump already on the current session"); + await TestWait.ForAsync(sync, "the window advise completes once the replay restarted the pump"); + await TestWait.UntilAsync( - () => Platforms.All(p => StateOf(driver, p) == HostState.Stopped), - "the replay re-advises the stale probes and their initial values arrive on the kept pump", + () => StateOf(driver, "LatePlatform") == HostState.Stopped + && Platforms.All(p => StateOf(driver, p) == HostState.Stopped), + "the window advise and the replay's re-advises all get their initial values", () => Describe(driver, sub)); - StateOf(driver, "LatePlatform").ShouldBe(HostState.Stopped); + driver.EventPumpInstanceForTests.ShouldNotBeSameAs(oldPump, "the replay restarted the stale pump"); + var windowCalls = sub.CallsSnapshot()[callsBeforeWindow..]; + var lateAdvise = Array.FindIndex(windowCalls, c => c.Contains("LatePlatform", StringComparison.Ordinal)); + var newStream = Array.IndexOf(windowCalls, "stream"); + newStream.ShouldBeGreaterThanOrEqualTo(0, $"the replay must start a new stream; calls=[{string.Join(" | ", windowCalls)}]"); + lateAdvise.ShouldBeGreaterThan(newStream, $"the window advise must follow the new stream; calls=[{string.Join(" | ", windowCalls)}]"); + windowCalls.Count(c => c == "stream").ShouldBe(1, "no competing stream may be started for the window advise"); + sub.DroppedInitialValues.ShouldBe(0); + } + + /// + /// The data paths too: a first SubscribeAsync and a first ReadAsync on a driver whose pump was never + /// started request the stream BEFORE advising, so their initial values arrive (a read of a value that + /// never changes again would otherwise time out). + /// + [Fact] + public async Task First_subscribe_and_read_advise_only_after_the_stream_is_requested() + { + var sub = new GatewayLikeSubscriber(); + await using var driver = new GalaxyDriver("drv", Opts(), hierarchySource: null, subscriber: sub) + { + ReadViaSubscribeDeadline = TimeSpan.FromSeconds(10), + }; + await driver.InitializeAsync("{}", CancellationToken.None); + var delivered = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + driver.OnDataChange += (_, e) => delivered.TrySetResult(e); + + await driver.SubscribeAsync(["Tank01.Level"], TimeSpan.Zero, CancellationToken.None); + var calls = sub.CallsSnapshot(); + Array.IndexOf(calls, "stream").ShouldBe(0, $"the stream must be requested before the advise; calls=[{string.Join(" | ", calls)}]"); + (await TestWait.ForAsync(delivered.Task, "the subscribe's initial value must arrive", () => Describe(driver, sub))) + .FullReference.ShouldBe("Tank01.Level"); + + var read = await driver.ReadAsync(["Tank02.Level"], CancellationToken.None); + read[0].StatusCode.ShouldBe(0u, "the read's initial value must arrive (not BadTimeout)"); sub.DroppedInitialValues.ShouldBe(0); } From 1c1705fa7748fbc4a2e6dc420d0a747054f5e28e Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Wed, 11 Nov 2026 13:00:19 -0500 Subject: [PATCH 5/9] fix(galaxy): #865 review round 3 - budget-safe stream wait, dead pump, fast-path drop - StreamRequestWait 1 s (under Tier-A Read 2 s / Subscribe 5 s budgets) and a stale pump is only waited on while the supervisor is reopening or replaying (or a restart is underway); in a reconnect backoff the advise proceeds at once. The no-stream log is a rate-limited Warning while Healthy, Debug otherwise. - A current-session pump whose loop has ended is not a stream: if the supervisor already reports Healthy (its fault was swallowed while it was recovering) the failure is re-reported to start a new cycle. - Subscribe/Read no longer start the pump directly (only AwaitCurrentEventStreamAsync does, honouring the replay restart flag); the teardown refusal is an explicit ThrowIfRuntimeStopping. - SubscriptionRegistry.ResolveOrPark fast path: the handle lookup and the open-window check were not atomic, so a subscribe that registered the handle and closed its window in between got its initial value dropped as stale (found by a stress run: the first event of a bulk advise lost). It re-looks-up before declaring stale. Deterministic test via a seam. - Tests: first Read on a fresh driver, replay of a data subscription. Claude-Session: https://claude.ai/code/session_01GT67AD99zQQqpLCVm3p7xk --- .../GalaxyDriver.cs | 120 +++++++++++++++--- .../Runtime/SubscriptionRegistry.cs | 19 ++- .../Runtime/EarlyEventWindowTests.cs | 25 ++++ .../Runtime/ProbeInitialValueTests.cs | 45 +++++++ 4 files changed, 189 insertions(+), 20 deletions(-) diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/GalaxyDriver.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/GalaxyDriver.cs index 7dbb6506d..c9dae1420 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/GalaxyDriver.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/GalaxyDriver.cs @@ -1703,7 +1703,7 @@ public sealed class GalaxyDriver var readSubscriber = _subscriber!; // A live pump must exist (values arrive through it); the read listens on ReadDataChange, not // on this instance, so it survives a reconnect's pump restart. - EnsureEventPumpStarted(); + ThrowIfRuntimeStopping(); // the pump itself is started by AwaitCurrentEventStreamAsync below // #865: a read waits ONLY for the initial value, which reaches only a current-session stream that is // already requested. The internal deadline bounds this too; a deadline (not caller) cancel falls // through so the read still answers per-ref, as it does for a deadline during the RPC. @@ -1960,7 +1960,7 @@ public sealed class GalaxyDriver // generation, so the Register below can tell its SubscribeBulk straddled a teardown. var generation = Volatile.Read(ref _runtimeGeneration); var subscriber = _subscriber; - var pump = EnsureEventPumpStarted(); + ThrowIfRuntimeStopping(); // the pump is started per attempt by AwaitCurrentEventStreamAsync (#865) var subscriptionId = _subscriptions.NextSubscriptionId(); if (fullReferences.Count == 0) @@ -2014,7 +2014,6 @@ public sealed class GalaxyDriver switch (TryRegisterForGeneration(subscriptionId, bindings, generation, session)) { case RegisterOutcome.Registered: - _ = pump; // keep the pump alive for the subscription's lifetime return new GalaxySubscriptionHandle(subscriptionId); case RegisterOutcome.TornDown: @@ -2113,11 +2112,25 @@ public sealed class GalaxyDriver } } + /// A Subscribe/Read is refused while a teardown is in progress (the pump would belong to a dying subscriber). + private void ThrowIfRuntimeStopping() + { + lock (_pumpLock) + { + if (_runtimeStopping) + { + throw new InvalidOperationException( + "GalaxyDriver runtime is shutting down — no subscription can be served until the next InitializeAsync."); + } + } + } + /// - /// Lazily start the on the first subscribe. The pump is - /// shared across every subscription on this driver — fan-out happens through the - /// reverse map, not by spinning a pump per - /// subscription. + /// Lazily start the on the first advise. Since #865 the advise paths + /// (probe seed, Subscribe, Read) go through , and only the + /// replay replaces an existing pump. The pump is shared across every subscription on this driver — + /// fan-out happens through the reverse map, not by spinning a pump + /// per subscription. /// private EventPump EnsureEventPumpStarted() { @@ -2168,8 +2181,12 @@ public sealed class GalaxyDriver } } - /// Upper bound on how long an advise waits for a current-session event stream (#865). - internal TimeSpan StreamRequestWait { get; set; } = TimeSpan.FromSeconds(5); + /// + /// Upper bound on how long an advise waits for a current-session event stream (#865). Kept well under + /// Galaxy's Tier-A Read (2 s) / Subscribe (5 s) call budgets, so the "proceed anyway" fallback is + /// reachable before the resilience pipeline times the call out. + /// + internal TimeSpan StreamRequestWait { get; set; } = TimeSpan.FromSeconds(1); /// /// #865 — make sure an advise about to be issued has an event stream for its initial value, and return @@ -2180,20 +2197,30 @@ public sealed class GalaxyDriver /// A fresh StreamEvents (after_worker_sequence = 0) is never replayed, so an initial value pushed before /// the stream attaches is lost — and a slow-changing attribute (ScanState, a setpoint) may not publish /// again for a long time. So: start the pump if there is none, and wait until its stream call has started - /// ( — this narrows, it cannot close, the attach race). + /// ( — this narrows, it cannot close, the attach race: the gateway + /// offers no attach acknowledgement). /// /// /// A pump bound to an OLDER session belongs to a gateway session a reconnect replaced (the reopen → /// replay window): an advise now would go to the recreated session with no stream for it, and — /// registered under the current session — the replay would not re-advise it. The pump is not replaced - /// here (only restarts it, so two streams never race for the single-subscriber - /// slot); the advise WAITS for the replay's pump instead. The returned session is the one the pump was - /// verified against: a reconnect after this check makes the caller's registration fail as a session - /// straddle (#861), whose retry comes back through here. + /// here (only restarts it, so two streams never race for the gateway's + /// single-subscriber slot); while the supervisor is actually reopening/replaying the advise WAITS for the + /// replay's pump. During a reconnect BACKOFF (TransportLost) no replay is imminent, so it does not wait. + /// The returned session is the one the pump was verified against: a reconnect after this check makes the + /// caller's registration fail as a session straddle (#861), whose retry comes back through here. /// /// - /// Bounded by ; on timeout the caller proceeds with the current session - /// (logged) — no worse than the pre-#865 behaviour. + /// Bounded by per call (a #861 retry starts a new budget). On timeout the + /// caller proceeds with the current session — no worse than the pre-#865 behaviour, but note the cost: + /// such an advise registers as current, so the replay will not re-advise it, and its initial value is + /// lost until the attribute next changes. + /// + /// + /// A current-session pump whose loop has ENDED is not a stream (its + /// completes on exit). If the supervisor already reports Healthy, its fault was swallowed while it was + /// recovering (it ignores faults outside Healthy) — the driver would be deaf while Healthy — so it is + /// reported again to start a new recovery cycle. /// /// /// Cancels the wait. @@ -2216,16 +2243,22 @@ public sealed class GalaxyDriver var remaining = StreamRequestWait - elapsed.Elapsed; if (pump is not null && pump.Session == session) { + if (pump.HasEnded) + { + ReportEndedPump(pump); + return session; + } if (remaining > TimeSpan.Zero) { try { await pump.StreamRequested.WaitAsync(remaining, cancellationToken).ConfigureAwait(false); + if (pump.HasEnded) ReportEndedPump(pump); return session; } catch (TimeoutException) { - // fall through to the warning + // fall through to the log } } else if (pump.StreamRequested.IsCompleted) @@ -2233,18 +2266,67 @@ public sealed class GalaxyDriver return session; } } - else if (remaining > TimeSpan.Zero) + else if (remaining > TimeSpan.Zero && ReplayExpectedSoon()) { // Reopen → replay window: the replay restarts the pump for the current session. await Task.Delay(TimeSpan.FromMilliseconds(20), cancellationToken).ConfigureAwait(false); continue; } + LogNoCurrentStream(); + return _subscriptions.CurrentSession; + } + } + + /// + /// True when a stale pump will be replaced shortly: the supervisor is reopening/replaying, or there is + /// no supervisor (the injected-seam path, where the replay is driven directly), or a restart is underway. + /// + private bool ReplayExpectedSoon() + { + if (Volatile.Read(ref _pumpRestarting)) return true; + var supervisor = _supervisor; + if (supervisor is null) return true; + var state = supervisor.CurrentState; + return state is ReconnectSupervisor.State.Reopening or ReconnectSupervisor.State.Replaying; + } + + /// A current-session pump's loop ended while the supervisor was not listening (see AwaitCurrentEventStreamAsync). + private void ReportEndedPump(EventPump pump) + { + var supervisor = _supervisor; + if (supervisor is null || _runtimeStopping || supervisor.CurrentState != ReconnectSupervisor.State.Healthy) return; + lock (_pumpLock) + { + if (!ReferenceEquals(_eventPump, pump)) return; // already replaced + } + _logger.LogWarning( + "GalaxyDriver {InstanceId} event stream has ended while the reconnect supervisor reports Healthy — re-reporting the transport failure.", + _driverInstanceId); + try { supervisor.ReportTransportFailure(new InvalidOperationException("The Galaxy event stream ended; no events can be received.")); } + catch (ObjectDisposedException) { /* driver disposing */ } + } + + private long _lastNoStreamWarningTicks; + + /// Warning at most once a minute while the supervisor is Healthy; Debug otherwise (an outage would flood). + private void LogNoCurrentStream() + { + var supervisor = _supervisor; + var now = Environment.TickCount64; + var last = Interlocked.Read(ref _lastNoStreamWarningTicks); + if ((supervisor is null || supervisor.CurrentState == ReconnectSupervisor.State.Healthy) + && (last == 0 || now - last >= 60_000) + && Interlocked.CompareExchange(ref _lastNoStreamWarningTicks, now, last) == last) + { _logger.LogWarning( "GalaxyDriver {InstanceId} had no started event stream for the current gateway session within {Wait} — advising anyway; initial values may be missed.", _driverInstanceId, StreamRequestWait); - return _subscriptions.CurrentSession; + return; } + _logger.LogDebug( + "GalaxyDriver {InstanceId} advising without a started current-session event stream (supervisor not Healthy or warning rate-limited).", + _driverInstanceId); } /// diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/SubscriptionRegistry.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/SubscriptionRegistry.cs index d7a76d37c..95e6f83d9 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/SubscriptionRegistry.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/SubscriptionRegistry.cs @@ -314,6 +314,12 @@ internal sealed class SubscriptionRegistry /// internal Action? AfterMapWriteForTests { get; set; } + /// + /// Test seam (#865): invoked on 's lock-free fast path after the handle + /// lookup missed, before the open-window check. + /// + internal Action? AfterFastPathMissForTests { get; set; } + /// /// Test seam (M-1): parks unconditionally, producing the parked list for a /// resolved handle that the atomic registration otherwise makes impossible. @@ -496,7 +502,18 @@ internal sealed class SubscriptionRegistry { subscribers = ResolveSubscribers(ev.ItemHandle); if (subscribers.Count > 0) return EarlyEventOutcome.Resolved; - if (Volatile.Read(ref _pendingSubscribes) == 0) return EarlyEventOutcome.Stale; + AfterFastPathMissForTests?.Invoke(); + if (Volatile.Read(ref _pendingSubscribes) == 0) + { + // #865 (found by its stress run): the lookup above and this read are not atomic. A + // subscribe can register the handle AND close its window in between — then "no window + // open" does not mean "stale", and dropping here lost the initial value for good (the + // first event of a bulk advise, while the dispatch thread was slow). The window closes + // only after its registration, so once no window is open a re-lookup sees every + // registration that window made; nothing is held, so no older event can be overtaken. + subscribers = ResolveSubscribers(ev.ItemHandle); + return subscribers.Count > 0 ? EarlyEventOutcome.Resolved : EarlyEventOutcome.Stale; + } } lock (_parkGate) diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/EarlyEventWindowTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/EarlyEventWindowTests.cs index 474841437..84c62252c 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/EarlyEventWindowTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/EarlyEventWindowTests.cs @@ -898,6 +898,31 @@ public sealed class EarlyEventWindowTests driver.SubscriptionsForTests.ResolveSubscribers(77).ShouldBeEmpty(); } + /// + /// #865 stress finding: the dispatch fast path looked the handle up, then checked for an open window — + /// not atomically. A subscribe that registered the handle AND closed its window in between made the + /// event "stale" and its initial value was dropped for good. It must resolve. + /// + [Fact] + public void A_registration_landing_between_the_fast_path_lookup_and_the_window_check_still_resolves() + { + var registry = new SubscriptionRegistry(); + var session = registry.CurrentSession; + var window = registry.BeginPendingSubscribe(); + registry.AfterFastPathMissForTests = () => + { + registry.AfterFastPathMissForTests = null; + registry.TryRegister(registry.NextSubscriptionId(), [new TagBinding("Tank.A", 7)], session).ShouldBeTrue(); + window.Dispose(); // the subscribe's window closes right after its registration + }; + + registry.ResolveOrPark(DataChange(7, 1), session, out var subscribers, out _).ShouldBe(EarlyEventOutcome.Resolved); + subscribers.Single().FullReference.ShouldBe("Tank.A"); + + // Positive control for the Stale branch: an unknown handle with no window open is still dropped. + registry.ResolveOrPark(DataChange(8, 1), session, out _, out _).ShouldBe(EarlyEventOutcome.Stale); + } + /// I-B: a purge that completes after a reconnect does not delete the new session's parked values. [Fact] public void A_purge_for_an_older_session_is_a_no_op() diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/ProbeInitialValueTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/ProbeInitialValueTests.cs index 774aa762e..36f152888 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/ProbeInitialValueTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/ProbeInitialValueTests.cs @@ -265,11 +265,56 @@ public sealed class ProbeInitialValueTests (await TestWait.ForAsync(delivered.Task, "the subscribe's initial value must arrive", () => Describe(driver, sub))) .FullReference.ShouldBe("Tank01.Level"); + sub.DroppedInitialValues.ShouldBe(0); + } + + /// A first ReadAsync on a fresh driver (no pump yet) requests the stream before its advise. + [Fact] + public async Task First_read_advises_only_after_the_stream_is_requested() + { + var sub = new GatewayLikeSubscriber(); + await using var driver = new GalaxyDriver("drv", Opts(), hierarchySource: null, subscriber: sub) + { + ReadViaSubscribeDeadline = TimeSpan.FromSeconds(10), + }; + await driver.InitializeAsync("{}", CancellationToken.None); + driver.HasEventPumpForTests.ShouldBeFalse("precondition: the read is the first advise"); + var read = await driver.ReadAsync(["Tank02.Level"], CancellationToken.None); + + var calls = sub.CallsSnapshot(); + Array.IndexOf(calls, "stream").ShouldBe(0, $"the stream must be requested before the read's advise; calls=[{string.Join(" | ", calls)}]"); read[0].StatusCode.ShouldBe(0u, "the read's initial value must arrive (not BadTimeout)"); sub.DroppedInitialValues.ShouldBe(0); } + /// + /// The replay's re-advise of DATA subscriptions (which has no per-advise hook) waits for the restarted + /// pump's stream, so each re-advised tag's new initial value arrives. The value flips so only a + /// post-replay event can satisfy the wait. + /// + [Fact] + public async Task Reconnect_replay_delivers_re_advised_data_subscriptions_initial_values() + { + var sub = new GatewayLikeSubscriber(); + await using var driver = new GalaxyDriver("drv", Opts(), hierarchySource: null, subscriber: sub); + await driver.InitializeAsync("{}", CancellationToken.None); + var last = new System.Collections.Concurrent.ConcurrentDictionary(); + driver.OnDataChange += (_, e) => last[e.FullReference] = e.Snapshot.Value; + + await driver.SubscribeAsync(["Tank01.Level"], TimeSpan.Zero, CancellationToken.None); + await TestWait.UntilAsync(() => last.TryGetValue("Tank01.Level", out var v) && v is true, + "positive control: the subscribe's initial value arrives", () => Describe(driver, sub)); + + sub.ScanStateRunning = false; // the fake's next initial value + await driver.InvokeReplayForTestAsync(CancellationToken.None); + + await TestWait.UntilAsync(() => last.TryGetValue("Tank01.Level", out var v) && v is false, + "the replay's re-advise must deliver its new initial value", + () => Describe(driver, sub) + $" last=[{string.Join(",", last.Select(kv => $"{kv.Key}={kv.Value}"))}] parked={driver.ParkedEarlyEventCountForTests} h2=[{string.Join(",", driver.SubscriptionsForTests.ResolveSubscribers(2).Select(b => b.FullReference))}] h1=[{string.Join(",", driver.SubscriptionsForTests.ResolveSubscribers(1).Select(b => b.FullReference))}]"); + sub.DroppedInitialValues.ShouldBe(0); + } + /// /// A probe sync on a driver whose pump was never started (no platform source, e.g. the /// DiscoverAsync path) still starts it before advising — the watcher's pre-subscribe hook, not just From 8653960ea3c841787606947d4475e9f751eeb443 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Wed, 11 Nov 2026 13:07:26 -0500 Subject: [PATCH 6/9] fix(galaxy): #865 review round 4 - replay fails on a dead stream, backoff-aware wait - ReplayAsync fails when its restarted pump's stream has already ended, so the supervisor backs off and stays Degraded instead of going Healthy over a dead stream (whose next advise re-report restarted at attempt 1 with no backoff - a tight reconnect loop). - ReconnectSupervisor.AttemptInProgress (reopen through replay); advises wait for a stale pump's replacement only while an attempt is running, not for a whole backoff (the state still reads Reopening/Replaying there). - ReconnectSupervisor: _disposed is set under the state lock and re-checked inside ReportTransportFailure, so a late report cannot start a recovery loop that StopAsync already missed. - EventPump.HasEnded is a flag set before StreamRequested is released, and a stream that ends/faults on its first MoveNext is marked ended first. - SubscriptionRegistry fast path: a miss now falls through to the locked path (lookup + session + window check atomically) instead of a lock-free re-lookup. - Tests: replay over an immediately-ending stream fails; AttemptInProgress across attempt/backoff; ReportTransportFailure after stop; the reopen-window test uses a generous wait bound. Claude-Session: https://claude.ai/code/session_01GT67AD99zQQqpLCVm3p7xk --- .../GalaxyDriver.cs | 21 ++++++-- .../Runtime/EventPump.cs | 21 +++++++- .../Runtime/ReconnectSupervisor.cs | 36 +++++++++++-- .../Runtime/SubscriptionRegistry.cs | 16 ++---- .../Runtime/ProbeInitialValueTests.cs | 34 +++++++++++++ .../Runtime/ReconnectSupervisorTests.cs | 50 +++++++++++++++++++ 6 files changed, 156 insertions(+), 22 deletions(-) diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/GalaxyDriver.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/GalaxyDriver.cs index c9dae1420..8800aa65f 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/GalaxyDriver.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/GalaxyDriver.cs @@ -1056,7 +1056,19 @@ public sealed class GalaxyDriver // #865: every re-advise below pushes its initial value once, to whatever stream is attached. EventPump? replayPump; lock (_pumpLock) replayPump = _eventPump; - if (replayPump is not null) await AwaitCurrentEventStreamAsync(cancellationToken).ConfigureAwait(false); + 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 }) + { + 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). @@ -2279,7 +2291,7 @@ public sealed class GalaxyDriver } /// - /// True when a stale pump will be replaced shortly: the supervisor is reopening/replaying, or there is + /// 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() @@ -2287,8 +2299,9 @@ public sealed class GalaxyDriver if (Volatile.Read(ref _pumpRestarting)) return true; var supervisor = _supervisor; if (supervisor is null) return true; - var state = supervisor.CurrentState; - return state is ReconnectSupervisor.State.Reopening or ReconnectSupervisor.State.Replaying; + // 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). 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 dc8905f85..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 @@ -155,7 +155,11 @@ internal sealed class EventPump : IAsyncDisposable internal int Session => _session; /// True once the stream loop has exited (faulted, completed by the gateway, or disposed). - internal bool HasEnded => _loop is { IsCompleted: true }; + 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); @@ -212,7 +216,19 @@ internal sealed class EventPump : IAsyncDisposable // 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(); + 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; @@ -274,6 +290,7 @@ 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) 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..41c1e5a52 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,15 @@ public sealed class ReconnectSupervisor : IDisposable get { lock (_stateLock) return _state; } } + private bool _attemptInProgress; + + /// + /// 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,6 +145,9 @@ 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) { @@ -208,13 +220,21 @@ 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 @@ -234,6 +254,10 @@ public sealed class ReconnectSupervisor : IDisposable lock (_stateLock) { _lastError = ex.Message; } continue; // back to backoff + retry } + finally + { + Volatile.Write(ref _attemptInProgress, false); + } // === Done === lock (_stateLock) @@ -282,12 +306,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 95e6f83d9..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 @@ -503,17 +503,11 @@ internal sealed class SubscriptionRegistry subscribers = ResolveSubscribers(ev.ItemHandle); if (subscribers.Count > 0) return EarlyEventOutcome.Resolved; AfterFastPathMissForTests?.Invoke(); - if (Volatile.Read(ref _pendingSubscribes) == 0) - { - // #865 (found by its stress run): the lookup above and this read are not atomic. A - // subscribe can register the handle AND close its window in between — then "no window - // open" does not mean "stale", and dropping here lost the initial value for good (the - // first event of a bulk advise, while the dispatch thread was slow). The window closes - // only after its registration, so once no window is open a re-lookup sees every - // registration that window made; nothing is held, so no older event can be overtaken. - subscribers = ResolveSubscribers(ev.ItemHandle); - return subscribers.Count > 0 ? EarlyEventOutcome.Resolved : EarlyEventOutcome.Stale; - } + // #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/Runtime/ProbeInitialValueTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/ProbeInitialValueTests.cs index 36f152888..7eabd47ab 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/ProbeInitialValueTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/ProbeInitialValueTests.cs @@ -51,6 +51,11 @@ public sealed class ProbeInitialValueTests /// 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; @@ -90,6 +95,11 @@ public sealed class ProbeInitialValueTests 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)) @@ -208,6 +218,9 @@ public sealed class ProbeInitialValueTests 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( @@ -315,6 +328,27 @@ public sealed class ProbeInitialValueTests 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"); + } + /// /// 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 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..b99f3b43c 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,56 @@ 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: 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() From d80bdf63690d13c7858ac3f7b2c9eb91929e290b Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Wed, 11 Nov 2026 13:12:33 -0500 Subject: [PATCH 7/9] fix(galaxy): #865 review round 5 - a failure during the replay fails the attempt ReconnectSupervisor swallowed a transport failure reported while Replaying, so a restarted event stream refused/ended a round trip after the replay began still let the replay go Healthy over a dead stream (and the next advise's re-report restarted at attempt 1 with no backoff). A failure reported during Replaying is now remembered and turns that attempt's success into a retry with backoff. ReplayAsync's synchronous dead-pump throw yields to a teardown's cancellation. Claude-Session: https://claude.ai/code/session_01GT67AD99zQQqpLCVm3p7xk --- .../GalaxyDriver.cs | 4 +++ .../Runtime/ReconnectSupervisor.cs | 33 ++++++++++++++++--- .../Runtime/ReconnectSupervisorTests.cs | 29 ++++++++++++++++ 3 files changed, 61 insertions(+), 5 deletions(-) diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/GalaxyDriver.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/GalaxyDriver.cs index 8800aa65f..2cfc682e4 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/GalaxyDriver.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/GalaxyDriver.cs @@ -1065,6 +1065,10 @@ public sealed class GalaxyDriver 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."); } 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 41c1e5a52..43ab2d920 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 @@ -106,6 +106,9 @@ public sealed class ReconnectSupervisor : IDisposable 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 @@ -151,7 +154,10 @@ public sealed class ReconnectSupervisor : IDisposable _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; } @@ -241,7 +247,11 @@ public sealed class ReconnectSupervisor : IDisposable } // === Replaying phase === - lock (_stateLock) TransitionLocked(State.Replaying, _lastError); + lock (_stateLock) + { + _faultDuringReplay = false; + TransitionLocked(State.Replaying, _lastError); + } try { @@ -259,11 +269,24 @@ public sealed class ReconnectSupervisor : IDisposable Volatile.Write(ref _attemptInProgress, false); } - // === Done === + // === Done — unless the transport failed again while the replay ran === + bool faultedDuringReplay; lock (_stateLock) { - _lastError = null; - TransitionLocked(State.Healthy, null); + faultedDuringReplay = _faultDuringReplay; + 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}) — retrying", attempt); + continue; } _logger.LogInformation("Galaxy reconnect succeeded after {Attempt} attempt(s)", attempt); return; 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 b99f3b43c..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 @@ -49,6 +49,35 @@ public sealed class ReconnectSupervisorTests 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() From 13cc7b8353d8e2ea8ffb7e9bcc50fae81726b691 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Wed, 11 Nov 2026 13:16:01 -0500 Subject: [PATCH 8/9] fix(galaxy): #865 review round 6 - ignore stream faults from a superseded pump Each EventPump's fault callback now carries the pump's identity; a fault from a pump that is no longer the live one (or belongs to an older registry session) is logged at Debug and not escalated, so a stale pump's late fault cannot fail a good replay now that a failure during Replaying fails the attempt. The replay-retry warning includes the fault. Claude-Session: https://claude.ai/code/session_01GT67AD99zQQqpLCVm3p7xk --- .../GalaxyDriver.cs | 24 ++++++++++++++++--- .../Runtime/ReconnectSupervisor.cs | 5 +++- 2 files changed, 25 insertions(+), 4 deletions(-) diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/GalaxyDriver.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/GalaxyDriver.cs index 2cfc682e4..e548a138a 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/GalaxyDriver.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/GalaxyDriver.cs @@ -2162,12 +2162,16 @@ 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; @@ -2441,7 +2445,7 @@ 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 void OnEventPumpStreamFault(EventPump? source, Exception cause) { if (_runtimeStopping) { @@ -2451,6 +2455,20 @@ 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 (its session was reopened; the replay replaces or + // already replaced it). Its fault says nothing about the current transport — and reported + // while Replaying it would fail a good replay (#865 review). + _logger.LogDebug(cause, "GalaxyDriver {InstanceId} ignored a stream fault from a superseded event pump.", _driverInstanceId); + return; + } + } + var supervisor = _supervisor; if (supervisor is null) { 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 43ab2d920..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 @@ -271,9 +271,11 @@ public sealed class ReconnectSupervisor : IDisposable // === Done — unless the transport failed again while the replay ran === bool faultedDuringReplay; + string? replayFault; lock (_stateLock) { faultedDuringReplay = _faultDuringReplay; + replayFault = _lastError; if (!faultedDuringReplay) { _lastError = null; @@ -285,7 +287,8 @@ public sealed class ReconnectSupervisor : IDisposable // 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}) — retrying", attempt); + "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); From 59d33ab93805617305ebb5fcc838b64d350cf0c8 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Wed, 11 Nov 2026 13:20:49 -0500 Subject: [PATCH 9/9] test(galaxy): #865 review round 7 - pin the superseded-pump fault filter Adds an escalated-stream-fault test seam and a test: the live pump's fault escalates (positive controls before and after), a superseded pump's late fault after a reopen does not; the absence is asserted only once both pumps' callbacks are known to have run (HasEnded). Verified to fail with the filter disabled. The filter's comment now says the identity check does the work and warns about a future session bump while Healthy. Claude-Session: https://claude.ai/code/session_01GT67AD99zQQqpLCVm3p7xk --- .../GalaxyDriver.cs | 14 ++++-- .../Runtime/ProbeInitialValueTests.cs | 45 +++++++++++++++++++ 2 files changed, 56 insertions(+), 3 deletions(-) diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/GalaxyDriver.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/GalaxyDriver.cs index e548a138a..fb301e855 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/GalaxyDriver.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/GalaxyDriver.cs @@ -2445,6 +2445,11 @@ public sealed class GalaxyDriver /// replay; without this hand-off a transient transport drop permanently kills /// the event stream and GetHealth() keeps reporting Healthy. /// + 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) @@ -2461,13 +2466,16 @@ public sealed class GalaxyDriver lock (_pumpLock) stale = !ReferenceEquals(_eventPump, source) || source.Session != _subscriptions.CurrentSession; if (stale) { - // A pump that is no longer the live one (its session was reopened; the replay replaces or - // already replaced it). Its fault says nothing about the current transport — and reported - // while Replaying it would fail a good replay (#865 review). + // 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/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/ProbeInitialValueTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/ProbeInitialValueTests.cs index 7eabd47ab..e496d7fae 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/ProbeInitialValueTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/ProbeInitialValueTests.cs @@ -116,6 +116,12 @@ public sealed class ProbeInitialValueTests } } + /// 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]; @@ -349,6 +355,45 @@ public sealed class ProbeInitialValueTests "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