From 6bc3f9b9918cdaa7bedad2e32d0367bbde3b52cf Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Fri, 7 Aug 2026 07:09:13 -0400 Subject: [PATCH] fix(WRK-21): make drain budget monotonic at the reserve boundary; guard the reply-too-large fallback MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review follow-ups on the WRK-21 cluster. 1. ResolveDrainReplyByteBudget was a step function, not a floor: just above the 64 KiB reserve the budget collapsed to a few bytes (at the validator-permitted floor MaxMessageBytes = 1024 + 64 KiB it was exactly 1024), too small to move a byte-heavy event, so DrainEvents truncated on every call and the drain-until- empty loop never terminated. It now takes the max of (frameMax - reserve) and frameMax/2, so the budget is monotonic and never below half the frame max. New test DrainEvents_AtValidatorFloorFrameMax_MakesProgressAndTerminates drives a byte-heavy queue at the exact validator floor and asserts it drains to empty with no head ever reported oversized. 2. The reply-too-large fallback write is now itself size-guarded (WriteReplyTooLargeFallbackAsync, used by both the control and STA reply seams): at a pathologically tiny negotiated max below the gateway's floor the fallback could also throw MessageTooLarge and — uncaught — kill the session, defeating the "no diagnostics command is session-fatal" invariant. It now log-and-swallows; comment notes WRK-24 adds the negotiated-max lower bound that makes it unreachable. 3. Corrected the RepeatedFieldOverheadBytes doc comments: WorkerEvent.CalculateSize() already includes the event's tag and length prefix (the same shape the reply's repeated events field packs), so the 8 bytes is pure slack over an already- conservative estimate, not compensation for a missing wrapper. --- .../Ipc/WorkerPipeSessionTests.cs | 70 +++++++++++++++++++ .../Ipc/WorkerPipeSession.cs | 62 +++++++++++++--- .../MxAccess/MxAccessEventQueue.cs | 19 +++-- 3 files changed, 134 insertions(+), 17 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 504cae8..ff9020e 100644 --- a/src/ZB.MOM.WW.MxGateway.Worker.Tests/Ipc/WorkerPipeSessionTests.cs +++ b/src/ZB.MOM.WW.MxGateway.Worker.Tests/Ipc/WorkerPipeSessionTests.cs @@ -633,6 +633,76 @@ public sealed class WorkerPipeSessionTests await SendShutdownAndWaitAsync(pipePair, runTask, cancellation.Token); } + /// + /// Regression for the reserve-boundary budget bug. The gateway validator accepts a Worker + /// frame maximum as low as 1024 + 64 KiB, and just above that boundary a naive + /// subtract-then-guard budget collapses to ~1024 bytes — too small to move even one + /// byte-heavy event, so every drain reports truncation with the same head blocked and the + /// drain-until-empty loop never terminates. The budget is now a floor (never below half the + /// negotiated maximum), so a byte-heavy queue drains to empty even at the validator floor. + /// + /// A task that represents the asynchronous operation. + [Fact] + public async Task DrainEvents_AtValidatorFloorFrameMax_MakesProgressAndTerminates() + { + // The lowest Worker.MaxMessageBytes GatewayOptionsValidator permits: the public gRPC floor + // (1024) plus the 64 KiB envelope-overhead reserve. The naive budget would be exactly 1024 + // here; the floored budget is half of the frame max (~33 KiB). + const uint validatorFloorFrameMax = 1024 + (64 * 1024); + using CancellationTokenSource cancellation = new(TimeSpan.FromSeconds(60)); + using PipePair pipePair = await PipePair.CreateAsync(cancellation.Token); + FakeRuntimeSession runtime = new() + { + SuppressDrainForBatchSize = 128, + BackingQueue = CreateByteHeavyQueue(200, ByteHeavyEventPayloadBytes), + }; + WorkerPipeSession session = CreatePipeSession(pipePair.WorkerStream, runtime); + Task runTask = session.RunAsync(cancellation.Token); + await CompleteGatewayHandshakeAsync(pipePair, validatorFloorFrameMax, cancellation.Token); + + int recovered = 0; + int replyCount = 0; + while (true) + { + string correlationId = $"floor-drain-{replyCount}"; + await pipePair.GatewayWriter + .WriteAsync( + CreateControlCommandEnvelope( + correlationId, + MxCommandKind.DrainEvents, + command => command.DrainEvents = new DrainEventsCommand { MaxEvents = 0 }), + cancellation.Token); + + WorkerEnvelope replyEnvelope = await ReadUntilAsync( + pipePair.GatewayReader, + WorkerEnvelope.BodyOneofCase.WorkerCommandReply, + envelope => envelope.WorkerCommandReply.Reply.CorrelationId == correlationId, + cancellation.Token); + replyCount++; + + MxCommandReply reply = replyEnvelope.WorkerCommandReply.Reply; + Assert.True( + replyEnvelope.CalculateSize() <= validatorFloorFrameMax, + $"reply {replyCount} serialized to {replyEnvelope.CalculateSize()} bytes."); + + int drainedThisReply = reply.DrainEvents.Events.Count; + if (drainedThisReply == 0) + { + break; + } + + // The head is never reported as oversized at this frame max: the ~33 KiB floored budget + // comfortably fits the ~1.8 KiB events, so each reply makes real progress. + Assert.DoesNotContain("alone exceeds", reply.DiagnosticMessage); + recovered += drainedThisReply; + Assert.True(replyCount < 200, "DrainEvents made no progress at the validator floor frame max."); + } + + Assert.Equal(200, recovered); + + await SendShutdownAndWaitAsync(pipePair, runTask, cancellation.Token); + } + /// /// Verifies the control-reply write seam is not session-fatal on size (WRK-21 backstop). The /// reply builders size their payloads, so this path needs a deliberately budget-blind drain diff --git a/src/ZB.MOM.WW.MxGateway.Worker/Ipc/WorkerPipeSession.cs b/src/ZB.MOM.WW.MxGateway.Worker/Ipc/WorkerPipeSession.cs index e584ae5..b8fdd5d 100644 --- a/src/ZB.MOM.WW.MxGateway.Worker/Ipc/WorkerPipeSession.cs +++ b/src/ZB.MOM.WW.MxGateway.Worker/Ipc/WorkerPipeSession.cs @@ -587,10 +587,45 @@ public sealed class WorkerPipeSession when (exception.ErrorCode == WorkerFrameProtocolErrorCode.MessageTooLarge) { LogControlReplyTooLarge(correlationId, kind, exception); + await WriteReplyTooLargeFallbackAsync(correlationId, kind, cancellationToken) + .ConfigureAwait(false); + } + } + + /// + /// Writes the small InvalidRequest reply that answers a correlation whose real reply + /// overshot the frame maximum. The fallback itself is a handful of bytes, so it fits any + /// sane negotiated maximum; the only way it can also throw MessageTooLarge is a + /// pathologically tiny negotiated maximum below the gateway's validation floor — the + /// pre-existing WRK-24 gap, which adds the negotiated-max lower bound that makes this + /// unreachable. Until then, a defensive swallow keeps the "no diagnostics command is + /// session-fatal" invariant true even in that degenerate config: the correlation goes + /// unanswered and the gateway's own per-command timeout covers it, but the session lives. + /// + private async Task WriteReplyTooLargeFallbackAsync( + string correlationId, + MxCommandKind kind, + CancellationToken cancellationToken) + { + try + { await WriteControlReplyAsync( CreateReplyTooLargeReply(correlationId, kind), cancellationToken).ConfigureAwait(false); } + catch (WorkerFrameProtocolException exception) + when (exception.ErrorCode == WorkerFrameProtocolErrorCode.MessageTooLarge) + { + _logger?.Error( + "WorkerControlReplyFallbackTooLarge", + new Dictionary + { + ["correlation_id"] = correlationId, + ["command_kind"] = kind.ToString(), + ["max_message_bytes"] = _options.MaxMessageBytes, + ["reason"] = exception.Message, + }); + } } private void LogControlReplyTooLarge( @@ -715,15 +750,21 @@ public sealed class WorkerPipeSession /// /// Byte budget for the events packed into one DrainEvents reply: the negotiated frame - /// maximum less a fixed reserve for the envelope/reply wrapper. A negotiated maximum below - /// the reserve would otherwise yield a non-positive budget and stall the drain forever, so - /// a tiny frame maximum falls back to half of itself — still ample headroom for a wrapper - /// measured in tens of bytes. + /// maximum less a fixed reserve for the envelope/reply wrapper, but never below half the + /// negotiated maximum. The lower bound must be a floor, not a step: a bare + /// subtract-then-guard-positive collapses the budget to a handful of bytes just above + /// the reserve (e.g. at the validator-permitted floor MaxMessageBytes = 1024 + 64 KiB the + /// subtraction leaves 1024, too small to move even one byte-heavy event, so every drain + /// truncates and the drain-until-empty caller never terminates). Taking the max with + /// half the negotiated maximum keeps the budget monotonic across the reserve boundary while + /// still leaving the full reserve for the wrapper whenever the frame max is large enough + /// that the reserve is the smaller subtraction — which is every configuration above 128 KiB. /// private int ResolveDrainReplyByteBudget() { - int budget = _options.MaxMessageBytes - DrainReplyFrameHeadroomBytes; - return budget > 0 ? budget : _options.MaxMessageBytes / 2; + return Math.Max( + _options.MaxMessageBytes - DrainReplyFrameHeadroomBytes, + _options.MaxMessageBytes / 2); } private static string CreateDrainTruncationMessage( @@ -826,11 +867,12 @@ public sealed class WorkerPipeSession { // An oversized STA command reply is a property of that one command, not of the // session. Answer the correlation with an error reply instead of falling into the - // generic catch below, which would fault the whole session for it. + // generic catch below, which would fault the whole session for it. The fallback + // write is itself size-guarded (see WriteReplyTooLargeFallbackAsync) so a degenerate + // negotiated maximum cannot make even this backstop session-fatal. LogControlReplyTooLarge(envelope.CorrelationId, command.Kind, sizeException); - await WriteControlReplyAsync( - CreateReplyTooLargeReply(envelope.CorrelationId, command.Kind), - cancellationToken).ConfigureAwait(false); + await WriteReplyTooLargeFallbackAsync(envelope.CorrelationId, command.Kind, cancellationToken) + .ConfigureAwait(false); } } catch (Exception exception) when (exception is not OperationCanceledException) diff --git a/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/MxAccessEventQueue.cs b/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/MxAccessEventQueue.cs index e8d65f4..ed27a67 100644 --- a/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/MxAccessEventQueue.cs +++ b/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/MxAccessEventQueue.cs @@ -26,9 +26,13 @@ public sealed class MxAccessEventQueue /// public const int DefaultCapacity = 10000; - // Per-event allowance added to WorkerEvent.CalculateSize() when charging the byte budget in - // Drain(maxEvents, maxTotalBytes): conservatively covers the repeated-field tag byte and the - // length varint the event costs once packed into DrainEventsReply. + // Extra per-event slack added to WorkerEvent.CalculateSize() when charging the byte budget in + // Drain(maxEvents, maxTotalBytes). CalculateSize() already accounts for the event's own tag and + // length-delimiter (the WorkerEvent wrapper serializes MxEvent as field 1, and the reply packs + // each MxEvent as DrainEventsReply.events field 1 with the identical tag+length shape), so this + // is a pure safety margin over an already-conservative estimate — not compensation for a missing + // wrapper. It keeps the running total strictly ahead of the true serialized size so a rounding + // edge can never push the packed reply past the frame maximum. private const int RepeatedFieldOverheadBytes = 8; private readonly int capacity; @@ -221,10 +225,11 @@ public sealed class MxAccessEventQueue /// /// The size decision happens inside the queue lock, so an event is dequeued only once it is /// known to fit: an event that does not fit stays at the head for the next call and is never - /// lost (WRK-21). Per-event cost is WorkerEvent.CalculateSize() plus - /// ; the wrapper slightly - /// overestimates the packed MxEvent and the constant conservatively covers the - /// repeated-field tag and length varint, so the estimate errs strictly on the safe side. + /// lost (WRK-21). Per-event cost is WorkerEvent.CalculateSize() — which already + /// includes the event's own tag and length prefix, the same shape the reply's + /// events repeated field packs it into — plus + /// of pure slack, so the running total stays strictly ahead of the true serialized size and + /// the estimate errs on the safe side. /// /// Maximum number of events to drain; 0 means "no count limit". /// Byte budget for the drained events' estimated serialized size.