diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/EventPumpBoundedChannelTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/EventPumpBoundedChannelTests.cs
index 9790bbcd..6ae7f2ff 100644
--- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/EventPumpBoundedChannelTests.cs
+++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/EventPumpBoundedChannelTests.cs
@@ -10,59 +10,109 @@ using ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Runtime;
namespace ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests.Runtime;
///
-/// PR 6.2 — pins the EventPump's bounded-channel + drop-newest behavior. We
-/// hold the dispatch loop with a slow handler so the channel fills, then verify
-/// the producer keeps reading from the gw stream and increments the
-/// galaxy.events.dropped counter rather than blocking.
+/// PR 6.2 — pins the EventPump's bounded-channel + drop-newest behavior. We hold the dispatch loop
+/// with a blocking handler so the channel fills, then verify the producer keeps reading from the gw
+/// stream and increments the galaxy.events.dropped counter rather than blocking.
///
+///
+/// These tests observe two background loops (RunAsync / DispatchLoopAsync) through a
+/// process-wide , so every wait here is a presence assertion polled up to
+/// — never a fixed sleep. A fixed sleep is a bet that both loops get scheduled
+/// within it, and #503 is what losing that bet looks like when the box is busy running other test
+/// assemblies.
+///
public sealed class EventPumpBoundedChannelTests
{
- /// Verifies that the event pump drops newest events when the bounded channel fills and records metrics for dropped events.
+ /// Upper bound on how long a poll waits for a background loop to reach its terminal state.
+ /// Generous on purpose: it is spent only when something is genuinely broken, and it cannot turn a
+ /// failing assertion into a passing one.
+ private static readonly TimeSpan Budget = TimeSpan.FromSeconds(15);
+
+ ///
+ /// With the dispatch loop genuinely held and the channel therefore full, the producer keeps
+ /// draining the gw stream and counts every rejected write on galaxy.events.dropped —
+ /// newest-dropped, not back-pressure.
+ ///
+ ///
+ ///
+ /// The arrangement is staged so the arithmetic is exact rather than scheduler-dependent
+ /// (#503): emit ONE event, wait until the dispatcher has actually entered the handler (so it
+ /// holds that event and the channel is empty), and only then emit the remaining nine. The
+ /// channel accepts capacity of them and the rest must be dropped —
+ /// 10 − 1 held − 2 buffered = 7, every run.
+ ///
+ ///
[Fact]
public async Task Drops_newest_when_channel_fills_and_records_metric()
{
- var counters = StartMeterCapture();
+ const int totalEvents = 10;
+ const int channelCapacity = 2;
+ // One event is held inside the handler; `channelCapacity` more fit in the channel; the rest are dropped.
+ const int expectedDropped = totalEvents - 1 - channelCapacity;
+
+ // Unique per run: the counters are filtered to this exact galaxy.client tag, which is what keeps a
+ // parallel sibling test's pump out of them (#503). A literal name would work today and break the
+ // day someone reuses it.
+ var clientName = $"PumpTest-{Guid.NewGuid():N}";
+ var counters = StartMeterCapture(clientName);
+
+ // A SYNCHRONOUS gate, deliberately. OnDataChange is EventHandler — void-returning — so an
+ // `async (_, _) => await gate` lambda is async void: it returns to the dispatch loop at the first
+ // await and holds nothing. That is exactly what #503 was: the loop drained freely, drops happened
+ // only when the producer happened to outrun the consumer, and the assertions rode on scheduling
+ // luck. Blocking the loop's own thread is what makes "the dispatcher is held" true.
+ var dispatchGate = new ManualResetEventSlim(false);
+ var handlerEntered = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+ EventPump? pump = null;
try
{
var subscriber = new ManualSubscriber();
var registry = new SubscriptionRegistry();
registry.Register(1, [new TagBinding("Tag.A", ItemHandle: 7)]);
- // Tiny channel + slow handler ⇒ producer hits FullMode.Wait → TryWrite false
- // for every overflow event.
- var dispatchGate = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
- await using var pump = new EventPump(
- subscriber, registry, channelCapacity: 2, clientName: "PumpTest");
- pump.OnDataChange += async (_, _) =>
+ pump = new EventPump(
+ subscriber, registry, channelCapacity: channelCapacity, clientName: clientName);
+ pump.OnDataChange += (_, _) =>
{
- // Block the dispatch loop until we've shoved enough events through to
- // overflow the bounded channel. Consume the gate exactly once.
- await dispatchGate.Task.ConfigureAwait(false);
+ handlerEntered.TrySetResult();
+ dispatchGate.Wait();
};
pump.Start();
- const int totalEvents = 10;
- for (var i = 0; i < totalEvents; i++)
+ // Stage 1 — one event, then wait for the dispatcher to take it and block. After this the
+ // channel is provably empty and exactly one event is in flight inside the handler.
+ await subscriber.EmitAsync(itemHandle: 7, value: 0);
+ await handlerEntered.Task.WaitAsync(Budget);
+
+ // Stage 2 — the rest. `channelCapacity` are accepted; the remainder must be dropped.
+ for (var i = 1; i < totalEvents; i++)
{
await subscriber.EmitAsync(itemHandle: 7, value: i);
}
- // Give the producer a beat to run TryWrite for every event.
- await Task.Delay(150);
- // Capacity 2 + 1 in-flight in the dispatcher = 3 may have been accepted; the
- // remainder should have hit the dropped counter. Don't pin exact counts —
- // the scheduler can interleave; pin the invariants instead.
- counters.Received.ShouldBeGreaterThanOrEqualTo(totalEvents);
- counters.Dropped.ShouldBeGreaterThan(0,
- "with capacity=2 and a held dispatcher we must drop at least one of 10 events");
- (counters.Received).ShouldBe(counters.Dispatched + counters.Dropped + counters.InFlight,
- "received = dispatched + dropped + (events still queued)");
+ // PRESENCE assertion on the discriminator (#500's rule): poll until the drop count reaches its
+ // terminal value rather than sleeping a fixed 150 ms and hoping the producer got scheduled.
+ // The budget is an upper bound before giving up — spending it costs nothing on the happy path,
+ // and no amount of it can make a genuinely broken pump pass. Dropped starts at 0, so this
+ // cannot be satisfied by the initial state.
+ await WaitUntilAsync(() => counters.Dropped == expectedDropped,
+ $"expected {expectedDropped} dropped events");
- // Release the dispatcher so DisposeAsync can drain cleanly.
- dispatchGate.TrySetResult();
+ counters.Received.ShouldBe(totalEvents,
+ "the producer must keep reading the gw stream through the overflow, not back-pressure");
+ counters.Dropped.ShouldBe(expectedDropped);
+ counters.Dispatched.ShouldBe(0,
+ "the dispatch loop is still blocked in the handler, so nothing has completed dispatch — " +
+ "if this is non-zero the handler is not actually holding the loop and the test proves nothing");
}
finally
{
+ // Release BEFORE disposing: DisposeAsync awaits the dispatch loop, and that loop is parked on
+ // this gate. Releasing it here (not at the end of the try) is what keeps a failed assertion a
+ // failure instead of a hang.
+ dispatchGate.Set();
+ if (pump is not null) await pump.DisposeAsync();
+ dispatchGate.Dispose();
counters.Dispose();
}
}
@@ -110,23 +160,21 @@ public sealed class EventPumpBoundedChannelTests
// Poll until at least one galaxy.events.received measurement tagged
// galaxy.client=Driver-X lands in the listener, rather than using a
// fixed delay that races under parallel test load on a busy box.
- var deadline = DateTime.UtcNow.AddSeconds(5);
- bool found = false;
- while (DateTime.UtcNow < deadline)
- {
- listener.RecordObservableInstruments();
- bool hasMatch;
- lock (captured)
+ // Budget, not 5 s: the same contention that produced #503 stretches how long the
+ // producer loop takes to be scheduled, and a presence poll costs nothing to over-budget.
+ await WaitUntilAsync(
+ () =>
{
- hasMatch = captured.Any(c =>
- c.Instrument == "galaxy.events.received" &&
- c.Tags.Any(t => t.Key == "galaxy.client" &&
- string.Equals((string?)t.Value, "Driver-X", StringComparison.Ordinal)));
- }
- if (hasMatch) { found = true; break; }
- await Task.Delay(25);
- }
- _ = found; // assertion happens below after dispose
+ listener.RecordObservableInstruments();
+ lock (captured)
+ {
+ return captured.Any(c =>
+ c.Instrument == "galaxy.events.received" &&
+ c.Tags.Any(t => t.Key == "galaxy.client" &&
+ string.Equals((string?)t.Value, "Driver-X", StringComparison.Ordinal)));
+ }
+ },
+ "a galaxy.events.received measurement tagged galaxy.client=Driver-X");
}
// The static Meter is shared across all EventPump instances in the test
@@ -146,7 +194,46 @@ public sealed class EventPumpBoundedChannelTests
ours.ShouldContain(c => c.Instrument == "galaxy.events.received");
}
- private static CounterCapture StartMeterCapture()
+ /// Polls until it holds or elapses, failing
+ /// with on timeout. The presence-assertion counterpart to a fixed delay.
+ private static async Task WaitUntilAsync(Func condition, string what)
+ {
+ var deadline = DateTime.UtcNow + Budget;
+ while (DateTime.UtcNow < deadline)
+ {
+ if (condition()) return;
+ await Task.Delay(10);
+ }
+ condition().ShouldBeTrue($"timed out after {Budget.TotalSeconds:0}s waiting: {what}");
+ }
+
+ ///
+ /// Captures the three EventPump counters for one specific pump, identified by its
+ /// galaxy.client tag.
+ ///
+ ///
+ ///
+ /// The galaxy.client filter is the fix for #503, and it is not optional. The pump's
+ /// is static — one instance for the whole process — and
+ /// InstrumentPublished can only select by meter, so an unfiltered listener
+ /// receives measurements from every alive anywhere in the assembly.
+ /// xUnit runs test classes in parallel and several of them build pumps
+ /// (EventPumpStreamFaultTests, the Driver-X test below), so a sibling's events
+ /// landed in these counters — that is why the flake needed a busy box: it needed the two
+ /// tests to overlap in time. The leak was measured directly, as Received = 11 in a run
+ /// that emitted 10.
+ ///
+ ///
+ /// The old Received == Dispatched + Dropped + InFlight assertion is what turned the
+ /// leak into a failure: a foreign increment landing between its four separate
+ /// calls made the two sides disagree. Hence the
+ /// reported subject, counters.Received.
+ ///
+ ///
+ /// The galaxy.client tag value to accept. Must be unique to the
+ /// calling test — pass a fresh GUID-suffixed name rather than a literal, so a future sibling test
+ /// cannot silently reintroduce the leak by reusing the same name.
+ private static CounterCapture StartMeterCapture(string clientName)
{
var capture = new CounterCapture();
var listener = new MeterListener();
@@ -154,8 +241,20 @@ public sealed class EventPumpBoundedChannelTests
{
if (instr.Meter.Name == EventPump.MeterName) l.EnableMeasurementEvents(instr);
};
- listener.SetMeasurementEventCallback((instr, value, _, _) =>
+ listener.SetMeasurementEventCallback((instr, value, tags, _) =>
{
+ var mine = false;
+ foreach (var tag in tags)
+ {
+ if (tag.Key == "galaxy.client" &&
+ string.Equals((string?)tag.Value, clientName, StringComparison.Ordinal))
+ {
+ mine = true;
+ break;
+ }
+ }
+ if (!mine) return;
+
switch (instr.Name)
{
case "galaxy.events.received": Interlocked.Add(ref capture._received, value); break;
@@ -178,8 +277,11 @@ public sealed class EventPumpBoundedChannelTests
public long Dispatched => Interlocked.Read(ref _dispatched);
/// Gets the count of dropped events.
public long Dropped => Interlocked.Read(ref _dropped);
- /// Gets the count of in-flight events.
- public long InFlight => Math.Max(0, Received - Dispatched - Dropped);
+ // There is deliberately no InFlight member. It could only be DERIVED as
+ // Received - Dispatched - Dropped, which made the old
+ // `Received == Dispatched + Dropped + InFlight` assertion a tautology: it could not fail on a
+ // real defect, only on a torn read across the four separate Interlocked.Read calls. Assert the
+ // three measured counters directly instead (#503).
/// Disposes the meter listener.
public void Dispose() => Listener?.Dispose();
}