From 4fdc3708f4567cd345aad56ead73268a3d2e27c4 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Wed, 11 Nov 2026 10:18:22 -0500 Subject: [PATCH] test(galaxy): presence waits survive a starved/loaded runner; absence asserted behind positive controls (#858) Under heavy parallel load (14 concurrent test processes + 14 CPU burners, with DOTNET_ThreadPool_ForceMinWorkerThreads=1 as the reproducer) the Galaxy tests with 1-3 s fixed waits failed 16/70 runs. Every failure was a background loop started with Task.Run (alarm feed, DeployWatcher, EventPump, ReconnectSupervisor) not being SCHEDULED within the ceiling - e.g. DeployWatcher's stream had not even been opened after 2 s. No production race: the code under test was checked for each (alarm feed dispatch is synchronous in-order, DeployWatcher/ReconnectSupervisor state is lock- or loop-local, subscribe registers before returning). - New shared TestWait (UntilAsync / ForAsync / HealthyAsync): 30 s presence ceiling, failure reports elapsed ms + an observed-state snapshot (#856 shape; GalaxyDriverFixRound1Tests.WaitUntilAsync now delegates to it). - Converted the short waits in GatewayGalaxyAlarmFeedTests, DeployWatcherTests, GalaxyDriverSubscribeTests, GalaxyDriverRawPathKeyingTests (its WaitForAsync never threw on timeout), EventPumpStreamFaultTests, EventPumpReplayGapTests, ReconnectSupervisorTests; FixRound2 InitBudget 2 s -> 10 s (the parked fetch never completes, so any finite bound keeps the verdict). - Absence by short sleep replaced with positive controls: DeployWatcher bootstrap and same-deploy-time (follow with a real change, assert exactly that one raise - verified red when suppression is removed), subscribe failed-tag / unsubscribed handle (follow with a live handle's event), faulted pump (wait for its onStreamFault instead of 100 ms). - Fixed sleeps that raced scheduling: subscribe-once read tests wait for the read's registration (TrackedSubscriptionCountForTests) instead of 50 ms; supervisor Dispose test waits for reopen entry instead of 50 ms; the already-Healthy WaitForHealthy test asserts synchronous completion instead of a wall-clock bound. FakeSubscriber.Map is now a ConcurrentDictionary. Loaded run (14 parallel x 5, 14 burners, ForceMinWorkerThreads=1): before 16/70 failed runs; after 0/350 (70 + 210 + 70). Without the thread-pool override: 0/70 before and after on this host. Serial: 472 passed, 8 skipped. Claude-Session: https://claude.ai/code/session_01GT67AD99zQQqpLCVm3p7xk --- .../Browse/DeployWatcherTests.cs | 78 ++++++++----- .../Runtime/EventPumpReplayGapTests.cs | 21 ++-- .../Runtime/EventPumpStreamFaultTests.cs | 26 ++--- .../Runtime/GalaxyDriverFixRound1Tests.cs | 21 +--- .../Runtime/GalaxyDriverFixRound2Tests.cs | 4 +- .../Runtime/GalaxyDriverRawPathKeyingTests.cs | 14 +-- .../Runtime/GalaxyDriverReadTests.cs | 8 +- .../Runtime/GalaxyDriverSubscribeTests.cs | 58 ++++++---- .../Runtime/GatewayGalaxyAlarmFeedTests.cs | 15 +-- .../Runtime/ReconnectSupervisorTests.cs | 37 +++--- .../TestWait.cs | 107 ++++++++++++++++++ 11 files changed, 256 insertions(+), 133 deletions(-) create mode 100644 tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/TestWait.cs diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Browse/DeployWatcherTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Browse/DeployWatcherTests.cs index f808bf645..9a7495aac 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Browse/DeployWatcherTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Browse/DeployWatcherTests.cs @@ -108,15 +108,23 @@ public sealed class DeployWatcherTests return captured; } - private static async Task WaitUntilAsync(Func condition, TimeSpan timeout) + // Presence wait with a generous ceiling and a self-describing failure (#858): under heavy parallel + // load the watcher's Task.Run loop alone could take > 2 s to be scheduled. + private static Task WaitUntilAsync( + Func condition, string because, FakeDeployWatchSource source, List captured) + => TestWait.UntilAsync(condition, because, () => Describe(source, captured)); + + private static string Describe(FakeDeployWatchSource source, List captured) { - var deadline = DateTimeOffset.UtcNow + timeout; - while (DateTimeOffset.UtcNow < deadline) + lock (captured) { - if (condition()) return; - await Task.Delay(10).ConfigureAwait(false); + return $"watchCalls={source.CallCount} raised=[{string.Join(", ", captured.Select(c => $"{c.Reason}@{c.ScopeHint}"))}]"; } - throw new TimeoutException("Condition was not met within timeout."); + } + + private static List Snapshot(List captured) + { + lock (captured) return [.. captured]; } /// Verifies that bootstrap deploy events are suppressed. @@ -128,16 +136,22 @@ public sealed class DeployWatcherTests using var watcher = new DeployWatcher(source); var captured = CaptureRediscoverEvents(watcher); + var t0 = DateTimeOffset.Parse("2026-01-01T00:00:00Z"); + var t1 = DateTimeOffset.Parse("2026-01-03T00:00:00Z"); + await watcher.StartAsync(CancellationToken.None); - // Push only the bootstrap event. - await channel.Writer.WriteAsync(Event(0, DateTimeOffset.Parse("2026-01-01T00:00:00Z")), TestContext.Current.CancellationToken); + // The bootstrap event, then a positive control that MUST raise. Events are handled in order, + // so once the control's raise is seen the bootstrap has been fully handled - no sleep needed + // to prove the bootstrap did not raise (#858). + await channel.Writer.WriteAsync(Event(0, t0), TestContext.Current.CancellationToken); + await channel.Writer.WriteAsync(Event(1, t1), TestContext.Current.CancellationToken); - // Give the loop a moment to consume + ack. - await WaitUntilAsync(() => source.CallCount > 0 && channel.Reader.Count == 0, TimeSpan.FromSeconds(2)); - await Task.Delay(50, TestContext.Current.CancellationToken); + await WaitUntilAsync(() => Snapshot(captured).Count >= 1, + "positive control: the post-bootstrap change must raise rediscovery", source, captured); - captured.ShouldBeEmpty(); + var raised = Snapshot(captured).ShouldHaveSingleItem("the bootstrap event must not raise"); + DateTimeOffset.Parse(raised.ScopeHint!).ToUniversalTime().ShouldBe(t1.ToUniversalTime()); await watcher.StopAsync(); } @@ -159,7 +173,8 @@ public sealed class DeployWatcherTests await channel.Writer.WriteAsync(Event(0, t0), TestContext.Current.CancellationToken); // bootstrap await channel.Writer.WriteAsync(Event(1, t1), TestContext.Current.CancellationToken); // real change - await WaitUntilAsync(() => captured.Count >= 1, TimeSpan.FromSeconds(2)); + await WaitUntilAsync(() => Snapshot(captured).Count >= 1, + "a deploy-time change must raise rediscovery", source, captured); captured.Count.ShouldBe(1); captured[0].Reason.ShouldBe("deploy-time-changed"); @@ -183,13 +198,18 @@ public sealed class DeployWatcherTests await watcher.StartAsync(CancellationToken.None); + var t1 = DateTimeOffset.Parse("2026-01-05T00:00:00Z"); await channel.Writer.WriteAsync(Event(0, t0), TestContext.Current.CancellationToken); // bootstrap await channel.Writer.WriteAsync(Event(2, t0), TestContext.Current.CancellationToken); // duplicate state — gateway re-sent + // Positive control (#858): a genuine change after the duplicate MUST raise. Events are handled in + // order, so once it is seen the duplicate has been handled - its absence is then a real result. + await channel.Writer.WriteAsync(Event(3, t1), TestContext.Current.CancellationToken); - await WaitUntilAsync(() => channel.Reader.Count == 0, TimeSpan.FromSeconds(2)); - await Task.Delay(50, TestContext.Current.CancellationToken); + await WaitUntilAsync(() => Snapshot(captured).Count >= 1, + "positive control: the change after the duplicate must raise rediscovery", source, captured); - captured.ShouldBeEmpty(); + var raised = Snapshot(captured).ShouldHaveSingleItem("the duplicate deploy time must not raise"); + DateTimeOffset.Parse(raised.ScopeHint!).ToUniversalTime().ShouldBe(t1.ToUniversalTime()); await watcher.StopAsync(); } @@ -212,7 +232,8 @@ public sealed class DeployWatcherTests // Now a deploy lands and the present flag flips. await channel.Writer.WriteAsync(Event(1, t1), TestContext.Current.CancellationToken); - await WaitUntilAsync(() => captured.Count >= 1, TimeSpan.FromSeconds(2)); + await WaitUntilAsync(() => Snapshot(captured).Count >= 1, + "a presence flip must raise rediscovery", source, captured); captured.Count.ShouldBe(1); captured[0].Reason.ShouldBe("deploy-time-changed"); @@ -233,13 +254,10 @@ public sealed class DeployWatcherTests // Push bootstrap so the loop enters its enumeration body before stop. await channel.Writer.WriteAsync(Event(0, DateTimeOffset.UtcNow), TestContext.Current.CancellationToken); - await WaitUntilAsync(() => source.CallCount > 0, TimeSpan.FromSeconds(2)); + await WaitUntilAsync(() => source.CallCount > 0, "the watch stream must open", source, []); - // StopAsync should complete without throwing and within a reasonable window. - var stopTask = watcher.StopAsync(); - var completed = await Task.WhenAny(stopTask, Task.Delay(TimeSpan.FromSeconds(5), TestContext.Current.CancellationToken)); - completed.ShouldBe(stopTask); - await stopTask; // observe (no) exception + // StopAsync should complete without throwing (a hang fails at the presence ceiling). + await TestWait.ForAsync(watcher.StopAsync(), "StopAsync must cancel the loop and return"); } /// Verifies that disposing stops a running watcher. @@ -252,13 +270,11 @@ public sealed class DeployWatcherTests await watcher.StartAsync(CancellationToken.None); await channel.Writer.WriteAsync(Event(0, DateTimeOffset.UtcNow), TestContext.Current.CancellationToken); - await WaitUntilAsync(() => source.CallCount > 0, TimeSpan.FromSeconds(2)); + await WaitUntilAsync(() => source.CallCount > 0, "the watch stream must open", source, []); // Should not throw, should not hang. - var disposeTask = Task.Run(watcher.Dispose, TestContext.Current.CancellationToken); - var completed = await Task.WhenAny(disposeTask, Task.Delay(TimeSpan.FromSeconds(5), TestContext.Current.CancellationToken)); - completed.ShouldBe(disposeTask); - await disposeTask; + await TestWait.ForAsync(Task.Run(watcher.Dispose, TestContext.Current.CancellationToken), + "Dispose must stop the loop and return"); } /// Verifies that a source exception triggers retry with backoff. @@ -292,13 +308,15 @@ public sealed class DeployWatcherTests await watcher.StartAsync(CancellationToken.None); // Wait for the second iteration (post-retry) to start. - await WaitUntilAsync(() => source.CallCount >= 2, TimeSpan.FromSeconds(2)); + await WaitUntilAsync(() => source.CallCount >= 2, + "the watcher must re-open the stream after the source threw", source, captured); // Now feed bootstrap + real event into the second channel. await secondChannel.Writer.WriteAsync(Event(0, t0), TestContext.Current.CancellationToken); await secondChannel.Writer.WriteAsync(Event(1, t1), TestContext.Current.CancellationToken); - await WaitUntilAsync(() => captured.Count >= 1, TimeSpan.FromSeconds(2)); + await WaitUntilAsync(() => Snapshot(captured).Count >= 1, + "the re-opened stream's change must raise rediscovery", source, captured); captured.Count.ShouldBe(1); captured[0].Reason.ShouldBe("deploy-time-changed"); diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/EventPumpReplayGapTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/EventPumpReplayGapTests.cs index e43211c18..255e7a89d 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/EventPumpReplayGapTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/EventPumpReplayGapTests.cs @@ -30,7 +30,6 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests.Runtime; /// public sealed class EventPumpReplayGapTests { - private const int WaitMs = 2_000; private static MxEvent GapSentinel(ulong requestedAfter, ulong oldestAvailable) => new() { @@ -57,8 +56,7 @@ public sealed class EventPumpReplayGapTests await subscriber.EmitAsync(GapSentinel(requestedAfter: 100, oldestAvailable: 250)); - (await Task.WhenAny(observed.Task, Task.Delay(WaitMs, TestContext.Current.CancellationToken))).ShouldBe(observed.Task, - "a replay-gap sentinel must reach the driver so it can re-snapshot"); + await TestWait.ForAsync(observed.Task, "a replay-gap sentinel must reach the driver so it can re-snapshot"); var gap = await observed.Task; gap.RequestedAfterSequence.ShouldBe(100UL); gap.OldestAvailableSequence.ShouldBe(250UL); @@ -93,7 +91,7 @@ public sealed class EventPumpReplayGapTests pump.Start(); await subscriber.EmitAsync(GapSentinel(requestedAfter: 1, oldestAvailable: 2)); - (await Task.WhenAny(gapSeen.Task, Task.Delay(WaitMs, TestContext.Current.CancellationToken))).ShouldBe(gapSeen.Task); + await TestWait.ForAsync(gapSeen.Task, "the replay-gap sentinel must reach the gap callback"); // A positive control: follow the sentinel with a real event on the same handle. Without // it, "nothing was dispatched" would also pass on a pump that dispatches nothing at all. @@ -106,8 +104,7 @@ public sealed class EventPumpReplayGapTests SourceTimestamp = Timestamp.FromDateTime(DateTime.UtcNow), }); - (await Task.WhenAny(realValue.Task, Task.Delay(WaitMs, TestContext.Current.CancellationToken))).ShouldBe(realValue.Task, - "the pump must keep dispatching real events after a sentinel"); + await TestWait.ForAsync(realValue.Task, "the pump must keep dispatching real events after a sentinel"); (await realValue.Task).Snapshot.Value.ShouldBe(12.5); Volatile.Read(ref dispatched).ShouldBe(1, "only the real event may be dispatched, never the sentinel"); } @@ -149,8 +146,7 @@ public sealed class EventPumpReplayGapTests await subscriber.EmitAsync(GapSentinel(requestedAfter: 5, oldestAvailable: 900)); - (await Task.WhenAny(observed.Task, Task.Delay(WaitMs, TestContext.Current.CancellationToken))).ShouldBe(observed.Task, - "the sentinel must not be droppable by channel back-pressure"); + await TestWait.ForAsync(observed.Task, "the sentinel must not be droppable by channel back-pressure"); (await observed.Task).OldestAvailableSequence.ShouldBe(900UL); blockDispatch.Set(); @@ -187,8 +183,7 @@ public sealed class EventPumpReplayGapTests Quality = 192, }); - (await Task.WhenAny(resumed.Task, Task.Delay(WaitMs, TestContext.Current.CancellationToken))).ShouldBe(resumed.Task, - "a failed re-snapshot must not stop the stream that is still delivering"); + await TestWait.ForAsync(resumed.Task, "a failed re-snapshot must not stop the stream that is still delivering"); faulted.ShouldBeFalse("a throwing gap callback is not a transport fault"); } @@ -217,8 +212,7 @@ public sealed class EventPumpReplayGapTests subscriber.CompleteStream(); - (await Task.WhenAny(faultObserved.Task, Task.Delay(WaitMs, TestContext.Current.CancellationToken))).ShouldBe(faultObserved.Task, - "a server-completed stream is as fatal as a faulted one and must reach the supervisor"); + await TestWait.ForAsync(faultObserved.Task, "a server-completed stream is as fatal as a faulted one and must reach the supervisor"); (await faultObserved.Task).Message.ShouldContain("closed or evicted"); } @@ -277,7 +271,8 @@ public sealed class EventPumpReplayGapTests SourceTimestamp = Timestamp.FromDateTime(DateTime.UtcNow), }); - (await Task.WhenAny(bothSeen.Task, Task.Delay(WaitMs, TestContext.Current.CancellationToken))).ShouldBe(bothSeen.Task); + await TestWait.ForAsync(bothSeen.Task, "both the undecodable and the control event must be dispatched", + () => { lock (received) return $"dispatched={received.Count}/2"; }); lock (received) { diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/EventPumpStreamFaultTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/EventPumpStreamFaultTests.cs index e976f4d63..094c39696 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/EventPumpStreamFaultTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/EventPumpStreamFaultTests.cs @@ -16,8 +16,6 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests.Runtime; /// public sealed class EventPumpStreamFaultTests { - private const int WaitMs = 2_000; - /// Verifies that stream fault invokes the callback with the exception. [Fact] public async Task StreamFault_InvokesOnStreamFaultCallback_WithTheCause() @@ -35,9 +33,7 @@ public sealed class EventPumpStreamFaultTests // Drop the gw stream — RunAsync's await foreach throws. subscriber.FaultStream(new IOException("simulated gateway transport drop")); - var completed = await Task.WhenAny(faultObserved.Task, Task.Delay(WaitMs, TestContext.Current.CancellationToken)); - completed.ShouldBe(faultObserved.Task, - "EventPump must invoke onStreamFault when the gw StreamEvents stream faults"); + await TestWait.ForAsync(faultObserved.Task, "EventPump must invoke onStreamFault when the gw StreamEvents stream faults"); (await faultObserved.Task).ShouldBeOfType(); } @@ -69,12 +65,10 @@ public sealed class EventPumpStreamFaultTests subscriber.FaultStream(new IOException("simulated gateway transport drop")); - (await Task.WhenAny(reopenRan.Task, Task.Delay(WaitMs, TestContext.Current.CancellationToken))).ShouldBe(reopenRan.Task, - "stream fault must trigger the supervisor's reopen path"); - (await Task.WhenAny(replayRan.Task, Task.Delay(WaitMs, TestContext.Current.CancellationToken))).ShouldBe(replayRan.Task, - "stream fault must trigger the supervisor's replay path"); + await TestWait.ForAsync(reopenRan.Task, "stream fault must trigger the supervisor's reopen path"); + await TestWait.ForAsync(replayRan.Task, "stream fault must trigger the supervisor's replay path"); - await supervisor.WaitForHealthyAsync(new CancellationTokenSource(WaitMs).Token); + await TestWait.HealthyAsync(supervisor, "the supervisor must recover to Healthy after reopen + replay"); supervisor.IsDegraded.ShouldBeFalse(); } @@ -93,11 +87,15 @@ public sealed class EventPumpStreamFaultTests // --- first pump: faults, then is "restarted" (no-op) and confirmed dead --- var faulted = new FaultingSubscriber(); var staleObserved = false; - var oldPump = new EventPump(faulted, registry, channelCapacity: 8, clientName: "Restart"); + var oldFaulted = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var oldPump = new EventPump(faulted, registry, channelCapacity: 8, clientName: "Restart", + onStreamFault: _ => oldFaulted.TrySetResult()); oldPump.OnDataChange += (_, _) => staleObserved = true; oldPump.Start(); faulted.FaultStream(new IOException("simulated gateway transport drop")); - await Task.Delay(100, TestContext.Current.CancellationToken); + // Wait for the old pump's loop to actually observe the fault (#858: a fixed 100 ms sleep was a + // guess at when the loop had exited). + await TestWait.ForAsync(oldFaulted.Task, "the old pump must observe its stream fault"); // In-place Start() after a fault is a no-op — the loop task is non-null but done. oldPump.Start(); @@ -114,9 +112,7 @@ public sealed class EventPumpStreamFaultTests await resubscribed.EmitAsync(itemHandle: 7, value: 123.0); - var completed = await Task.WhenAny(resumed.Task, Task.Delay(WaitMs, TestContext.Current.CancellationToken)); - completed.ShouldBe(resumed.Task, - "a fresh EventPump created after a fault must resume dispatching OnDataChange"); + await TestWait.ForAsync(resumed.Task, "a fresh EventPump created after a fault must resume dispatching OnDataChange"); (await resumed.Task).FullReference.ShouldBe("Tank.Level"); staleObserved.ShouldBeFalse("the faulted pump must not dispatch after its stream dropped"); } 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 072382229..b22f8747c 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 @@ -144,23 +144,12 @@ public sealed class GalaxyDriverFixRound1Tests 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. + /// Presence wait: polls for up to via the shared + /// . On timeout the failure carries the elapsed time and, when given, + /// 's snapshot of the state actually observed. /// - internal static async Task WaitUntilAsync(Func condition, string because, Func? describe) - { - var sw = System.Diagnostics.Stopwatch.StartNew(); - while (!condition()) - { - 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); - } - } + internal static Task WaitUntilAsync(Func condition, string because, Func? describe) => + TestWait.UntilAsync(condition, because, describe, TimeSpan.FromMilliseconds(WaitMs)); internal static HostState? StateOf(GalaxyDriver driver, string host) => driver.GetHostStatuses().FirstOrDefault(s => s.HostName == host)?.State; 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 4006afbb9..1902aec56 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 @@ -24,7 +24,9 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests.Runtime; /// public sealed class GalaxyDriverFixRound2Tests { - private static readonly TimeSpan InitBudget = TimeSpan.FromSeconds(2); + // The fetches these tests park never complete, so ANY finite budget separates "init does not wait on + // them" from "init hangs"; 10 s keeps that verdict while surviving a heavily loaded runner (#858). + private static readonly TimeSpan InitBudget = TimeSpan.FromSeconds(10); /// A platform source whose fetches park until released (optionally ignoring cancellation). private sealed class GatedPlatformSource : IGalaxyPlatformSource diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/GalaxyDriverRawPathKeyingTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/GalaxyDriverRawPathKeyingTests.cs index 89d49f732..632c07910 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/GalaxyDriverRawPathKeyingTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/GalaxyDriverRawPathKeyingTests.cs @@ -95,7 +95,7 @@ public sealed class GalaxyDriverRawPathKeyingTests "g", OptsWithRawTags(), hierarchySource: null, dataReader: null, dataWriter: null, subscriber: subscriber); var captured = new List(); - driver.OnDataChange += (_, args) => captured.Add(args); + driver.OnDataChange += (_, args) => { lock (captured) captured.Add(args); }; await driver.SubscribeAsync([RawPath], TimeSpan.FromSeconds(1), CancellationToken.None); @@ -104,7 +104,7 @@ public sealed class GalaxyDriverRawPathKeyingTests subscriber.Map.ShouldNotContainKey(RawPath); await subscriber.EmitOnDataChangeAsync(subscriber.Map[AttributeRef], 42.0); - await WaitForAsync(() => captured.Count >= 1); + await GalaxyDriverSubscribeTests.WaitForCapturedAsync(captured, 1, "the value change must be dispatched"); captured.ShouldHaveSingleItem(); captured[0].FullReference.ShouldBe(RawPath); // surfaced under the RawPath identity @@ -148,16 +148,6 @@ public sealed class GalaxyDriverRawPathKeyingTests // ===== helpers / fakes ===== - private static async Task WaitForAsync(Func condition, int timeoutMs = 2000) - { - var deadline = DateTime.UtcNow.AddMilliseconds(timeoutMs); - while (DateTime.UtcNow < deadline) - { - if (condition()) return; - await Task.Delay(10); - } - } - private static GalaxyAttribute Attr(string name, int sec) => new() { AttributeName = name, MxDataType = 2 /*Float32*/, SecurityClassification = sec }; diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/GalaxyDriverReadTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/GalaxyDriverReadTests.cs index ed5368357..9de5c39ba 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/GalaxyDriverReadTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/GalaxyDriverReadTests.cs @@ -109,7 +109,10 @@ public sealed class GalaxyDriverReadTests var readTask = driver.ReadAsync(["Tank.Level"], CancellationToken.None); // Push the "initial value" event the gw would emit immediately after SubscribeBulk. - await Task.Delay(50, TestContext.Current.CancellationToken); // give SubscribeBulk a beat to register + handler to attach + // Wait until the read's bindings are registered (#858: a fixed 50 ms beat was a timing guess — an + // event emitted before registration is dropped by the pump and the read then times out). + await TestWait.UntilAsync(() => driver.TrackedSubscriptionCountForTests == 1, + "the subscribe-once read must register its bindings"); var itemHandle = subscriber.Map["Tank.Level"]; await subscriber.EmitOnDataChangeAsync(itemHandle, 42.0); @@ -131,7 +134,8 @@ public sealed class GalaxyDriverReadTests "g", Opts(), hierarchySource: null, dataReader: null, dataWriter: null, subscriber: subscriber); var readTask = driver.ReadAsync(["Good", "Bad"], CancellationToken.None); - await Task.Delay(50, TestContext.Current.CancellationToken); + await TestWait.UntilAsync(() => driver.TrackedSubscriptionCountForTests == 1, + "the subscribe-once read must register its bindings"); await subscriber.EmitOnDataChangeAsync(subscriber.Map["Good"], 1.0); var result = await readTask; diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/GalaxyDriverSubscribeTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/GalaxyDriverSubscribeTests.cs index 637900ee0..48cf0139e 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/GalaxyDriverSubscribeTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/GalaxyDriverSubscribeTests.cs @@ -1,3 +1,4 @@ +using System.Collections.Concurrent; using System.Threading.Channels; using Google.Protobuf.WellKnownTypes; using ZB.MOM.WW.MxGateway.Contracts.Proto; @@ -29,7 +30,8 @@ public sealed class GalaxyDriverSubscribeTests private readonly Channel _events = Channel.CreateUnbounded(); /// Gets the mapping of tag references to subscription handles. - public Dictionary Map { get; } = new(); + /// Concurrent: the driver may run SubscribeBulk on another thread while a test polls it. + public ConcurrentDictionary Map { get; } = new(); /// Gets the list of unsubscribed handles. public List UnsubscribedHandles { get; } = []; @@ -121,14 +123,14 @@ public sealed class GalaxyDriverSubscribeTests "g", Opts(), hierarchySource: null, dataReader: null, dataWriter: null, subscriber: subscriber); var captured = new List(); - driver.OnDataChange += (_, args) => captured.Add(args); + driver.OnDataChange += (_, args) => { lock (captured) captured.Add(args); }; var handle = await driver.SubscribeAsync(["Tank.Level"], TimeSpan.FromSeconds(1), CancellationToken.None); var itemHandle = subscriber.Map["Tank.Level"]; await subscriber.EmitOnDataChangeAsync(itemHandle, 42.0); - await WaitForAsync(() => captured.Count >= 1); + await WaitForCapturedAsync(captured, 1, "the value change must be dispatched"); captured.Count.ShouldBe(1); captured[0].SubscriptionHandle.ShouldBe(handle); captured[0].FullReference.ShouldBe("Tank.Level"); @@ -144,7 +146,7 @@ public sealed class GalaxyDriverSubscribeTests "g", Opts(), hierarchySource: null, dataReader: null, dataWriter: null, subscriber: subscriber); var captured = new List(); - driver.OnDataChange += (_, args) => captured.Add(args); + driver.OnDataChange += (_, args) => { lock (captured) captured.Add(args); }; var handle1 = await driver.SubscribeAsync(["A"], TimeSpan.FromSeconds(1), CancellationToken.None); var handle2 = await driver.SubscribeAsync(["A"], TimeSpan.FromSeconds(1), CancellationToken.None); @@ -162,7 +164,7 @@ public sealed class GalaxyDriverSubscribeTests // SubscriptionRegistryTests cover the multi-sub-same-handle case directly. await subscriber.EmitOnDataChangeAsync(subscriber.Map["A"], 7.0); - await WaitForAsync(() => captured.Count >= 1); + await WaitForCapturedAsync(captured, 1, "the value change must reach a subscription"); // At least one delivery — depending on which subscription owns the handle, // either handle1 or handle2 receives. The fan-out invariant (a single handle @@ -188,16 +190,20 @@ public sealed class GalaxyDriverSubscribeTests "g", Opts(), hierarchySource: null, dataReader: null, dataWriter: null, subscriber: subscriber); var captured = new List(); - driver.OnDataChange += (_, args) => captured.Add(args); + driver.OnDataChange += (_, args) => { lock (captured) captured.Add(args); }; await driver.SubscribeAsync(["Good", "Bad"], TimeSpan.FromSeconds(1), CancellationToken.None); // Good has an itemHandle; Bad doesn't (item handle 0). An event with handle 0 // must NOT be dispatched (no subscribers registered against it). await subscriber.EmitOnDataChangeAsync(itemHandle: 0, value: 999.0); - await Task.Delay(50, TestContext.Current.CancellationToken); // give the pump a chance + // Positive control (#858): the pump dispatches in stream order, so once Good's event is seen the + // handle-0 event has been handled — its absence is a real result, not a short sleep. + await subscriber.EmitOnDataChangeAsync(subscriber.Map["Good"], 1.0); + await WaitForCapturedAsync(captured, 1, "positive control: Good's event must be dispatched"); - captured.ShouldBeEmpty(); + var only = Snapshot(captured).ShouldHaveSingleItem("the handle-0 event must not be dispatched"); + only.FullReference.ShouldBe("Good"); } /// Verifies unsubscribe removes registration and calls gateway unsubscribe. @@ -210,6 +216,8 @@ public sealed class GalaxyDriverSubscribeTests var handle = await driver.SubscribeAsync(["X"], TimeSpan.FromSeconds(1), CancellationToken.None); var itemHandle = subscriber.Map["X"]; + // A second, still-live subscription provides the positive control below. + await driver.SubscribeAsync(["Control"], TimeSpan.FromSeconds(1), CancellationToken.None); await driver.UnsubscribeAsync(handle, CancellationToken.None); @@ -217,10 +225,15 @@ public sealed class GalaxyDriverSubscribeTests // Subsequent events for the dropped handle don't dispatch. var captured = new List(); - driver.OnDataChange += (_, args) => captured.Add(args); + driver.OnDataChange += (_, args) => { lock (captured) captured.Add(args); }; await subscriber.EmitOnDataChangeAsync(itemHandle, 11.0); - await Task.Delay(50, TestContext.Current.CancellationToken); - captured.ShouldBeEmpty(); + // Positive control (#858): events dispatch in stream order, so once the live subscription's event + // is seen the dropped handle's event has been handled. + await subscriber.EmitOnDataChangeAsync(subscriber.Map["Control"], 12.0); + await WaitForCapturedAsync(captured, 1, "positive control: the live subscription's event must be dispatched"); + + var only = Snapshot(captured).ShouldHaveSingleItem("the unsubscribed handle's event must not be dispatched"); + only.FullReference.ShouldBe("Control"); } /// Verifies unsubscribing with an unknown handle is handled. @@ -308,14 +321,21 @@ public sealed class GalaxyDriverSubscribeTests public string DiagnosticId => "foreign-x"; } - private static async Task WaitForAsync(Func predicate, int timeoutMs = 1000) + /// Presence wait (#858): until holds at least events. + internal static Task WaitForCapturedAsync(List captured, int count, string because) => + TestWait.UntilAsync( + () => { lock (captured) return captured.Count >= count; }, + because, + () => + { + lock (captured) + { + return $"dispatched={captured.Count}/{count} [{string.Join(", ", captured.Select(c => $"{c.FullReference}={c.Snapshot.Value}"))}]"; + } + }); + + internal static List Snapshot(List captured) { - var deadline = DateTime.UtcNow.AddMilliseconds(timeoutMs); - while (DateTime.UtcNow < deadline) - { - if (predicate()) return; - await Task.Delay(10); - } - predicate().ShouldBeTrue("Predicate did not become true within timeout."); + lock (captured) return [.. captured]; } } diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/GatewayGalaxyAlarmFeedTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/GatewayGalaxyAlarmFeedTests.cs index fe1fb6cde..ede471356 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/GatewayGalaxyAlarmFeedTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/GatewayGalaxyAlarmFeedTests.cs @@ -53,8 +53,8 @@ public sealed class GatewayGalaxyAlarmFeedTests }; feed.Start(); - (await Task.WhenAny(got3.Task, Task.Delay(TimeSpan.FromSeconds(2), TestContext.Current.CancellationToken))) - .ShouldBe(got3.Task, "snapshot + transition should dispatch within 2s"); + await TestWait.ForAsync(got3.Task, "snapshot + transition should dispatch", + () => { lock (observed) return $"dispatched={observed.Count}/3"; }); observed.Count.ShouldBe(3); @@ -109,7 +109,8 @@ public sealed class GatewayGalaxyAlarmFeedTests }; feed.Start(); - (await Task.WhenAny(got2.Task, Task.Delay(TimeSpan.FromSeconds(2), TestContext.Current.CancellationToken))).ShouldBe(got2.Task); + await TestWait.ForAsync(got2.Task, "restated snapshot Clear + live Retrigger should dispatch", + () => { lock (observed) return $"dispatched={observed.Count}/2"; }); observed[0].TransitionKind.ShouldBe(GalaxyAlarmTransitionKind.Clear); observed[0].IsRestatement.ShouldBeTrue(); @@ -145,8 +146,8 @@ public sealed class GatewayGalaxyAlarmFeedTests }; feed.Start(); - (await Task.WhenAny(gotOne.Task, Task.Delay(TimeSpan.FromSeconds(2), TestContext.Current.CancellationToken))) - .ShouldBe(gotOne.Task); + await TestWait.ForAsync(gotOne.Task, "the well-formed Raise should dispatch", + () => { lock (observed) return $"dispatched={observed.Count}"; }); // Only the well-formed Raise survives; the Unspecified + empty messages drop. observed.ShouldHaveSingleItem(); @@ -187,8 +188,8 @@ public sealed class GatewayGalaxyAlarmFeedTests }; feed.Start(); - (await Task.WhenAny(gotOne.Task, Task.Delay(TimeSpan.FromSeconds(3), TestContext.Current.CancellationToken))) - .ShouldBe(gotOne.Task, "the feed should reopen the stream and deliver after a fault"); + await TestWait.ForAsync(gotOne.Task, "the feed should reopen the stream and deliver after a fault", + () => $"streamOpens={Volatile.Read(ref calls)}"); calls.ShouldBeGreaterThanOrEqualTo(2); observed.ShouldHaveSingleItem(); 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 b875f8f73..accc819cf 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 @@ -11,8 +11,6 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests.Runtime; /// public sealed class ReconnectSupervisorTests { - private const int WaitMs = 2_000; - private static ReconnectOptions FastBackoff() => new(InitialBackoffOverride: TimeSpan.FromMilliseconds(5), MaxBackoffOverride: TimeSpan.FromMilliseconds(20)); @@ -42,7 +40,7 @@ public sealed class ReconnectSupervisorTests sup.StateChanged += (_, t) => { lock (lockObj) transitions.Add(t); }; sup.ReportTransportFailure(new IOException("transport drop")); - await sup.WaitForHealthyAsync(new CancellationTokenSource(WaitMs).Token); + await TestWait.HealthyAsync(sup, "the supervisor must recover to Healthy"); // Expected sequence: Healthy → TransportLost → Reopening → Replaying → Healthy. IReadOnlyList snapshot; @@ -67,7 +65,7 @@ public sealed class ReconnectSupervisorTests options: FastBackoff()); sup.ReportTransportFailure(new IOException("kick off")); - await sup.WaitForHealthyAsync(new CancellationTokenSource(WaitMs).Token); + await TestWait.HealthyAsync(sup, "the supervisor must recover to Healthy"); attempts.ShouldBe(3); sup.CurrentState.ShouldBe(ReconnectSupervisor.State.Healthy); @@ -86,7 +84,7 @@ public sealed class ReconnectSupervisorTests options: FastBackoff()); sup.ReportTransportFailure(new IOException("kick off")); - await sup.WaitForHealthyAsync(new CancellationTokenSource(WaitMs).Token); + await TestWait.HealthyAsync(sup, "the supervisor must recover to Healthy"); // First cycle: reopen succeeds, replay fails. Second cycle: both succeed. reopens.ShouldBe(2); @@ -115,7 +113,7 @@ public sealed class ReconnectSupervisorTests sup.ReportTransportFailure(new IOException($"rapid-{i}")); } - await sup.WaitForHealthyAsync(new CancellationTokenSource(WaitMs).Token); + await TestWait.HealthyAsync(sup, "the supervisor must recover to Healthy"); // One Reopen call regardless of how many failures arrived during recovery. attempts.ShouldBe(1); @@ -137,12 +135,10 @@ public sealed class ReconnectSupervisorTests // Wait for the loop's first failed reopen to land rather than sleeping a fixed 100 ms: on a // loaded CI runner the recovery loop had not been scheduled yet, so LastError still held the // reported cause ("initial") — #606. A presence wait returns the instant the condition holds. - var deadline = DateTime.UtcNow.AddMilliseconds(WaitMs); - while (DateTime.UtcNow < deadline - && sup.LastError?.Contains("reopen broke", StringComparison.OrdinalIgnoreCase) != true) - { - await Task.Delay(5, TestContext.Current.CancellationToken); - } + await TestWait.UntilAsync( + () => sup.LastError?.Contains("reopen broke", StringComparison.OrdinalIgnoreCase) == true, + "the loop's failed reopen must replace the reported cause", + () => $"state={sup.CurrentState} lastError={sup.LastError ?? ""}"); sup.LastError.ShouldNotBeNull(); sup.LastError.ShouldContain("reopen broke"); // updates from the loop's failed reopen attempts @@ -153,17 +149,21 @@ public sealed class ReconnectSupervisorTests public async Task Dispose_CancelsRunningRecoveryLoop_Cleanly() { var cancelled = false; + var reopenEntered = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); var sup = new ReconnectSupervisor( reopen: async ct => { - try { await Task.Delay(10_000, ct).ConfigureAwait(false); } + reopenEntered.TrySetResult(); + try { await Task.Delay(Timeout.Infinite, ct).ConfigureAwait(false); } catch (OperationCanceledException) { cancelled = true; throw; } }, replay: _ => Task.CompletedTask, options: FastBackoff()); sup.ReportTransportFailure(new IOException("kick off")); - await Task.Delay(50, TestContext.Current.CancellationToken); // let the loop start the long reopen + // Wait for the loop to be INSIDE the long reopen (#858: a fixed 50 ms sleep raced the loop's + // scheduling under load, and a Dispose before reopen starts never reaches the catch). + await TestWait.ForAsync(reopenEntered.Task, "the recovery loop must enter reopen"); Should.NotThrow(() => sup.Dispose()); cancelled.ShouldBeTrue(); } @@ -182,9 +182,10 @@ public sealed class ReconnectSupervisorTests public async Task WaitForHealthy_ReturnsImmediately_WhenAlreadyHealthy() { using var sup = new ReconnectSupervisor(_ => Task.CompletedTask, _ => Task.CompletedTask, FastBackoff()); - // No failure reported — should be Healthy from the start. - var deadline = DateTime.UtcNow.AddMilliseconds(50); - await sup.WaitForHealthyAsync(new CancellationTokenSource(50).Token); - DateTime.UtcNow.ShouldBeLessThan(deadline.AddMilliseconds(100)); // returned promptly + // No failure reported — should be Healthy from the start, so the wait completes synchronously + // (deterministic, unlike a wall-clock bound that a descheduled thread can overrun — #858). + var wait = sup.WaitForHealthyAsync(CancellationToken.None); + wait.IsCompletedSuccessfully.ShouldBeTrue("an already-Healthy supervisor must not wait at all"); + await wait; } } diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/TestWait.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/TestWait.cs new file mode 100644 index 000000000..36fd949da --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/TestWait.cs @@ -0,0 +1,107 @@ +using System.Diagnostics; +using Shouldly; +using ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Runtime; + +namespace ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests; + +/// +/// Shared presence waits for the Galaxy driver tests (#856, #858). +/// +/// +/// +/// Every wait here is a PRESENCE wait: it returns the instant the awaited thing happens, so a +/// generous ceiling costs nothing on a passing run. Under heavy parallel load (#858: 14 concurrent +/// test processes + 14 CPU burners, or a starved thread pool) a background loop started with +/// Task.Run can take seconds just to be scheduled — the old 1–3 s ceilings failed there while +/// the code under test was correct. +/// +/// +/// On timeout the failure reports the elapsed time and, when given, a describe snapshot of the +/// state actually observed, so a flake says which link of the chain never happened. +/// +/// +/// Never assert ABSENCE with a short sleep: follow the event that must not fire with one that MUST +/// arrive (a positive control), wait for that, then assert the absence. +/// +/// +internal static class TestWait +{ + /// Ceiling for a presence wait. Only a genuinely broken run ever waits this long. + internal static readonly TimeSpan Presence = TimeSpan.FromSeconds(30); + + /// Polls until it holds or elapses. + /// The condition to wait for. + /// What the condition means; leads the failure message. + /// Optional snapshot of the observed state, appended on timeout. + /// Optional ceiling overriding . + internal static async Task UntilAsync( + Func condition, string because, Func? describe = null, TimeSpan? timeout = null) + { + var limit = timeout ?? Presence; + var sw = Stopwatch.StartNew(); + while (!condition()) + { + if (sw.Elapsed > limit) + { + throw new ShouldAssertException(Message(because, sw, describe)); + } + await Task.Delay(10).ConfigureAwait(false); + } + } + + /// Waits for to complete, or fails after . + /// The task that must complete (e.g. a TaskCompletionSource.Task). + /// What completion means; leads the failure message. + /// Optional snapshot of the observed state, appended on timeout. + /// A task that completes when does (its fault/cancellation propagates). + internal static async Task ForAsync(Task task, string because, Func? describe = null) + { + var sw = Stopwatch.StartNew(); + var winner = await Task.WhenAny(task, Task.Delay(Presence)).ConfigureAwait(false); + if (winner != task) + { + throw new ShouldAssertException(Message(because, sw, describe)); + } + await task.ConfigureAwait(false); + } + + /// Waits for and returns its result, or fails after . + /// The result type. + /// The task that must complete. + /// What completion means; leads the failure message. + /// Optional snapshot of the observed state, appended on timeout. + /// The task's result. + internal static async Task ForAsync(Task task, string because, Func? describe = null) + { + await ForAsync((Task)task, because, describe).ConfigureAwait(false); + return await task.ConfigureAwait(false); + } + + /// + /// Drives the production under the presence + /// ceiling; a timeout fails with the supervisor's observed state rather than a bare + /// . + /// + /// The supervisor that must return to Healthy. + /// What recovery means here; leads the failure message. + internal static async Task HealthyAsync(ReconnectSupervisor supervisor, string because) + { + var sw = Stopwatch.StartNew(); + using var cts = new CancellationTokenSource(Presence); + try + { + await supervisor.WaitForHealthyAsync(cts.Token).ConfigureAwait(false); + } + catch (OperationCanceledException) when (cts.IsCancellationRequested) + { + throw new ShouldAssertException(Message(because, sw, + () => $"state={supervisor.CurrentState} lastError={supervisor.LastError ?? ""}")); + } + } + + private static string Message(string because, Stopwatch sw, Func? describe) + { + var observed = describe is null ? "" : $" Observed: {describe()}"; + return $"{because} (not met after {sw.ElapsedMilliseconds} ms).{observed}"; + } +}