From a2565604df228e039830ac84b9c3941225200c45 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Fri, 7 Aug 2026 06:50:17 -0400 Subject: [PATCH] test(WRK-21): keep the drain-to-empty walk inside the pipe harness envelope PipePair runs both ends of a duplex pipe in one process with blocking FlushFileBuffers under every frame write, so it wedges after roughly 85 large round trips. Drain the full 10,000 byte-heavy events to empty at the queue layer, where the no-loss property actually lives, and keep the pipe walk at 1,000 events (29 replies) so it still proves the split end to end. Also give the truncation test's budget slack: item handle 0 is a proto3 default and is not serialized, so the probe measurement is a lower bound on the fixture's per-event cost. --- .../Ipc/WorkerPipeSessionTests.cs | 22 +++---- .../MxAccess/MxAccessEventQueueTests.cs | 60 ++++++++++++++++++- 2 files changed, 70 insertions(+), 12 deletions(-) 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 56fe252..504cae8 100644 --- a/src/ZB.MOM.WW.MxGateway.Worker.Tests/Ipc/WorkerPipeSessionTests.cs +++ b/src/ZB.MOM.WW.MxGateway.Worker.Tests/Ipc/WorkerPipeSessionTests.cs @@ -20,15 +20,17 @@ public sealed class WorkerPipeSessionTests private const string Nonce = "nonce-secret"; // Byte-heavy drain fixture (WRK-21). 10,000 events at ~1.7 KiB each is ~17 MB of queue — far - // more than one frame — so DrainEvents must split across replies. + // more than one frame — so DrainEvents must truncate. // - // The negotiated frame maximum is deliberately small. What is under test is the byte cap, and - // it behaves identically at any frame size, but this harness is not the production gateway: - // PipePair has no continuous read pump, so the test thread only drains the pipe while it sits - // in ReadUntilAsync. Multi-megabyte frames interleaved with the heartbeat loop can therefore - // wedge both ends inside FlushFileBuffers, each waiting for the other to read. A frame maximum - // well under the pipe buffer keeps the harness honest without weakening a single assertion. + // Two limits below are harness accommodations, not properties of the fix. PipePair runs both + // ends of a duplex pipe inside one process, with no continuous read pump and with blocking + // FlushFileBuffers under every frame write, so it tolerates neither multi-megabyte frames nor + // hundreds of large round trips before both ends wedge waiting on each other. Hence a small + // negotiated frame maximum, and a smaller queue for the drain-to-empty walk. The byte cap + // behaves identically at any frame size; exhaustive no-loss over the full 10,000 events is + // covered without a pipe by MxAccessEventQueueTests. private const int ByteHeavyEventCount = 10_000; + private const int RepeatedDrainEventCount = 1_000; private const int ByteHeavyEventPayloadBytes = 1_800; private const uint NegotiatedMaxFrameBytes = 128 * 1024; @@ -576,7 +578,7 @@ public sealed class WorkerPipeSessionTests FakeRuntimeSession runtime = new() { SuppressDrainForBatchSize = 128, - BackingQueue = CreateByteHeavyQueue(ByteHeavyEventCount, ByteHeavyEventPayloadBytes), + BackingQueue = CreateByteHeavyQueue(RepeatedDrainEventCount, ByteHeavyEventPayloadBytes), }; WorkerPipeSession session = CreatePipeSession(pipePair.WorkerStream, runtime); Task runTask = session.RunAsync(cancellation.Token); @@ -616,13 +618,13 @@ public sealed class WorkerPipeSessionTests recovered.Add(drained.WorkerSequence); } - Assert.True(replyCount < 1_000, "DrainEvents made no progress across 1,000 replies."); + Assert.True(replyCount < 200, "DrainEvents made no progress across 200 replies."); } // More than one reply proves the drain really split; every event came back exactly once, in // enqueue order. Assert.True(replyCount > 2, $"Expected the byte cap to split the drain, saw {replyCount} replies."); - Assert.Equal(ByteHeavyEventCount, recovered.Count); + Assert.Equal(RepeatedDrainEventCount, recovered.Count); for (int index = 0; index < recovered.Count; index++) { Assert.Equal((ulong)(index + 1), recovered[index]); diff --git a/src/ZB.MOM.WW.MxGateway.Worker.Tests/MxAccess/MxAccessEventQueueTests.cs b/src/ZB.MOM.WW.MxGateway.Worker.Tests/MxAccess/MxAccessEventQueueTests.cs index a405e7f..5c3d639 100644 --- a/src/ZB.MOM.WW.MxGateway.Worker.Tests/MxAccess/MxAccessEventQueueTests.cs +++ b/src/ZB.MOM.WW.MxGateway.Worker.Tests/MxAccess/MxAccessEventQueueTests.cs @@ -142,9 +142,12 @@ public sealed class MxAccessEventQueueTests queue.Enqueue(CreateEventWithPayload(itemHandle, payloadLength: 256)); } + // One-and-a-half events' worth of budget: the head fits, the next does not, and the next is + // comfortably smaller than the whole budget so it is a plain truncation rather than the + // oversized-head case. WorkerEventDrainResult result = queue.Drain( maxEvents: 0, - maxTotalBytes: MeasureDrainCost(payloadLength: 256)); + maxTotalBytes: MeasureDrainCost(payloadLength: 256) * 3 / 2); Assert.Single(result.Events); Assert.True(result.TruncatedBySize); @@ -175,6 +178,57 @@ public sealed class MxAccessEventQueueTests Assert.Equal(2, queue.Count); } + /// + /// The no-loss half of the WRK-21 acceptance criterion, at full scale. Draining the review's + /// 10,000 byte-heavy events under a budget that fits only a fraction of them per call must + /// return every event exactly once and in order: the pre-fix drain removed events from the + /// queue before the reply was sized, so a rejected frame destroyed them. This runs at the + /// queue layer because the property is the queue's, and because the pipe harness that covers + /// the same walk end to end cannot sustain hundreds of large round trips. + /// + [Fact] + public void Drain_ByteBudget_RepeatedCalls_RecoverAllEventsInOrderWithoutLoss() + { + const int eventCount = 10_000; + const int payloadLength = 1_800; + MxAccessEventQueue queue = new(eventCount); + for (int index = 0; index < eventCount; index++) + { + queue.Enqueue(CreateEventWithPayload(index, payloadLength)); + } + + // A budget that fits roughly 35 events, so the walk takes hundreds of calls. + int budget = MeasureDrainCost(payloadLength) * 35; + List recovered = new(); + int calls = 0; + while (true) + { + WorkerEventDrainResult result = queue.Drain(maxEvents: 0, maxTotalBytes: budget); + calls++; + if (result.Events.Count == 0) + { + break; + } + + foreach (WorkerEvent drained in result.Events) + { + recovered.Add(drained.Event.WorkerSequence); + } + + Assert.Equal(eventCount - recovered.Count, result.RemainingCount); + Assert.True(calls < eventCount, "Drain made no progress."); + } + + Assert.True(calls > 100, $"Expected the byte budget to split the drain, saw {calls} calls."); + Assert.Equal(eventCount, recovered.Count); + for (int index = 0; index < recovered.Count; index++) + { + Assert.Equal((ulong)(index + 1), recovered[index]); + } + + Assert.Equal(0, queue.Count); + } + /// /// Verifies the count cap still binds when the byte budget is generous: the byte cap is an /// additional bound, not a replacement. @@ -254,7 +308,9 @@ public sealed class MxAccessEventQueueTests /// /// Measures what the queue charges one event of the given payload size against the byte budget: /// the serialized as it exists after Enqueue (sequence and timestamp - /// stamped) plus the repeated-field allowance. + /// stamped) plus the repeated-field allowance. The probe uses item handle 0, a proto3 default + /// that is not serialized, so this is a lower bound on the fixtures' real per-event cost — the + /// budgets above carry slack rather than assuming byte equality. /// /// Length of the event's raw-status payload string. /// The per-event byte cost.