From 50322bacb365f8b09b022430ef19f27a05ea8aa5 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Mon, 10 Aug 2026 08:13:30 -0400 Subject: [PATCH] test(worker): deflake EventBurst_DrainLoopCoalescesFlushes MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Root cause. The test sampled its flush baseline after a bare `Task.Delay(100)` and then charged every later flush to the 128-event burst. Two facts make that window unsound: 1. `RunHeartbeatLoopAsync` sends its first beat *immediately* on entering the message loop — the far-off `HeartbeatInterval` only spaces later beats — so the pre-burst frames are WorkerHello, WorkerReady, and a heartbeat, not the two the test's comment assumed. 2. `WorkerFrameWriter.DrainQueuedFramesAsync` flushes *after* writing a drained batch, so the bytes reach the pipe before the flush runs. Reading a frame off the gateway side is therefore no evidence that its flush has been counted, and no sleep makes it evidence. Under load on the shared windows-x86 runner the first heartbeat's flush was scheduled after the 100 ms sample, so it landed inside the measured window and the assertion saw two flushes for the burst — exactly the observed `Expected: 1 / Actual: 2` (Gitea run 675, job 2558, and the same failure since a346d51). Nothing about the coalescing behavior was wrong; only the test's timing assumption. Fix. Replace the sleep with explicit synchronization, no widened timeouts. `FlushCountingPassthroughStream` now records the flush *shape* — the number of stream writes coalesced into each flush — and exposes `WaitForAllWritesFlushedAsync`, a TaskCompletionSource signal released when the next flush drains the pending writes. The test reads the first heartbeat (the last pre-burst frame), waits for its flush, and only then samples the baseline; after the burst it takes the same edge before asserting. The assertion is also sharpened from a bare count to the shape: exactly one flush beyond the baseline *and* that flush carried all 128 event frames, so a split batch fails even if the reader observes it mid-split. docs/WorkerFrameProtocol.md notes the peer-visible ordering the fix turns on: frames reach the pipe before the flush that follows them, so an observer of the flush must wait for it rather than infer it from frames arriving. --- docs/WorkerFrameProtocol.md | 6 +- .../Ipc/WorkerPipeSessionTests.cs | 179 ++++++++++++++++-- 2 files changed, 169 insertions(+), 16 deletions(-) diff --git a/docs/WorkerFrameProtocol.md b/docs/WorkerFrameProtocol.md index ea1aa59..b623707 100644 --- a/docs/WorkerFrameProtocol.md +++ b/docs/WorkerFrameProtocol.md @@ -123,7 +123,11 @@ runs after the whole batch, and only then does every successfully-written frame's completion resolve — so a caller's `WriteAsync` still does not complete until its bytes are both written *and* flushed, but a batch that happened to contain several queued frames pays one flush instead of one per -frame. The event drain loop (`WorkerPipeSession.RunEventDrainLoopAsync`) +frame. Note the ordering this implies at the peer: the frames reach the pipe +before the flush that follows them, so the gateway can read a whole batch +while the writer has not yet flushed it. Anything observing the flush itself +(a test counting flushes, for instance) must wait for the flush, not infer it +from frames arriving. The event drain loop (`WorkerPipeSession.RunEventDrainLoopAsync`) submits a whole drained event batch through `WriteBatchAsync`, which enqueues every frame under one `_gate` acquisition, takes the write lock once, and drains them together, so a burst of N events costs one flush rather than N — diff --git a/src/ZB.MOM.WW.MxGateway.Worker.Tests/Ipc/WorkerPipeSessionTests.cs b/src/ZB.MOM.WW.MxGateway.Worker.Tests/Ipc/WorkerPipeSessionTests.cs index 208c54a..82110a8 100644 --- a/src/ZB.MOM.WW.MxGateway.Worker.Tests/Ipc/WorkerPipeSessionTests.cs +++ b/src/ZB.MOM.WW.MxGateway.Worker.Tests/Ipc/WorkerPipeSessionTests.cs @@ -1214,7 +1214,9 @@ public sealed class WorkerPipeSessionTests using CancellationTokenSource cancellation = new(TimeSpan.FromSeconds(15)); using PipePair pipePair = await PipePair.CreateAsync(cancellation.Token); FakeRuntimeSession runtime = new(); - // A far-off heartbeat interval keeps heartbeat flushes out of the measurement window. + // A far-off heartbeat interval keeps every heartbeat after the first out of the measurement + // window; the first beat is sent immediately on entering the message loop (see + // RunHeartbeatLoopAsync) and is closed out explicitly below. FlushCountingPassthroughStream countingStream = new(pipePair.WorkerStream); WorkerPipeSession session = CreatePipeSession( countingStream, @@ -1227,8 +1229,18 @@ public sealed class WorkerPipeSessionTests Task runTask = session.RunAsync(cancellation.Token); await CompleteGatewayHandshakeAsync(pipePair, cancellation.Token); - // Let the idle drain loop settle (no events yet → no flushes) and record the baseline. - await Task.Delay(100, cancellation.Token); + // Close the pre-burst window on an explicit signal rather than a sleep. Three frames precede + // the burst — WorkerHello, WorkerReady, and the immediate first heartbeat — and each is + // flushed only after its bytes are already on the pipe, so having read a frame is no evidence + // that its flush has been counted. Read the first beat (the last pre-burst frame), then wait + // for the writer's deferred flush before sampling the baseline. A fixed delay left the first + // beat's flush free to land after the sample under CI scheduling pressure, where it was + // charged to the burst and the assertion below saw two flushes instead of one. + await ReadUntilAsync( + pipePair.GatewayReader, + WorkerEnvelope.BodyOneofCase.WorkerHeartbeat, + cancellation.Token); + await countingStream.WaitForAllWritesFlushedAsync(cancellation.Token); int baselineFlushes = countingStream.FlushCount; // Enqueue a full 128-event batch atomically so the drain loop sees it as one batch. @@ -1250,8 +1262,15 @@ public sealed class WorkerPipeSessionTests cancellation.Token); } - // The whole burst cost exactly one additional flush. - Assert.Equal(1, countingStream.FlushCount - baselineFlushes); + // Take the same edge on the burst's own flush — the 128 frames are on the wire before the + // writer flushes them — then assert on the recorded flush shape: exactly one flush beyond the + // baseline, and that flush carried every event frame of the burst. Asserting the shape (not + // just the count) is what makes the coalescing claim faithful: a split batch would show its + // first flush carrying fewer than the whole burst. + await countingStream.WaitForAllWritesFlushedAsync(cancellation.Token); + IReadOnlyList flushWriteCounts = countingStream.SnapshotFlushWriteCounts(); + Assert.Equal(baselineFlushes + 1, flushWriteCounts.Count); + Assert.Equal(burst, flushWriteCounts[baselineFlushes]); await SendShutdownAndWaitAsync(pipePair, runTask, cancellation.Token); } @@ -2154,13 +2173,22 @@ public sealed class WorkerPipeSessionTests } } - // Wraps the worker side of the pipe and counts FlushAsync calls so a test can assert the event - // drain loop coalesces a burst into a single flush. Delegates every other operation to the inner - // stream; does not own the inner stream's lifetime (PipePair disposes it). + // Wraps the worker side of the pipe and records the flush shape of the frames the writer emits — + // how many stream writes each flush coalesced — so a test can assert the event drain loop turns a + // burst into a single flush. Delegates every other operation to the inner stream; does not own the + // inner stream's lifetime (PipePair disposes it). + // + // The writer flushes only after writing a whole drained batch, so the peer can read every frame of + // that batch before the flush runs. Neither "I read the frames" nor any fixed sleep is evidence + // that a flush has been counted; WaitForAllWritesFlushedAsync is the explicit edge a test must + // take before sampling the counters. private sealed class FlushCountingPassthroughStream : Stream { private readonly Stream inner; - private int flushCount; + private readonly object gate = new(); + private readonly List flushWriteCounts = new(); + private readonly List waiters = new(); + private int writesSinceLastFlush; /// Initializes the passthrough over the given inner stream. /// The stream to delegate to. @@ -2169,8 +2197,53 @@ public sealed class WorkerPipeSessionTests this.inner = inner; } - /// Gets the number of calls observed so far. - public int FlushCount => Volatile.Read(ref flushCount); + /// Gets the number of flushes observed so far. + public int FlushCount + { + get + { + lock (gate) + { + return flushWriteCounts.Count; + } + } + } + + /// + /// Returns the number of stream writes coalesced into each flush, in flush order. + /// + /// A snapshot of the per-flush write counts. + public IReadOnlyList SnapshotFlushWriteCounts() + { + lock (gate) + { + return flushWriteCounts.ToArray(); + } + } + + /// + /// Completes once every write issued so far has been flushed, giving the caller a + /// happens-before edge on the writer's deferred flush instead of a timing guess. + /// + /// Token to abandon the wait. + /// A task that completes when no write is left unflushed. + public Task WaitForAllWritesFlushedAsync(CancellationToken cancellationToken) + { + FlushWaiter waiter; + lock (gate) + { + if (writesSinceLastFlush == 0) + { + return Task.CompletedTask; + } + + // Any flush drains every pending write, so the next flush is exactly the edge wanted. + waiter = new FlushWaiter(flushWriteCounts.Count + 1); + waiters.Add(waiter); + } + + return AwaitFlushWaiterAsync(waiter, cancellationToken); + } /// public override bool CanRead => inner.CanRead; @@ -2194,14 +2267,14 @@ public sealed class WorkerPipeSessionTests /// public override void Flush() { - Interlocked.Increment(ref flushCount); + RecordFlush(); inner.Flush(); } /// public override Task FlushAsync(CancellationToken cancellationToken) { - Interlocked.Increment(ref flushCount); + RecordFlush(); return inner.FlushAsync(cancellationToken); } @@ -2219,11 +2292,87 @@ public sealed class WorkerPipeSessionTests public override void SetLength(long value) => inner.SetLength(value); /// - public override void Write(byte[] buffer, int offset, int count) => inner.Write(buffer, offset, count); + public override void Write(byte[] buffer, int offset, int count) + { + RecordWrite(); + inner.Write(buffer, offset, count); + } /// public override Task WriteAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken) - => inner.WriteAsync(buffer, offset, count, cancellationToken); + { + RecordWrite(); + return inner.WriteAsync(buffer, offset, count, cancellationToken); + } + + private static async Task AwaitFlushWaiterAsync( + FlushWaiter waiter, + CancellationToken cancellationToken) + { + using (cancellationToken.Register(() => waiter.Completion.TrySetCanceled())) + { + await waiter.Completion.Task.ConfigureAwait(false); + } + } + + // Counted at write issue, not completion: the peer can observe the bytes as soon as the write + // is issued, so the pending count must already reflect the write by then. + private void RecordWrite() + { + lock (gate) + { + writesSinceLastFlush++; + } + } + + private void RecordFlush() + { + List? released = null; + lock (gate) + { + flushWriteCounts.Add(writesSinceLastFlush); + writesSinceLastFlush = 0; + for (int index = waiters.Count - 1; index >= 0; index--) + { + if (waiters[index].TargetFlushCount <= flushWriteCounts.Count) + { + released ??= new List(); + released.Add(waiters[index]); + waiters.RemoveAt(index); + } + } + } + + if (released is null) + { + return; + } + + foreach (FlushWaiter waiter in released) + { + waiter.Completion.TrySetResult(true); + } + } + + // A pending WaitForAllWritesFlushedAsync call: completes once the recorded flush count + // reaches TargetFlushCount. + private sealed class FlushWaiter + { + /// Initializes a waiter released at the given flush ordinal. + /// Flush count that releases the waiter. + public FlushWaiter(int targetFlushCount) + { + TargetFlushCount = targetFlushCount; + Completion = new TaskCompletionSource( + TaskCreationOptions.RunContinuationsAsynchronously); + } + + /// Gets the flush count at which this waiter completes. + public int TargetFlushCount { get; } + + /// Gets the completion signaled when the target flush count is reached. + public TaskCompletionSource Completion { get; } + } } private sealed class PipePair : IDisposable