From 988046cbe41c42b61f7607e8ed97052efdcaf6ab Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Wed, 11 Nov 2026 10:00:44 -0500 Subject: [PATCH] fix(galaxy): WatchedPlatforms cannot show a platform before its probe binding is routable (#856) PerPlatformProbeWatcher records a platform's handle and publishes the probe binding into the driver registry as one step under _syncLock, but WatchedPlatforms read the handle map lock-free - so it could report a platform one step ahead of the registry Upsert. A ScanState event raised in that window was dispatched by the EventPump, missed ResolveSubscribers and was dropped as "stale after unsubscribe", leaving the host status empty. That is the #856 flake: A_fetch_completing_after_init_populates_host_statuses waited for WatchedPlatforms, emitted ScanState, and timed out at 10 s on the final Running wait. Reproduced 11/700 (28 parallel runners x 25 + 14 CPU burners), every failure at that step with the binding present by the time of diagnosis and hostStatuses=[] - i.e. the event was dropped. 0/1400 after. WatchedPlatforms now reads under _syncLock (test seam only; the dispatch path already checks the map after the registry, the safe order). New PerPlatformProbeWatcherVisibilityTests spins a reader on another thread and asserts the registry entry exists the moment the platform is visible (old code: torn in 1691/2000 iterations). WaitUntilAsync gains a describe overload; the #856 test's waits now report elapsed time and the observed seed/binding/pump/host-status chain on timeout. Claude-Session: https://claude.ai/code/session_01GT67AD99zQQqpLCVm3p7xk --- .../Health/PerPlatformProbeWatcher.cs | 13 ++++- .../PerPlatformProbeWatcherVisibilityTests.cs | 55 +++++++++++++++++++ .../Runtime/GalaxyDriverFixRound1Tests.cs | 18 +++++- .../Runtime/GalaxyDriverFixRound2Tests.cs | 22 +++++++- 4 files changed, 101 insertions(+), 7 deletions(-) create mode 100644 tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Health/PerPlatformProbeWatcherVisibilityTests.cs 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 64080d524..0e9433b77 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 @@ -192,8 +192,17 @@ public sealed class PerPlatformProbeWatcher : IDisposable } } - /// Snapshot of platform tag names currently watched. - public IReadOnlyCollection WatchedPlatforms => [.. _itemHandlesByPlatform.Keys]; + /// + /// Snapshot of platform tag names currently watched. Read under the sync lock so a platform + /// is never visible here before has published its probe + /// binding into the driver registry (the record and the publish are one locked step): a + /// caller that sees the platform and then raises a ScanState event must find it routable, + /// not dropped by the pump as "stale after unsubscribe" (#856). + /// + public IReadOnlyCollection WatchedPlatforms + { + get { lock (_syncLock) return [.. _itemHandlesByPlatform.Keys]; } + } /// /// Reconcile the watched platform set against . diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Health/PerPlatformProbeWatcherVisibilityTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Health/PerPlatformProbeWatcherVisibilityTests.cs new file mode 100644 index 000000000..0dc3fb82d --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Health/PerPlatformProbeWatcherVisibilityTests.cs @@ -0,0 +1,55 @@ +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Health; +using ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Runtime; +using static ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests.Runtime.GalaxyDriverFixRound1Tests; + +namespace ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests.Health; + +/// +/// #856: a platform must never be observable in +/// before its probe binding resolves through the driver registry. The watcher records the handle +/// and then publishes the binding inside one lock, but WatchedPlatforms used to read the +/// handle map lock-free — so a caller could see the platform, raise a ScanState event, and have the +/// pump drop it as "stale after unsubscribe" (host status stuck at Unknown). That is what made +/// A_fetch_completing_after_init_populates_host_statuses time out under load. +/// +public sealed class PerPlatformProbeWatcherVisibilityTests +{ + /// + /// A reader spinning on WatchedPlatforms from another thread checks, at the first moment + /// it sees the platform, that the registry already holds the probe entry. Before the fix this + /// caught the torn state in ~85% of iterations; after it, the publish is part of what the + /// reader's lock waits for, so the torn state is unobservable. + /// + [Fact] + public async Task A_platform_visible_as_watched_already_resolves_through_the_registry() + { + const int iterations = 500; + var ct = TestContext.Current.CancellationToken; + var torn = 0; + for (var i = 0; i < iterations; i++) + { + var registry = new SubscriptionRegistry(); + var sub = new ParkableSubscriber(); + using var watcher = new PerPlatformProbeWatcher(sub, new HostStatusAggregator(), registry); + using var readerRunning = new ManualResetEventSlim(); + var observed = Task.Run(() => + { + readerRunning.Set(); + var deadline = DateTime.UtcNow.AddMilliseconds(WaitMs); + while (watcher.WatchedPlatforms.Count == 0) + { + if (DateTime.UtcNow > deadline) throw new ShouldAssertException("the sync never recorded the platform"); + } + return registry.Contains(SubscriptionRegistry.ProbeSubscriptionId); + }, ct); + readerRunning.Wait(ct); + await Task.Run(() => watcher.SyncPlatformsAsync(["P"], CancellationToken.None), ct); + if (!await observed) torn++; + } + + torn.ShouldBe(0, + $"{torn}/{iterations} iterations saw a watched platform whose probe binding was not yet in the registry"); + } +} 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 a55d79b05..072382229 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 @@ -140,12 +140,24 @@ public sealed class GalaxyDriverFixRound1Tests }); } - internal static async Task WaitUntilAsync(Func condition, string because) + internal static Task WaitUntilAsync(Func condition, string because) => + WaitUntilAsync(condition, because, describe: null); + + /// + /// Presence wait: polls for up to . On timeout the + /// failure carries the elapsed time and, when given, 's snapshot of the + /// state actually observed — so a flake under load says which link of the chain never happened. + /// + internal static async Task WaitUntilAsync(Func condition, string because, Func? describe) { - var deadline = DateTime.UtcNow.AddMilliseconds(WaitMs); + var sw = System.Diagnostics.Stopwatch.StartNew(); while (!condition()) { - if (DateTime.UtcNow > deadline) throw new ShouldAssertException(because); + if (sw.ElapsedMilliseconds > WaitMs) + { + var observed = describe is null ? "" : $" Observed: {describe()}"; + throw new ShouldAssertException($"{because} (not met after {sw.ElapsedMilliseconds} ms).{observed}"); + } await Task.Delay(10); } } diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/GalaxyDriverFixRound2Tests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/GalaxyDriverFixRound2Tests.cs index 27134ba66..4006afbb9 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/GalaxyDriverFixRound2Tests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/GalaxyDriverFixRound2Tests.cs @@ -171,12 +171,30 @@ public sealed class GalaxyDriverFixRound2Tests platforms.Release(); } + // WatchedPlatforms is read under the watcher's sync lock, so seeing the platform here means its + // probe binding is already in the registry — the ScanState event below cannot be dropped as + // "stale after unsubscribe" (#856: it used to be read lock-free, one step ahead of the publish). await WaitUntilAsync(() => driver.ProbeWatcherForTests?.WatchedPlatforms.Count == 1, - "the late seed must reach the probe watcher"); + "the late seed must reach the probe watcher", () => DescribeSeed(driver, platforms, sub)); await driver.SubscribeAsync(["Tank01.Level"], TimeSpan.Zero, CancellationToken.None); // starts the pump await sub.EmitBoolAsync(sub.HandleByRef["Plat01.ScanState"], true); await WaitUntilAsync(() => StateOf(driver, "Plat01") == HostState.Running, - "a ScanState event must surface the late-seeded platform"); + "a ScanState event must surface the late-seeded platform", () => DescribeSeed(driver, platforms, sub)); + } + + /// Self-diagnosis for the seed → probe binding → pump → host-status chain. + private static string DescribeSeed(GalaxyDriver driver, GatedPlatformSource platforms, ParkableSubscriber sub) + { + var watcher = driver.ProbeWatcherForTests; + string subscribed; + lock (sub.Subscribed) subscribed = string.Join(",", sub.Subscribed); + var watched = watcher is null ? "" : string.Join(",", watcher.WatchedPlatforms); + var bindings = watcher is null ? "" + : string.Join(",", watcher.CurrentBindings.Select(b => $"{b.FullReference}#{b.ItemHandle}")); + var hosts = string.Join(",", driver.GetHostStatuses().Select(h => $"{h.HostName}={h.State}")); + return $"fetchCalls={platforms.Calls}, watched=[{watched}], probeBindings=[{bindings}], " + + $"gwSubscribed=[{subscribed}], pump={driver.HasEventPumpForTests}, " + + $"trackedSubscriptions={driver.TrackedSubscriptionCountForTests}, hostStatuses=[{hosts}]"; } // ───────────────────────────── N1 generation + N2: straddling worker ─────────────────────────────