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 ─────────────────────────────