From 95295210b08b81a3ee868a72a51a74cf6bf630df Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Sun, 26 Jul 2026 10:29:27 -0400 Subject: [PATCH] fix(test): scope the EventPump counter capture to its own pump (#503) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Root cause was cross-test measurement leakage, not a timing budget. EventPump's `Meter` is STATIC — one instance per process — and a `MeterListener` can only select by *meter* in `InstrumentPublished`. `StartMeterCapture` therefore received measurements from every `EventPump` alive anywhere in the assembly. xUnit runs test classes in parallel and several build pumps (`EventPumpStreamFaultTests`, the `Driver-X` test in this same file), so a sibling's events landed in these counters. That is why the flake needed a busy box: it needed two tests to overlap. Measured directly during the gate — `Received = 11` in a run that emitted 10. The old `Received == Dispatched + Dropped + InFlight` assertion is what turned the leak into a failure. `InFlight` is DERIVED as `Received - Dispatched - Dropped`, so the identity is a tautology that cannot fail on a real defect — only on a foreign increment landing between its four separate `Interlocked.Read` calls. Hence the reported subject, `counters.Received`. It is deleted, along with the `InFlight` member, and replaced by direct assertions on the three MEASURED counters. Three further defects in the same test, all found on the way: - **The "held" dispatch loop was never held.** `OnDataChange` is `EventHandler` — void-returning — so `async (_, _) => await gate` is async VOID: it returned to the dispatch loop at the first await and blocked nothing. Measured `Dispatched = 2..3`, never 0. Drops happened only when the producer happened to outrun the consumer, so the arrangement the doc comment describes was fiction and the assertions rode on scheduling luck. Now a synchronous `ManualResetEventSlim.Wait()` on the loop's own thread, pinned by `Dispatched.ShouldBe(0)`. - **Fixed `Task.Delay(150)`** replaced by a presence poll on the drop count (#500's rule). Dropped starts at 0, so it cannot be satisfied by the initial state. - **Emission is staged** — one event, wait for the dispatcher to enter the handler, then the other nine — so the arithmetic is exact (`10 - 1 held - 2 buffered = 7`) rather than scheduler-dependent. The gate release moved into `finally`, ahead of disposal: `DisposeAsync` awaits the dispatch loop, which is now genuinely parked on that gate, so a failed assertion would otherwise hang instead of failing. Verified: 15/15 clean under the exact contention the issue measured (Runtime + AdminUI + Core concurrently) — where the half-fixed version still failed on run 4. Falsifiability: restoring the async-void handler turns the new assertions red. --- .../Runtime/EventPumpBoundedChannelTests.cs | 200 +++++++++++++----- 1 file changed, 151 insertions(+), 49 deletions(-) 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(); }