Joseph Doherty
95295210b0
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.
331 lines
17 KiB
C#
331 lines
17 KiB
C#
using System.Diagnostics.Metrics;
|
||
using System.Threading.Channels;
|
||
using Google.Protobuf.WellKnownTypes;
|
||
using ZB.MOM.WW.MxGateway.Contracts.Proto;
|
||
using Shouldly;
|
||
using Xunit;
|
||
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
|
||
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 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 <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>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()
|
||
{
|
||
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)]);
|
||
|
||
pump = new EventPump(
|
||
subscriber, registry, channelCapacity: channelCapacity, clientName: clientName);
|
||
pump.OnDataChange += (_, _) =>
|
||
{
|
||
handlerEntered.TrySetResult();
|
||
dispatchGate.Wait();
|
||
};
|
||
pump.Start();
|
||
|
||
// 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);
|
||
}
|
||
|
||
// 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");
|
||
|
||
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();
|
||
}
|
||
}
|
||
|
||
/// <summary>Verifies that the event pump throws an exception when the channel capacity is invalid.</summary>
|
||
[Fact]
|
||
public async Task Throws_when_channelCapacity_is_invalid()
|
||
{
|
||
var subscriber = new ManualSubscriber();
|
||
var registry = new SubscriptionRegistry();
|
||
Should.Throw<ArgumentOutOfRangeException>(() =>
|
||
new EventPump(subscriber, registry, channelCapacity: 0));
|
||
Should.Throw<ArgumentOutOfRangeException>(() =>
|
||
new EventPump(subscriber, registry, channelCapacity: -1));
|
||
await Task.CompletedTask;
|
||
}
|
||
|
||
/// <summary>Verifies that event pump metrics are tagged with the client name for tracking multiple driver hosts.</summary>
|
||
[Fact]
|
||
public async Task Tags_metrics_with_client_name_for_multi_driver_hosts()
|
||
{
|
||
var captured = new List<(string Instrument, KeyValuePair<string, object?>[] Tags)>();
|
||
using var listener = new MeterListener();
|
||
listener.InstrumentPublished = (instr, l) =>
|
||
{
|
||
if (instr.Meter.Name == EventPump.MeterName) l.EnableMeasurementEvents(instr);
|
||
};
|
||
// The callback fires on the thread that calls Counter.Add() — that is the
|
||
// RunAsync background Task. Use lock(captured) everywhere to avoid torn reads.
|
||
listener.SetMeasurementEventCallback<long>((instr, _, tags, _) =>
|
||
{
|
||
lock (captured) { captured.Add((instr.Name, tags.ToArray())); }
|
||
});
|
||
listener.Start();
|
||
|
||
var subscriber = new ManualSubscriber();
|
||
var registry = new SubscriptionRegistry();
|
||
registry.Register(1, [new TagBinding("Tag.A", ItemHandle: 7)]);
|
||
|
||
await using (var pump = new EventPump(subscriber, registry, channelCapacity: 4, clientName: "Driver-X"))
|
||
{
|
||
pump.Start();
|
||
await subscriber.EmitAsync(7, 42.0);
|
||
|
||
// 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.
|
||
// 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(
|
||
() =>
|
||
{
|
||
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
|
||
// assembly; xUnit may run other pump tests in parallel and their
|
||
// measurements land on the same listener. Filter to our pump's tag value.
|
||
List<(string Instrument, KeyValuePair<string, object?>[] Tags)> ours;
|
||
lock (captured)
|
||
{
|
||
ours = captured
|
||
.Where(c => c.Tags.Any(t => t.Key == "galaxy.client"
|
||
&& string.Equals((string?)t.Value, "Driver-X", StringComparison.Ordinal)))
|
||
.ToList();
|
||
}
|
||
|
||
ours.ShouldNotBeEmpty(
|
||
"at least one measurement from this test's pump must carry galaxy.client=Driver-X");
|
||
ours.ShouldContain(c => c.Instrument == "galaxy.events.received");
|
||
}
|
||
|
||
/// <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();
|
||
listener.InstrumentPublished = (instr, l) =>
|
||
{
|
||
if (instr.Meter.Name == EventPump.MeterName) l.EnableMeasurementEvents(instr);
|
||
};
|
||
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;
|
||
case "galaxy.events.dispatched": Interlocked.Add(ref capture._dispatched, value); break;
|
||
case "galaxy.events.dropped": Interlocked.Add(ref capture._dropped, value); break;
|
||
}
|
||
});
|
||
listener.Start();
|
||
capture.Listener = listener;
|
||
return capture;
|
||
}
|
||
|
||
private sealed class CounterCapture : IDisposable
|
||
{
|
||
public MeterListener? Listener;
|
||
internal long _received, _dispatched, _dropped;
|
||
/// <summary>Gets the count of received events.</summary>
|
||
public long Received => Interlocked.Read(ref _received);
|
||
/// <summary>Gets the count of dispatched events.</summary>
|
||
public long Dispatched => Interlocked.Read(ref _dispatched);
|
||
/// <summary>Gets the count of dropped events.</summary>
|
||
public long Dropped => Interlocked.Read(ref _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).
|
||
/// <summary>Disposes the meter listener.</summary>
|
||
public void Dispose() => Listener?.Dispose();
|
||
}
|
||
|
||
private sealed class ManualSubscriber : IGalaxySubscriber
|
||
{
|
||
private readonly Channel<MxEvent> _stream =
|
||
Channel.CreateUnbounded<MxEvent>(new UnboundedChannelOptions { SingleReader = true });
|
||
|
||
/// <summary>Subscribes to a bulk list of tag references.</summary>
|
||
/// <param name="fullReferences">The list of full references to subscribe to.</param>
|
||
/// <param name="bufferedUpdateIntervalMs">The buffered update interval in milliseconds.</param>
|
||
/// <param name="cancellationToken">The cancellation token.</param>
|
||
/// <returns>An empty result list.</returns>
|
||
public Task<IReadOnlyList<SubscribeResult>> SubscribeBulkAsync(
|
||
IReadOnlyList<string> fullReferences, int bufferedUpdateIntervalMs, CancellationToken cancellationToken)
|
||
=> Task.FromResult<IReadOnlyList<SubscribeResult>>([]);
|
||
|
||
/// <summary>Unsubscribes from a bulk list of item handles.</summary>
|
||
/// <param name="itemHandles">The list of item handles to unsubscribe from.</param>
|
||
/// <param name="cancellationToken">The cancellation token.</param>
|
||
/// <returns>A completed task.</returns>
|
||
public Task UnsubscribeBulkAsync(IReadOnlyList<int> itemHandles, CancellationToken cancellationToken)
|
||
=> Task.CompletedTask;
|
||
|
||
/// <summary>Streams events asynchronously.</summary>
|
||
/// <param name="cancellationToken">The cancellation token.</param>
|
||
/// <returns>An async enumerable of MxEvent objects.</returns>
|
||
public IAsyncEnumerable<MxEvent> StreamEventsAsync(CancellationToken cancellationToken)
|
||
=> _stream.Reader.ReadAllAsync(cancellationToken);
|
||
|
||
/// <summary>Emits a test event with the specified item handle and value.</summary>
|
||
/// <param name="itemHandle">The item handle.</param>
|
||
/// <param name="value">The event value.</param>
|
||
/// <returns>A completed value task.</returns>
|
||
public ValueTask EmitAsync(int itemHandle, double value) =>
|
||
_stream.Writer.WriteAsync(new MxEvent
|
||
{
|
||
Family = MxEventFamily.OnDataChange,
|
||
ItemHandle = itemHandle,
|
||
Value = new MxValue { DoubleValue = value },
|
||
Quality = 192,
|
||
SourceTimestamp = Timestamp.FromDateTime(DateTime.UtcNow),
|
||
});
|
||
}
|
||
}
|