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.
This commit is contained in:
@@ -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]);
|
||||
|
||||
@@ -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);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 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.
|
||||
/// </summary>
|
||||
[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<ulong> 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);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 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
|
||||
/// <summary>
|
||||
/// Measures what the queue charges one event of the given payload size against the byte budget:
|
||||
/// the serialized <see cref="WorkerEvent"/> 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.
|
||||
/// </summary>
|
||||
/// <param name="payloadLength">Length of the event's raw-status payload string.</param>
|
||||
/// <returns>The per-event byte cost.</returns>
|
||||
|
||||
Reference in New Issue
Block a user