fix(test): scope the EventPump counter capture to its own pump (#503)

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<T>`
  — 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.

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;
 
 /// <summary>
-///     PR 6.2 — pins the EventPump's bounded-channel + drop-newest behavior. We
+///     PR 6.2 — pins the EventPump's bounded-channel + drop-newest behavior. We hold the dispatch loop
-///     hold the dispatch loop with a slow handler so the channel fills, then verify
+///     with a blocking handler so the channel fills, then verify the producer keeps reading from the gw
-///     the producer keeps reading from the gw stream and increments the
+///     stream and increments the <c>galaxy.events.dropped</c> counter rather than blocking.
-///     <c>galaxy.events.dropped</c> counter rather than blocking.
 /// </summary>
+/// <remarks>
+///     These tests observe two background loops (<c>RunAsync</c> / <c>DispatchLoopAsync</c>) through a
+///     process-wide <see cref="Meter"/>, so every wait here is a <b>presence</b> assertion polled up to
+///     <see cref="Budget"/> — 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.
+/// </remarks>
 public sealed class EventPumpBoundedChannelTests
 {
-    /// <summary>Verifies that the event pump drops newest events when the bounded channel fills and records metrics for dropped events.</summary>
+    /// <summary>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.</summary>
+    private static readonly TimeSpan Budget = TimeSpan.FromSeconds(15);
+
+    /// <summary>
+    ///     With the dispatch loop genuinely held and the channel therefore full, the producer keeps
+    ///     draining the gw stream and counts every rejected write on <c>galaxy.events.dropped</c> —
+    ///     newest-dropped, not back-pressure.
+    /// </summary>
+    /// <remarks>
+    ///     <para>
+    ///         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 <c>capacity</c> of them and the rest must be dropped —
+    ///         <c>10 − 1 held − 2 buffered = 7</c>, every run.
+    ///     </para>
+    /// </remarks>
     [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<T> — 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
+            pump = new EventPump(
-            // for every overflow event.
+                subscriber, registry, channelCapacity: channelCapacity, clientName: clientName);
-            var dispatchGate = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+            pump.OnDataChange += (_, _) =>
-            await using var pump = new EventPump(
-                subscriber, registry, channelCapacity: 2, clientName: "PumpTest");
-            pump.OnDataChange += async (_, _) =>
             {
-                // Block the dispatch loop until we've shoved enough events through to
+                handlerEntered.TrySetResult();
-                // overflow the bounded channel. Consume the gate exactly once.
+                dispatchGate.Wait();
-                await dispatchGate.Task.ConfigureAwait(false);
             };
             pump.Start();
 
-            const int totalEvents = 10;
+            // Stage 1 — one event, then wait for the dispatcher to take it and block. After this the
-            for (var i = 0; i < totalEvents; i++)
+            // 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
+            // PRESENCE assertion on the discriminator (#500's rule): poll until the drop count reaches its
-            // remainder should have hit the dropped counter. Don't pin exact counts —
+            // terminal value rather than sleeping a fixed 150 ms and hoping the producer got scheduled.
-            // the scheduler can interleave; pin the invariants instead.
+            // The budget is an upper bound before giving up — spending it costs nothing on the happy path,
-            counters.Received.ShouldBeGreaterThanOrEqualTo(totalEvents);
+            // and no amount of it can make a genuinely broken pump pass. Dropped starts at 0, so this
-            counters.Dropped.ShouldBeGreaterThan(0,
+            // cannot be satisfied by the initial state.
-                "with capacity=2 and a held dispatcher we must drop at least one of 10 events");
+            await WaitUntilAsync(() => counters.Dropped == expectedDropped,
-            (counters.Received).ShouldBe(counters.Dispatched + counters.Dropped + counters.InFlight,
+                $"expected {expectedDropped} dropped events");
-                "received = dispatched + dropped + (events still queued)");
 
-            // Release the dispatcher so DisposeAsync can drain cleanly.
+            counters.Received.ShouldBe(totalEvents,
-            dispatchGate.TrySetResult();
+                "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);
+            // Budget, not 5 s: the same contention that produced #503 stretches how long the
-            bool found = false;
+            // producer loop takes to be scheduled, and a presence poll costs nothing to over-budget.
-            while (DateTime.UtcNow < deadline)
+            await WaitUntilAsync(
-            {
+                () =>
-                listener.RecordObservableInstruments();
-                bool hasMatch;
-                lock (captured)
                 {
-                    hasMatch = captured.Any(c =>
+                    listener.RecordObservableInstruments();
-                        c.Instrument == "galaxy.events.received" &&
+                    lock (captured)
-                        c.Tags.Any(t => t.Key == "galaxy.client" &&
+                    {
-                            string.Equals((string?)t.Value, "Driver-X", StringComparison.Ordinal)));
+                        return captured.Any(c =>
-                }
+                            c.Instrument == "galaxy.events.received" &&
-                if (hasMatch) { found = true; break; }
+                            c.Tags.Any(t => t.Key == "galaxy.client" &&
-                await Task.Delay(25);
+                                string.Equals((string?)t.Value, "Driver-X", StringComparison.Ordinal)));
-            }
+                    }
-            _ = found; // assertion happens below after dispose
+                },
+                "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()
+    /// <summary>Polls <paramref name="condition"/> until it holds or <see cref="Budget"/> elapses, failing
+    /// with <paramref name="what"/> on timeout. The presence-assertion counterpart to a fixed delay.</summary>
+    private static async Task WaitUntilAsync(Func<bool> 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}");
+    }
+
+    /// <summary>
+    ///     Captures the three EventPump counters for <b>one specific pump</b>, identified by its
+    ///     <c>galaxy.client</c> tag.
+    /// </summary>
+    /// <remarks>
+    ///     <para>
+    ///         The <c>galaxy.client</c> filter is the fix for #503, and it is not optional. The pump's
+    ///         <see cref="Meter"/> is <b>static</b> — one instance for the whole process — and
+    ///         <c>InstrumentPublished</c> can only select by <i>meter</i>, so an unfiltered listener
+    ///         receives measurements from every <see cref="EventPump"/> alive anywhere in the assembly.
+    ///         xUnit runs test classes in parallel and several of them build pumps
+    ///         (<c>EventPumpStreamFaultTests</c>, the <c>Driver-X</c> 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 <c>Received = 11</c> in a run
+    ///         that emitted 10.
+    ///     </para>
+    ///     <para>
+    ///         The old <c>Received == Dispatched + Dropped + InFlight</c> assertion is what turned the
+    ///         leak into a failure: a foreign increment landing between its four separate
+    ///         <see cref="Interlocked.Read(ref long)"/> calls made the two sides disagree. Hence the
+    ///         reported subject, <c>counters.Received</c>.
+    ///     </para>
+    /// </remarks>
+    /// <param name="clientName">The <c>galaxy.client</c> 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.</param>
+    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<long>((instr, value, _, _) =>
+        listener.SetMeasurementEventCallback<long>((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);
         /// <summary>Gets the count of dropped events.</summary>
         public long Dropped => Interlocked.Read(ref _dropped);
-        /// <summary>Gets the count of in-flight events.</summary>
+        // There is deliberately no InFlight member. It could only be DERIVED as
-        public long InFlight => Math.Max(0, Received - Dispatched - Dropped);
+        // 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).
         /// <summary>Disposes the meter listener.</summary>
         public void Dispose() => Listener?.Dispose();
     }