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
2026-11-11 10:00:44 -05:00
parent 03e5145796
commit 988046cbe4
4 changed files with 101 additions and 7 deletions
@@ -192,8 +192,17 @@ public sealed class PerPlatformProbeWatcher : IDisposable
        }
    }

-    /// <summary>Snapshot of platform tag names currently watched.</summary>
-    public IReadOnlyCollection<string> WatchedPlatforms => [.. _itemHandlesByPlatform.Keys];
+    /// <summary>
+    ///     Snapshot of platform tag names currently watched. Read under the sync lock so a platform
+    ///     is never visible here before <see cref="SubscribeAndRecordAsync"/> 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).
+    /// </summary>
+    public IReadOnlyCollection<string> WatchedPlatforms
+    {
+        get { lock (_syncLock) return [.. _itemHandlesByPlatform.Keys]; }
+    }

    /// <summary>
    ///     Reconcile the watched platform set against <paramref name="platformTagNames"/>.
@@ -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;
+
+/// <summary>
+///     #856: a platform must never be observable in <see cref="PerPlatformProbeWatcher.WatchedPlatforms"/>
+///     before its probe binding resolves through the driver registry. The watcher records the handle
+///     and then publishes the binding inside one lock, but <c>WatchedPlatforms</c> 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
+///     <c>A_fetch_completing_after_init_populates_host_statuses</c> time out under load.
+/// </summary>
+public sealed class PerPlatformProbeWatcherVisibilityTests
+{
+    /// <summary>
+    ///     A reader spinning on <c>WatchedPlatforms</c> 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.
+    /// </summary>
+    [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");
+    }
+}
@@ -140,12 +140,24 @@ public sealed class GalaxyDriverFixRound1Tests
        });
    }

-    internal static async Task WaitUntilAsync(Func<bool> condition, string because)
+    internal static Task WaitUntilAsync(Func<bool> condition, string because) =>
+        WaitUntilAsync(condition, because, describe: null);
+
+    /// <summary>
+    ///     Presence wait: polls <paramref name="condition"/> for up to <see cref="WaitMs"/>. On timeout the
+    ///     failure carries the elapsed time and, when given, <paramref name="describe"/>'s snapshot of the
+    ///     state actually observed — so a flake under load says which link of the chain never happened.
+    /// </summary>
+    internal static async Task WaitUntilAsync(Func<bool> condition, string because, Func<string>? 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);
        }
    }
@@ -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));
+    }
+
+    /// <summary>Self-diagnosis for the seed → probe binding → pump → host-status chain.</summary>
+    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 ? "<no watcher>" : string.Join(",", watcher.WatchedPlatforms);
+        var bindings = watcher is null ? "<no watcher>"
+            : 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 ─────────────────────────────