dohertj2/lmxopcua
1
0
Fork 0
Code Issues 5 Pull Requests Actions Packages Projects Releases Wiki Activity

Merge #858 Galaxy.Tests: generous self-diagnosing presence waits, positive-control absence checks

Browse Source 
Claude-Session: https://claude.ai/code/session_01GT67AD99zQQqpLCVm3p7xk
This commit is contained in:
Joseph Doherty
2026-11-11 10:18:45 -05:00
parent 4b68729a8a 4fdc3708f4
commit 701ece9973
11 changed files with 256 additions and 133 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Browse/DeployWatcherTests.cs
+48 -30
View File
@@ -108,15 +108,23 @@ public sealed class DeployWatcherTests
return captured;
}
private static async Task WaitUntilAsync(Func<bool> 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<bool> condition, string because, FakeDeployWatchSource source, List<RediscoveryEventArgs> captured)
=> TestWait.UntilAsync(condition, because, () => Describe(source, captured));
private static string Describe(FakeDeployWatchSource source, List<RediscoveryEventArgs> 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<RediscoveryEventArgs> Snapshot(List<RediscoveryEventArgs> captured)
{
lock (captured) return [.. captured];
}
/// <summary>Verifies that bootstrap deploy events are suppressed.</summary>
@@ -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");
}
/// <summary>Verifies that disposing stops a running watcher.</summary>
@@ -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");
}
/// <summary>Verifies that a source exception triggers retry with backoff.</summary>
@@ -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");
tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/EventPumpReplayGapTests.cs
+8 -13
View File
@@ -30,7 +30,6 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests.Runtime;
/// </remarks>
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)
{
tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/EventPumpStreamFaultTests.cs
+11 -15
View File
@@ -16,8 +16,6 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests.Runtime;
/// </summary>
public sealed class EventPumpStreamFaultTests
{
private const int WaitMs = 2_000;
/// <summary>Verifies that stream fault invokes the callback with the exception.</summary>
[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<IOException>();
}
@@ -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");
}
tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/GalaxyDriverFixRound1Tests.cs
+5 -16
View File
@@ -144,23 +144,12 @@ public sealed class GalaxyDriverFixRound1Tests
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.
/// Presence wait: polls <paramref name="condition"/> for up to <see cref="WaitMs"/> via the shared
/// <see cref="TestWait.UntilAsync"/>. On timeout the failure carries the elapsed time and, when given,
/// <paramref name="describe"/>'s snapshot of the state actually observed.
/// </summary>
internal static async Task WaitUntilAsync(Func<bool> condition, string because, Func<string>? 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<bool> condition, string because, Func<string>? 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;
tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/GalaxyDriverFixRound2Tests.cs
+3 -1
View File
@@ -24,7 +24,9 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests.Runtime;
/// </summary>
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);
/// <summary>A platform source whose fetches park until released (optionally ignoring cancellation).</summary>
private sealed class GatedPlatformSource : IGalaxyPlatformSource
tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/GalaxyDriverRawPathKeyingTests.cs
+2 -12
View File
@@ -95,7 +95,7 @@ public sealed class GalaxyDriverRawPathKeyingTests
"g", OptsWithRawTags(), hierarchySource: null, dataReader: null, dataWriter: null, subscriber: subscriber);
var captured = new List<DataChangeEventArgs>();
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<bool> 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 };
tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/GalaxyDriverReadTests.cs
+6 -2
View File
@@ -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;
tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/GalaxyDriverSubscribeTests.cs
+39 -19
View File
@@ -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<MxEvent> _events = Channel.CreateUnbounded<MxEvent>();
/// <summary>Gets the mapping of tag references to subscription handles.</summary>
public Dictionary<string, int> Map { get; } = new();
/// <remarks>Concurrent: the driver may run SubscribeBulk on another thread while a test polls it.</remarks>
public ConcurrentDictionary<string, int> Map { get; } = new();
/// <summary>Gets the list of unsubscribed handles.</summary>
public List<int> UnsubscribedHandles { get; } = [];
@@ -121,14 +123,14 @@ public sealed class GalaxyDriverSubscribeTests
"g", Opts(), hierarchySource: null, dataReader: null, dataWriter: null, subscriber: subscriber);
var captured = new List<DataChangeEventArgs>();
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<DataChangeEventArgs>();
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<DataChangeEventArgs>();
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");
}
/// <summary>Verifies unsubscribe removes registration and calls gateway unsubscribe.</summary>
@@ -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<DataChangeEventArgs>();
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");
}
/// <summary>Verifies unsubscribing with an unknown handle is handled.</summary>
@@ -308,14 +321,21 @@ public sealed class GalaxyDriverSubscribeTests
public string DiagnosticId => "foreign-x";
}
private static async Task WaitForAsync(Func<bool> predicate, int timeoutMs = 1000)
/// <summary>Presence wait (#858): until <paramref name="captured"/> holds at least <paramref name="count"/> events.</summary>
internal static Task WaitForCapturedAsync(List<DataChangeEventArgs> 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<DataChangeEventArgs> Snapshot(List<DataChangeEventArgs> 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];
}
}
tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/GatewayGalaxyAlarmFeedTests.cs
+8 -7
View File
@@ -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();
tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/ReconnectSupervisorTests.cs
+19 -18
View File
@@ -11,8 +11,6 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests.Runtime;
/// </summary>
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<StateTransition> 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 ?? "<none>"}");
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;
}
}
tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/TestWait.cs
+107
View File
@@ -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;
/// <summary>
/// Shared presence waits for the Galaxy driver tests (#856, #858).
/// </summary>
/// <remarks>
/// <para>
/// 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
/// <c>Task.Run</c> can take seconds just to be scheduled — the old 1–3 s ceilings failed there while
/// the code under test was correct.
/// </para>
/// <para>
/// On timeout the failure reports the elapsed time and, when given, a <c>describe</c> snapshot of the
/// state actually observed, so a flake says which link of the chain never happened.
/// </para>
/// <para>
/// 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.
/// </para>
/// </remarks>
internal static class TestWait
{
/// <summary>Ceiling for a presence wait. Only a genuinely broken run ever waits this long.</summary>
internal static readonly TimeSpan Presence = TimeSpan.FromSeconds(30);
/// <summary>Polls <paramref name="condition"/> until it holds or <see cref="Presence"/> elapses.</summary>
/// <param name="condition">The condition to wait for.</param>
/// <param name="because">What the condition means; leads the failure message.</param>
/// <param name="describe">Optional snapshot of the observed state, appended on timeout.</param>
/// <param name="timeout">Optional ceiling overriding <see cref="Presence"/>.</param>
internal static async Task UntilAsync(
Func<bool> condition, string because, Func<string>? 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);
}
}
/// <summary>Waits for <paramref name="task"/> to complete, or fails after <see cref="Presence"/>.</summary>
/// <param name="task">The task that must complete (e.g. a <c>TaskCompletionSource.Task</c>).</param>
/// <param name="because">What completion means; leads the failure message.</param>
/// <param name="describe">Optional snapshot of the observed state, appended on timeout.</param>
/// <returns>A task that completes when <paramref name="task"/> does (its fault/cancellation propagates).</returns>
internal static async Task ForAsync(Task task, string because, Func<string>? 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);
}
/// <summary>Waits for <paramref name="task"/> and returns its result, or fails after <see cref="Presence"/>.</summary>
/// <typeparam name="T">The result type.</typeparam>
/// <param name="task">The task that must complete.</param>
/// <param name="because">What completion means; leads the failure message.</param>
/// <param name="describe">Optional snapshot of the observed state, appended on timeout.</param>
/// <returns>The task's result.</returns>
internal static async Task<T> ForAsync<T>(Task<T> task, string because, Func<string>? describe = null)
{
await ForAsync((Task)task, because, describe).ConfigureAwait(false);
return await task.ConfigureAwait(false);
}
/// <summary>
/// Drives the production <see cref="ReconnectSupervisor.WaitForHealthyAsync"/> under the presence
/// ceiling; a timeout fails with the supervisor's observed state rather than a bare
/// <see cref="TaskCanceledException"/>.
/// </summary>
/// <param name="supervisor">The supervisor that must return to Healthy.</param>
/// <param name="because">What recovery means here; leads the failure message.</param>
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 ?? "<none>"}"));
}
}
private static string Message(string because, Stopwatch sw, Func<string>? describe)
{
var observed = describe is null ? "" : $" Observed: {describe()}";
return $"{because} (not met after {sw.ElapsedMilliseconds} ms).{observed}";
}
}