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.