diff --git a/archreview/2026-07-12/remediation/00-tracking.md b/archreview/2026-07-12/remediation/00-tracking.md index 73bdcb8..ac9af59 100644 --- a/archreview/2026-07-12/remediation/00-tracking.md +++ b/archreview/2026-07-12/remediation/00-tracking.md @@ -43,9 +43,9 @@ Sequenced by cluster; a cluster is one change set. | GWC-25 | Medium | S | CLI-35/36 (coord) | Not started | Empty-ring ReplayGap sentinel carries `oldest_available_sequence = 0`, dead-streaming a compliant client | | CLI-35 | Medium | S | GWC-25 (coord) | Not started | Python CLI `stream-events` crashes on a ReplayGap | | CLI-36 | Medium | S | GWC-25 (coord) | Not started | Go CLI `stream-events` silently destroys the ReplayGap signal | -| WRK-21 | Medium | M | owns IPC-23 fix; WRK-28 same batch | Not started | DrainEvents bound is count-based only; oversized reply kills the session and loses the drained events | -| IPC-23 | Medium | S | WRK-21 | Not started | DrainEvents contract requirements (reply fits negotiated max, no event loss, drain-until-empty) + proto-comment/doc wave | -| IPC-30 | Low | M | WRK-21 (same batch) | Not started | Oversized event frame stays session-fatal by design, but the death becomes structured (fault frame + logged identity) | +| WRK-21 | Medium | M | owns IPC-23 fix; WRK-28 same batch | Done | DrainEvents bound is count-based only; oversized reply kills the session and loses the drained events | +| IPC-23 | Medium | S | WRK-21 | In progress — mechanics landed with WRK-21; proto-comment/doc wave pending | DrainEvents contract requirements (reply fits negotiated max, no event loss, drain-until-empty) + proto-comment/doc wave | +| IPC-30 | Low | M | WRK-21 (same batch) | Done | Oversized event frame stays session-fatal by design, but the death becomes structured (fault frame + logged identity) | | SEC-31 | Medium | M | — | Not started | Failure limiter partitions on attacker-controlled key id and blocks before verification (lockout DoS) | | SEC-32 | Low | S | SEC-31 | Not started | Failure-limiter LRU flushable by junk-token spray; token prefix never validated | | IPC-24 | Medium | S | — | Not started | CI's unconditional Java churn-revert masks real drift | @@ -71,27 +71,27 @@ Full design + implementation for each row lives in the linked domain doc under i | ID | Sev | Tier | Eff | Dep | Status | Title | |---|---|:-:|:-:|---|---|---| -| WRK-21 | Medium | P0 | M | IPC-23 (fix owned here); WRK-28 | Not started | DrainEvents bound count-based only; oversized reply kills session and loses drained events | +| WRK-21 | Medium | P0 | M | IPC-23 (fix owned here); WRK-28 | Done | DrainEvents bound count-based only; oversized reply kills session and loses drained events | | WRK-22 | Low | — | S | IPC-26 (fix owned here) | Not started | Cancelled `WriteAsync` leaves its frame queued; it is still written later | -| WRK-23 | Low | — | S | WRK-21 | Not started | Rejected frames consume sequence numbers, producing wire gaps | +| WRK-23 | Low | — | S | WRK-21 | Done | Rejected frames consume sequence numbers, producing wire gaps | | WRK-24 | Low | — | S | — | Not started | `AdoptNegotiatedMaxMessageBytes` has no lower-bound sanity check | | WRK-25 | Low | P2 | S | WRK-22 (shared seam) | Not started | WRK-12 flush coalescing never engages on the event hot path | | WRK-26 | Low | P1 | S | WRK-23 (soft); discharges IPC-29 | Not started | Write-priority and overflow doc drift from the WRK-07 change | | WRK-27 | Low | — | S | — | Not started | Alarm poll bypasses the watchdog's in-flight suppression (15 s vs 75 s) | -| WRK-28 | Low | — | S | WRK-21 (same batch) | Not started | 10,000 drain cap is a duplicated magic constant | +| WRK-28 | Low | — | S | WRK-21 (same batch) | Done | 10,000 drain cap is a duplicated magic constant | ### Contracts & IPC — [30-contracts-ipc.md](30-contracts-ipc.md) | ID | Sev | Tier | Eff | Dep | Status | Title | |---|---|:-:|:-:|---|---|---| -| IPC-23 | Medium | P0 | S | WRK-21 (mechanics) | Not started | DrainEvents byte-blindness — contract requirements + proto-comment/doc wave | +| IPC-23 | Medium | P0 | S | WRK-21 (mechanics) | In progress — mechanics landed with WRK-21; proto-comment/doc wave pending | DrainEvents byte-blindness — contract requirements + proto-comment/doc wave | | IPC-24 | Medium | P0 | S | — | Not started | CI's unconditional Java churn-revert masks real drift | | IPC-25 | Medium | P0 | M | — | Not started | Stale Go/Python worker bindings: regenerate (pinned toolchains) + check-codegen Check 4 | | IPC-26 | Low | P2 | S | WRK-22 (mechanics) | Not started | Cancelled write leaves ghost frame — cancelled means never written | | IPC-27 | Low | P2 | S | — | Not started | Descriptor-freshness test blind to enums/services/galaxy descriptor | | IPC-28 | Low | — | S | — | Not started | docs/Grpc.md missing CommandTooLarge → ResourceExhausted mapping | | IPC-29 | Low | — | S | WRK-26 (discharged by) | Not started | WorkerFrameProtocol.md missing write-scheduling/sequencing section | -| IPC-30 | Low | P0 | M | WRK-21 (same batch) | Not started | Oversized event frame: keep session-fatal, make the death structured | +| IPC-30 | Low | P0 | M | WRK-21 (same batch) | Done | Oversized event frame: keep session-fatal, make the death structured | | IPC-31 | Info | — | — | — | N/A | Gateway creation-time sequence stamping accepted; diagnostic-only, decision recorded | | IPC-32 | Info | — | S | IPC-25 (folded in) | Not started | check-codegen banner relabel 1/4…4/4 | diff --git a/archreview/2026-07-12/remediation/20-worker.md b/archreview/2026-07-12/remediation/20-worker.md index 1f22aa3..8b03ccb 100644 --- a/archreview/2026-07-12/remediation/20-worker.md +++ b/archreview/2026-07-12/remediation/20-worker.md @@ -16,14 +16,14 @@ members, no positional records). The worker builds and tests only on the Windows | ID | Sev | Tier | Eff | Dep | Status | Title | |----|-----|------|-----|-----|--------|-------| -| WRK-21 | Medium | P0 | M | IPC-23 (same defect, fix owned here); WRK-28 (same lines) | Not started | DrainEvents bound is count-based only; an oversized reply still kills the session and loses the drained events | +| WRK-21 | Medium | P0 | M | IPC-23 (same defect, fix owned here); WRK-28 (same lines) | Done | DrainEvents bound is count-based only; an oversized reply still kills the session and loses the drained events | | WRK-22 | Low | — | S | IPC-26 (same defect, fix owned here) | Not started | Cancelled `WriteAsync` leaves its frame queued; it is still written later | -| WRK-23 | Low | — | S | WRK-21 (rejection path becomes backstop-only) | Not started | Rejected frames consume sequence numbers, producing wire gaps | +| WRK-23 | Low | — | S | WRK-21 (rejection path becomes backstop-only) | Done | Rejected frames consume sequence numbers, producing wire gaps | | WRK-24 | Low | — | S | — | Not started | `AdoptNegotiatedMaxMessageBytes` has no lower-bound sanity check | | WRK-25 | Low | P2 | S | WRK-22 (both touch enqueue/dequeue) | Not started | WRK-12 flush coalescing never engages on the event hot path | | WRK-26 | Low | P1 | S | WRK-23 (soft — sequence prose); discharges IPC-29 | Not started | Write-priority and overflow doc drift from the WRK-07 change | | WRK-27 | Low | — | S | — | Not started | Alarm poll bypasses the watchdog's in-flight suppression (15 s vs 75 s) | -| WRK-28 | Low | — | S | WRK-21 (land in the same commit cluster) | Not started | 10,000 drain cap is a duplicated magic constant with a comment-only sync contract | +| WRK-28 | Low | — | S | WRK-21 (land in the same commit cluster) | Done | 10,000 drain cap is a duplicated magic constant with a comment-only sync contract | --- diff --git a/archreview/2026-07-12/remediation/30-contracts-ipc.md b/archreview/2026-07-12/remediation/30-contracts-ipc.md index 94f4fda..e495d42 100644 --- a/archreview/2026-07-12/remediation/30-contracts-ipc.md +++ b/archreview/2026-07-12/remediation/30-contracts-ipc.md @@ -12,14 +12,14 @@ All `path:line` citations were re-verified against the working tree at `4f5371f` | ID | Sev | Tier | Eff | Dep | Status | Title | |----|-----|------|-----|-----|--------|-------| -| IPC-23 | Medium | P0 | S¹ | WRK-21 | Not started | DrainEvents bound is count-based only; byte-heavy queue still builds a session-killing reply frame (contract requirements here; fix mechanics in WRK-21) | +| IPC-23 | Medium | P0 | S¹ | WRK-21 | In progress — mechanics landed with WRK-21; proto-comment/doc wave pending | DrainEvents bound is count-based only; byte-heavy queue still builds a session-killing reply frame (contract requirements here; fix mechanics in WRK-21) | | IPC-24 | Medium | P0 | S | — | Not started | CI's unconditional Java churn-revert masks real generated-code drift for message-level proto changes | | IPC-25 | Medium | P0 | M | — | Not started | Committed Go/Python worker bindings are stale at HEAD; no guard covers them | | IPC-26 | Low | P2 | S¹ | WRK-22 | Not started | Cancelled write leaves a ghost frame that is still written (contract requirement here; fix mechanics in WRK-22) | | IPC-27 | Low | P2 | S | — | Not started | Descriptor freshness test blind to enums, enum values, services/methods, and the Galaxy contract | | IPC-28 | Low | — | S | — | Not started | `docs/Grpc.md` omits the `CommandTooLarge` → `ResourceExhausted` mapping | | IPC-29 | Low | — | S | — | Not started | Worker writer priority scheduling and write-time sequence stamping undocumented in the frame-protocol doc | -| IPC-30 | Low | P0 | M | WRK-21 (same file/batch) | Not started | Oversized worker→gateway event frame is session-fatal — make the death deliberate, structured, and diagnosable | +| IPC-30 | Low | P0 | M | WRK-21 (same file/batch) | Done | Oversized worker→gateway event frame is session-fatal — make the death deliberate, structured, and diagnosable | | IPC-31 | Info | — | — | — | N/A | Gateway stamps sequence at creation, worker at write — accepted divergence; sequence is documented diagnostic-only (`gateway.md:328-330`); revisit only if sequence ever becomes load-bearing | | IPC-32 | Info | — | S | IPC-25 | Not started | `check-codegen.ps1` check labels miscounted (folded into the IPC-25 script edit) | diff --git a/docs/MxAccessWorkerInstanceDesign.md b/docs/MxAccessWorkerInstanceDesign.md index 936601d..0693b2b 100644 --- a/docs/MxAccessWorkerInstanceDesign.md +++ b/docs/MxAccessWorkerInstanceDesign.md @@ -378,6 +378,20 @@ If event conversion throws, catch it inside the event handler, record a structured `WorkerFault`, and keep the worker alive only if the fault policy allows it. +The event drain loop streams queued events as `WorkerEvent` frames. A single +event whose envelope exceeds the negotiated frame maximum is **undeliverable end +to end** — the pipe maximum sits only the envelope-overhead reserve above the +public gRPC cap, so a frame the pipe rejects would also be rejected on the +client-facing stream. The session therefore faults on it rather than dropping it +(a silent drop makes the event stream unfaithful, and a synthesized placeholder +is barred by the no-synthesized-events rule), but the death is structured: the +worker logs the event's identity — family, handles, worker sequence, and sizes, +never the value — writes a `WorkerFault` with category `ProtocolViolation` and +command method `EventDrain` carrying the same identity, and only then exits. +Operator remediation is configuration: raise `MxGateway:Worker:MaxMessageBytes` +for that workload. Other per-frame rejection codes keep their previous behavior +because they indicate worker bugs, not workload size. + ## Command Queue The pipe reader converts `WorkerCommand` messages into `StaCommand` entries. @@ -440,6 +454,29 @@ Diagnostics: - `DrainEvents` - `ShutdownWorker` +`DrainEvents` is answered on the message-loop thread, not the STA, and its reply +is bounded on **two** axes because no diagnostics command may be session-fatal: + +- **Count** — `GatewayContractInfo.MaxDrainEventsPerCommand` (10,000) is the + single home of the ceiling, shared by the gateway's request validator (which + rejects a larger `max_events` at the public boundary) and this worker clamp + (which also interprets `max_events = 0`, "as many as available"). +- **Bytes** — the count cap alone is not sufficient: byte-heavy events (large + string or array `MxValue`s) overshoot the negotiated frame maximum long before + 10,000 events. The drain is therefore byte-budgeted against the negotiated + maximum less a 64 KiB envelope/reply-wrapper reserve, and the size decision + happens inside the event queue's lock, so an event is dequeued only once it is + known to fit. An event that does not fit stays at the head of the queue and is + never lost. + +Truncation is reported in the reply's existing `DiagnosticMessage` +("N events returned, M remain; repeat DrainEvents for the rest") rather than in a +new field, so the contract is unchanged and callers drain iteratively until a +reply comes back empty. In the degenerate case where the head event alone exceeds +the budget, the reply returns whatever fit before it (possibly nothing) and names +the blocked event's worker sequence so an operator can find the offending tag; +that event needs a larger `MxGateway:Worker:MaxMessageBytes` to move at all. + Implement method-specific dispatch instead of a generic string method invoker. Parity tests need stable command-specific request and reply shapes. @@ -623,6 +660,14 @@ queue fills: Production coalescing may be added later, but it must be explicit and tested. Do not drop or coalesce events in v1. +No control reply is session-fatal on size. Reply builders size their payloads +against the negotiated frame maximum, and the two reply-write seams (the +control-command path and the STA command path) additionally catch a +`MessageTooLarge` per-frame rejection and answer that correlation with a small +`InvalidRequest` reply instead of unwinding the session. Oversized *event* +frames keep the opposite policy — see Event Sink — because an event above the +frame maximum cannot be delivered to the client at all. + ## Heartbeat And Watchdog `WorkerPipeSession` starts the heartbeat loop after the gateway validates diff --git a/docs/WorkerFrameProtocol.md b/docs/WorkerFrameProtocol.md index b74ff6e..e03867e 100644 --- a/docs/WorkerFrameProtocol.md +++ b/docs/WorkerFrameProtocol.md @@ -29,6 +29,29 @@ default. A `max_frame_bytes` of 0 (an older gateway that never set the field) means "use the worker's built-in default". This keeps both ends framing to the same limit rather than depending on matched compile-time constants. +Every worker-to-gateway frame must serialize within this limit, control replies +included, so reply builders truncate to fit rather than emit a frame the writer +will reject. `WorkerPipeSession` pre-sizes a `DrainEvents` reply below the +negotiated maximum (less a fixed envelope/reply-wrapper reserve) and reports the +truncation in the reply's `DiagnosticMessage`; the caller contract is to repeat +`DrainEvents` until it returns an empty reply. Should a reply still overshoot, +`MessageTooLarge` at the reply-write seam is answered with a small +`InvalidRequest` reply for that correlation, not with session teardown — no +diagnostics command may kill a session. + +An oversized *event* frame is the deliberate exception. Such an event is +undeliverable end to end (the pipe maximum sits only the envelope-overhead +reserve above the public gRPC cap), so the session faults: the worker logs the +event's identity and sizes — never its value — writes a `WorkerFault` with +category `ProtocolViolation` and command method `EventDrain`, then exits. +Remediation is raising `MxGateway:Worker:MaxMessageBytes` for that workload. + +A per-frame rejection does not consume an envelope `sequence`. The writer stamps +a candidate sequence, runs the empty-payload and size checks against the stamped +envelope, and commits the counter only immediately before the stream write, so +the sequences observed on the wire stay contiguous across rejections and an +operator reading a pipe capture never sees a phantom gap. + ## Envelope Validation `WorkerFrameReader` and `WorkerFrameWriter` validate each envelope against the diff --git a/gateway.md b/gateway.md index 9f1e452..38e300e 100644 --- a/gateway.md +++ b/gateway.md @@ -447,10 +447,14 @@ Optional diagnostics: - `Ping` - `GetSessionState` - `GetWorkerInfo` -- `DrainEvents` — diagnostic; `max_events` is bounded (the gateway rejects requests - above a public ceiling, and the worker caps each reply at its own per-reply limit, - treating `max_events = 0` as "the default cap") so one drain cannot pack an - unbounded, session-killing reply frame. +- `DrainEvents` — diagnostic; the reply is bounded on two axes so one drain cannot + pack an unbounded, session-killing reply frame. By **count**: the gateway rejects + requests above the shared ceiling `GatewayContractInfo.MaxDrainEventsPerCommand` + and the worker clamps to the same value, treating `max_events = 0` as "the default + cap". By **bytes**: the worker sizes the reply while draining, against the + negotiated worker-frame maximum, so a byte-heavy queue truncates instead of + overshooting and events that do not fit stay queued. Truncation is reported in the + reply's `DiagnosticMessage`; callers drain iteratively until an empty reply. - `ShutdownWorker` Do not compress MXAccess semantics into generic verbs too early. A command enum diff --git a/src/ZB.MOM.WW.MxGateway.Contracts/GatewayContractInfo.cs b/src/ZB.MOM.WW.MxGateway.Contracts/GatewayContractInfo.cs index ef698d2..0d070c1 100644 --- a/src/ZB.MOM.WW.MxGateway.Contracts/GatewayContractInfo.cs +++ b/src/ZB.MOM.WW.MxGateway.Contracts/GatewayContractInfo.cs @@ -17,6 +17,21 @@ public static class GatewayContractInfo /// Default backend name identifying the MXAccess worker process type. public const string DefaultBackendName = "mxaccess-worker"; + /// + /// Ceiling on how many events one DrainEvents command may move in a single reply. + /// Shared so the gateway's request-validation ceiling + /// (MxAccessGrpcRequestValidator, which rejects a larger max_events loudly at + /// the public boundary) and the worker's per-reply clamp + /// (WorkerPipeSession.CreateDrainEventsReply, the backstop that also interprets + /// max_events = 0) cannot drift apart. A count cap alone is necessary but not + /// sufficient: the worker additionally caps the reply by serialized bytes against the + /// negotiated frame maximum (WRK-21), so a reply may carry fewer events than this ceiling + /// and fewer than are queued. Callers drain iteratively until an empty reply. + /// This is a documented behavioral bound, not wire schema — it is deliberately a C# + /// constant and not a .proto field. + /// + public const uint MaxDrainEventsPerCommand = 10_000; + /// /// Environment variable name that opts an xUnit suite into running live /// MXAccess COM tests. Single source of truth shared by both diff --git a/src/ZB.MOM.WW.MxGateway.Server/Grpc/MxAccessGrpcRequestValidator.cs b/src/ZB.MOM.WW.MxGateway.Server/Grpc/MxAccessGrpcRequestValidator.cs index 5a53786..1ac1f89 100644 --- a/src/ZB.MOM.WW.MxGateway.Server/Grpc/MxAccessGrpcRequestValidator.cs +++ b/src/ZB.MOM.WW.MxGateway.Server/Grpc/MxAccessGrpcRequestValidator.cs @@ -1,17 +1,11 @@ using Grpc.Core; +using ZB.MOM.WW.MxGateway.Contracts; using ZB.MOM.WW.MxGateway.Contracts.Proto; namespace ZB.MOM.WW.MxGateway.Server.Grpc; public sealed class MxAccessGrpcRequestValidator { - // Upper bound on a single DrainEvents request. DrainEvents is a diagnostics RPC that returns - // buffered events in one non-streaming reply, so an unbounded max_events could pack the whole - // queue into a session-killing frame. The worker independently caps each reply at its - // own MaxDrainEventsPerReply; this public bound rejects an obviously-abusive request loudly at - // the boundary. max_events = 0 is allowed and means "the worker's default batch cap". - private const uint MaxDrainEventsPerRequest = 10_000; - /// Validates an open session request. /// The request to validate. public void ValidateOpenSession(OpenSessionRequest request) @@ -78,10 +72,18 @@ public sealed class MxAccessGrpcRequestValidator } // The payload case now matches the kind, so command.DrainEvents is non-null here. - if (command.Kind is MxCommandKind.DrainEvents && command.DrainEvents.MaxEvents > MaxDrainEventsPerRequest) + // DrainEvents is a diagnostics RPC that returns buffered events in one non-streaming + // reply, so an unbounded max_events could pack the whole queue into a session-killing + // frame. The worker independently clamps every reply to the same shared ceiling and + // additionally caps it by serialized bytes; this public bound rejects an obviously-abusive + // request loudly at the boundary. max_events = 0 is allowed and means "the worker's + // default batch cap". + if (command.Kind is MxCommandKind.DrainEvents + && command.DrainEvents.MaxEvents > GatewayContractInfo.MaxDrainEventsPerCommand) { throw InvalidArgument( - $"DrainEvents max_events ({command.DrainEvents.MaxEvents}) must not exceed {MaxDrainEventsPerRequest}; " + $"DrainEvents max_events ({command.DrainEvents.MaxEvents}) must not exceed " + + $"{GatewayContractInfo.MaxDrainEventsPerCommand}; " + "use 0 to request the worker default batch cap."); } } diff --git a/src/ZB.MOM.WW.MxGateway.Worker.Tests/Ipc/WorkerFrameProtocolTests.cs b/src/ZB.MOM.WW.MxGateway.Worker.Tests/Ipc/WorkerFrameProtocolTests.cs index 9570823..6cd43d3 100644 --- a/src/ZB.MOM.WW.MxGateway.Worker.Tests/Ipc/WorkerFrameProtocolTests.cs +++ b/src/ZB.MOM.WW.MxGateway.Worker.Tests/Ipc/WorkerFrameProtocolTests.cs @@ -410,6 +410,48 @@ public sealed class WorkerFrameProtocolTests } } + /// + /// Verifies a per-frame rejection does not burn a sequence number (WRK-23). The sequence is a + /// diagnostic counter, so a gap breaks nothing functionally — but an operator correlating a pipe + /// capture reads a gap as a lost frame and chases a bug that does not exist, and the gap-free + /// guarantee the concurrent-write test asserts would otherwise only hold until the first + /// rejection. + /// + /// A task that represents the asynchronous operation. + [Fact] + public async Task WriteAsync_PerFrameRejection_DoesNotConsumeSequence() + { + const int maxMessageBytes = 512; + WorkerFrameProtocolOptions options = new( + SessionId, + GatewayContractInfo.WorkerProtocolVersion, + Nonce, + maxMessageBytes); + using MemoryStream stream = new(); + WorkerFrameWriter writer = new(stream, options); + + await writer.WriteAsync(CreateEventEnvelope()); + + WorkerEnvelope oversized = CreateGatewayHelloEnvelope(); + oversized.GatewayHello.GatewayVersion = new string('x', maxMessageBytes * 2); + WorkerFrameProtocolException exception = + await Assert.ThrowsAsync( + async () => await writer.WriteAsync(oversized)); + Assert.Equal(WorkerFrameProtocolErrorCode.MessageTooLarge, exception.ErrorCode); + + await writer.WriteAsync(CreateEventEnvelope()); + + stream.Position = 0; + WorkerFrameReader reader = new(stream, options); + WorkerEnvelope first = await reader.ReadAsync(); + WorkerEnvelope second = await reader.ReadAsync(); + + // Two frames reached the wire; the rejected frame in between left no gap. + Assert.Equal(1UL, first.Sequence); + Assert.Equal(2UL, second.Sequence); + Assert.Equal(stream.Length, stream.Position); + } + /// Verifies a zero negotiated frame maximum keeps the constructor default. [Fact] public void AdoptNegotiatedMaxMessageBytes_WithZero_KeepsDefault() 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 4b71b16..65e8468 100644 --- a/src/ZB.MOM.WW.MxGateway.Worker.Tests/Ipc/WorkerPipeSessionTests.cs +++ b/src/ZB.MOM.WW.MxGateway.Worker.Tests/Ipc/WorkerPipeSessionTests.cs @@ -19,6 +19,14 @@ public sealed class WorkerPipeSessionTests private const string SessionId = "session-1"; 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. The negotiated frame maximum + // is deliberately smaller than the compile-time default so the split happens in a handful of + // multi-MB frames instead of moving 17 MB through the test pipe. + private const int ByteHeavyEventCount = 10_000; + private const int ByteHeavyEventPayloadBytes = 1_800; + private const uint NegotiatedMaxFrameBytes = 2 * 1024 * 1024; + /// Verifies that valid gateway hello triggers worker hello and ready responses. /// A task that represents the asynchronous operation. [Fact] @@ -487,6 +495,291 @@ public sealed class WorkerPipeSessionTests await SendShutdownAndWaitAsync(pipePair, runTask, cancellation.Token); } + /// + /// The WRK-21 repro. A queue full of byte-heavy events (large string values — the payload + /// profile this gateway exists for) used to make DrainEvents max_events = 0 build a + /// reply above the negotiated frame maximum: the writer rejected the frame, the exception + /// unwound the session, and the already-dequeued events were gone. The drain is now + /// byte-budgeted, so the reply fits, the truncation is reported in the reply's diagnostic + /// message, and the session keeps serving. + /// + /// A task that represents the asynchronous operation. + [Fact] + public async Task DrainEvents_ByteHeavyQueue_ReplyIsBoundedAndSessionSurvives() + { + using CancellationTokenSource cancellation = new(TimeSpan.FromSeconds(60)); + using PipePair pipePair = await PipePair.CreateAsync(cancellation.Token); + FakeRuntimeSession runtime = new() + { + SuppressDrainForBatchSize = 128, + BackingQueue = CreateByteHeavyQueue(ByteHeavyEventCount, ByteHeavyEventPayloadBytes), + }; + WorkerPipeSession session = CreatePipeSession(pipePair.WorkerStream, runtime); + Task runTask = session.RunAsync(cancellation.Token); + await CompleteGatewayHandshakeAsync(pipePair, NegotiatedMaxFrameBytes, cancellation.Token); + + await pipePair.GatewayWriter + .WriteAsync( + CreateControlCommandEnvelope( + "drain-heavy-1", + MxCommandKind.DrainEvents, + command => command.DrainEvents = new DrainEventsCommand { MaxEvents = 0 }), + cancellation.Token); + + WorkerEnvelope replyEnvelope = await ReadUntilAsync( + pipePair.GatewayReader, + WorkerEnvelope.BodyOneofCase.WorkerCommandReply, + cancellation.Token); + + MxCommandReply reply = replyEnvelope.WorkerCommandReply.Reply; + Assert.Equal(ProtocolStatusCode.Ok, reply.ProtocolStatus.Code); + + // The whole queue is far larger than one frame, so the reply is a strict subset that fits. + Assert.True( + replyEnvelope.CalculateSize() <= NegotiatedMaxFrameBytes, + $"DrainEvents reply serialized to {replyEnvelope.CalculateSize()} bytes, above the negotiated {NegotiatedMaxFrameBytes}."); + Assert.InRange(reply.DrainEvents.Events.Count, 1, ByteHeavyEventCount - 1); + Assert.Contains("remain", reply.DiagnosticMessage); + Assert.Contains("repeat DrainEvents", reply.DiagnosticMessage); + + // The session is alive: it still answers a ping, and RunAsync has not unwound. + await pipePair.GatewayWriter + .WriteAsync(CreatePingCommandEnvelope("ping-after-drain", "still-here"), cancellation.Token); + WorkerEnvelope pingReply = await ReadUntilAsync( + pipePair.GatewayReader, + WorkerEnvelope.BodyOneofCase.WorkerCommandReply, + envelope => envelope.WorkerCommandReply.Reply.CorrelationId == "ping-after-drain", + cancellation.Token); + Assert.Equal("still-here", pingReply.WorkerCommandReply.Reply.DiagnosticMessage); + Assert.False(runTask.IsCompleted, "The session must survive a byte-heavy DrainEvents."); + + await SendShutdownAndWaitAsync(pipePair, runTask, cancellation.Token); + } + + /// + /// Verifies the byte-budgeted drain loses nothing: repeating DrainEvents until it comes back + /// empty recovers every enqueued event exactly once, in order, across the split replies. The + /// pre-fix drain removed events from the queue before sizing the reply, so a rejected frame + /// destroyed them — no-loss is the half of the P0 criterion a catch-only fix cannot deliver. + /// + /// A task that represents the asynchronous operation. + [Fact] + public async Task DrainEvents_RepeatedCalls_RecoverAllEventsWithoutLoss() + { + using CancellationTokenSource cancellation = new(TimeSpan.FromSeconds(90)); + using PipePair pipePair = await PipePair.CreateAsync(cancellation.Token); + FakeRuntimeSession runtime = new() + { + SuppressDrainForBatchSize = 128, + BackingQueue = CreateByteHeavyQueue(ByteHeavyEventCount, ByteHeavyEventPayloadBytes), + }; + WorkerPipeSession session = CreatePipeSession(pipePair.WorkerStream, runtime); + Task runTask = session.RunAsync(cancellation.Token); + await CompleteGatewayHandshakeAsync(pipePair, NegotiatedMaxFrameBytes, cancellation.Token); + + List recovered = new(); + int replyCount = 0; + while (true) + { + string correlationId = $"drain-loop-{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() <= NegotiatedMaxFrameBytes, + $"DrainEvents reply {replyCount} serialized to {replyEnvelope.CalculateSize()} bytes."); + if (reply.DrainEvents.Events.Count == 0) + { + break; + } + + foreach (MxEvent drained in reply.DrainEvents.Events) + { + recovered.Add(drained.WorkerSequence); + } + + Assert.True(replyCount < 100, "DrainEvents made no progress across 100 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); + for (int index = 0; index < recovered.Count; index++) + { + Assert.Equal((ulong)(index + 1), recovered[index]); + } + + 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 + /// to reach — but that is the point: a future command or a sizing bug must degrade to an + /// error reply for that correlation, never to a dead session. + /// + /// A task that represents the asynchronous operation. + [Fact] + public async Task ControlReplyTooLarge_WritesErrorReplyInsteadOfDying() + { + const uint tinyMaxFrameBytes = 4096; + using CancellationTokenSource cancellation = new(TimeSpan.FromSeconds(15)); + using PipePair pipePair = await PipePair.CreateAsync(cancellation.Token); + FakeRuntimeSession runtime = new() + { + SuppressDrainForBatchSize = 128, + BackingQueue = CreateByteHeavyQueue(eventCount: 1, payloadBytes: 16 * 1024), + IgnoreDrainByteBudget = true, + }; + WorkerPipeSession session = CreatePipeSession(pipePair.WorkerStream, runtime); + Task runTask = session.RunAsync(cancellation.Token); + await CompleteGatewayHandshakeAsync(pipePair, tinyMaxFrameBytes, cancellation.Token); + + await pipePair.GatewayWriter + .WriteAsync( + CreateControlCommandEnvelope( + "drain-too-large", + MxCommandKind.DrainEvents, + command => command.DrainEvents = new DrainEventsCommand { MaxEvents = 0 }), + cancellation.Token); + + WorkerEnvelope replyEnvelope = await ReadUntilAsync( + pipePair.GatewayReader, + WorkerEnvelope.BodyOneofCase.WorkerCommandReply, + cancellation.Token); + + MxCommandReply reply = replyEnvelope.WorkerCommandReply.Reply; + Assert.Equal("drain-too-large", reply.CorrelationId); + Assert.Equal(MxCommandKind.DrainEvents, reply.Kind); + Assert.Equal(ProtocolStatusCode.InvalidRequest, reply.ProtocolStatus.Code); + Assert.Contains("frame maximum", reply.ProtocolStatus.Message); + Assert.False(runTask.IsCompleted, "An oversized control reply must not end the session."); + + await SendShutdownAndWaitAsync(pipePair, runTask, cancellation.Token); + } + + /// + /// Verifies the same backstop on the STA command path: an oversized reply from a dispatched + /// command answers its correlation with an error reply instead of falling into the generic + /// catch that faults the whole session with MxaccessCommandFailed. + /// + /// A task that represents the asynchronous operation. + [Fact] + public async Task CommandReplyTooLarge_WritesErrorReplyInsteadOfFaulting() + { + const uint tinyMaxFrameBytes = 4096; + using CancellationTokenSource cancellation = new(TimeSpan.FromSeconds(15)); + using PipePair pipePair = await PipePair.CreateAsync(cancellation.Token); + FakeRuntimeSession runtime = new() + { + DispatchReplyDiagnosticMessage = new string('x', 16 * 1024), + }; + WorkerPipeSession session = CreatePipeSession(pipePair.WorkerStream, runtime); + Task runTask = session.RunAsync(cancellation.Token); + await CompleteGatewayHandshakeAsync(pipePair, tinyMaxFrameBytes, cancellation.Token); + + await pipePair.GatewayWriter + .WriteAsync(CreateCommandEnvelope("command-too-large"), cancellation.Token); + + WorkerEnvelope replyEnvelope = await ReadUntilAsync( + pipePair.GatewayReader, + WorkerEnvelope.BodyOneofCase.WorkerCommandReply, + envelope => envelope.WorkerCommandReply.Reply.CorrelationId == "command-too-large", + cancellation.Token); + + MxCommandReply reply = replyEnvelope.WorkerCommandReply.Reply; + Assert.Equal(MxCommandKind.Register, reply.Kind); + Assert.Equal(ProtocolStatusCode.InvalidRequest, reply.ProtocolStatus.Code); + + // No fault, and the session still reports itself Ready rather than Faulted. + await pipePair.GatewayWriter + .WriteAsync( + CreateControlCommandEnvelope( + "state-after-too-large", + MxCommandKind.GetSessionState, + command => command.GetSessionState = new GetSessionStateCommand()), + cancellation.Token); + WorkerEnvelope stateEnvelope = await ReadUntilAsync( + pipePair.GatewayReader, + WorkerEnvelope.BodyOneofCase.WorkerCommandReply, + envelope => envelope.WorkerCommandReply.Reply.CorrelationId == "state-after-too-large", + cancellation.Token); + Assert.Equal(SessionState.Ready, stateEnvelope.WorkerCommandReply.Reply.SessionState.State); + Assert.False(runTask.IsCompleted, "An oversized command reply must not fault the session."); + + await SendShutdownAndWaitAsync(pipePair, runTask, cancellation.Token); + } + + /// + /// IPC-30. An event above the negotiated frame maximum is undeliverable end to end (the pipe + /// maximum sits only an envelope reserve above the public gRPC cap), so the session stays + /// fatal by design — but the death must be structured: a WorkerFault naming the event, with + /// no value payload in it, before the process exits. Silently dropping the event or + /// synthesizing a placeholder were both rejected. + /// + /// A task that represents the asynchronous operation. + [Fact] + public async Task RunAsync_EventFrameTooLarge_WritesStructuredFaultThenEndsSession() + { + const uint tinyMaxFrameBytes = 4096; + using CancellationTokenSource cancellation = new(TimeSpan.FromSeconds(15)); + using PipePair pipePair = await PipePair.CreateAsync(cancellation.Token); + FakeRuntimeSession runtime = new(); + RecordingWorkerLogger logger = new(); + WorkerPipeSession session = CreatePipeSession( + pipePair.WorkerStream, + runtime, + new WorkerPipeSessionOptions + { + HeartbeatInterval = TimeSpan.FromMilliseconds(100), + HeartbeatGrace = TimeSpan.FromSeconds(5), + }, + logger); + runtime.EnqueueEvent(CreateOversizedWorkerEvent(sequence: 77, payloadBytes: 16 * 1024)); + Task runTask = session.RunAsync(cancellation.Token); + await CompleteGatewayHandshakeAsync(pipePair, tinyMaxFrameBytes, cancellation.Token); + + WorkerEnvelope faultEnvelope = await ReadUntilAsync( + pipePair.GatewayReader, + WorkerEnvelope.BodyOneofCase.WorkerFault, + cancellation.Token); + + WorkerFault fault = faultEnvelope.WorkerFault; + Assert.Equal(WorkerFaultCategory.ProtocolViolation, fault.Category); + Assert.Equal("EventDrain", fault.CommandMethod); + Assert.Contains("77", fault.DiagnosticMessage); + Assert.Contains("MaxMessageBytes", fault.DiagnosticMessage); + // The identity is reported; the value payload never is. + Assert.DoesNotContain(new string('x', 64), fault.DiagnosticMessage); + + Assert.Contains( + logger.Events, + entry => entry.EventName == "WorkerEventFrameTooLarge" + && entry.Fields.TryGetValue("worker_sequence", out object? sequence) + && sequence is ulong sequenceValue + && sequenceValue == 77UL); + + // The session ends, and the fault frame parsed cleanly off the same stream above — the + // rejected event never corrupted the wire. + Task completedTask = await Task.WhenAny(runTask, Task.Delay(TimeSpan.FromSeconds(5), cancellation.Token)); + Assert.Same(runTask, completedTask); + await Assert.ThrowsAsync(async () => await runTask); + } + /// /// Verifies that ShutdownWorker returns its OK reply BEFORE the graceful /// shutdown runs and disposes the runtime session, and that the message @@ -1241,6 +1534,15 @@ public sealed class WorkerPipeSessionTests Stream stream, FakeRuntimeSession runtime, WorkerPipeSessionOptions sessionOptions) + { + return CreatePipeSession(stream, runtime, sessionOptions, logger: null); + } + + private static WorkerPipeSession CreatePipeSession( + Stream stream, + FakeRuntimeSession runtime, + WorkerPipeSessionOptions sessionOptions, + ZB.MOM.WW.MxGateway.Worker.Bootstrap.IWorkerLogger? logger) { WorkerFrameProtocolOptions options = CreateOptions(); return new WorkerPipeSession( @@ -1249,7 +1551,8 @@ public sealed class WorkerPipeSessionTests options, () => 1234, sessionOptions, - () => runtime); + () => runtime, + logger); } private static WorkerFrameProtocolOptions CreateOptions() @@ -1270,7 +1573,8 @@ public sealed class WorkerPipeSessionTests private static WorkerEnvelope CreateGatewayHelloEnvelope( string nonce = Nonce, uint supportedProtocolVersion = GatewayContractInfo.WorkerProtocolVersion, - ulong sequence = 1) + ulong sequence = 1, + uint maxFrameBytes = 0) { return new WorkerEnvelope { @@ -1282,6 +1586,10 @@ public sealed class WorkerPipeSessionTests SupportedProtocolVersion = supportedProtocolVersion, Nonce = nonce, GatewayVersion = "test-gateway", + // 0 leaves the worker on its compile-time default; a non-zero value is adopted + // during the handshake and becomes the frame maximum every later assertion is + // measured against. + MaxFrameBytes = maxFrameBytes, }, }; } @@ -1392,12 +1700,55 @@ public sealed class WorkerPipeSessionTests }; } - private static async Task CompleteGatewayHandshakeAsync( + private static WorkerEvent CreateOversizedWorkerEvent(ulong sequence, int payloadBytes) + { + WorkerEvent workerEvent = CreateWorkerEvent(sequence); + workerEvent.Event.ItemHandle = 42; + workerEvent.Event.RawStatus = new string('x', payloadBytes); + return workerEvent; + } + + /// + /// Fills a real event queue with byte-heavy events — a large string field stands in for the + /// array/string MxValue payloads that make a count-capped drain overshoot the frame + /// maximum. The queue is real (not the fake's plain list) so the production byte-budgeting runs. + /// + /// Number of events to enqueue. + /// Size of each event's raw-status payload string. + /// The populated queue. + private static MxAccessEventQueue CreateByteHeavyQueue(int eventCount, int payloadBytes) + { + MxAccessEventQueue queue = new(Math.Max(eventCount, 1)); + string payload = new string('x', payloadBytes); + for (int index = 0; index < eventCount; index++) + { + queue.Enqueue(new MxEvent + { + SessionId = SessionId, + Family = MxEventFamily.OnDataChange, + ItemHandle = index, + RawStatus = payload, + OnDataChange = new OnDataChangeEvent(), + }); + } + + return queue; + } + + private static Task CompleteGatewayHandshakeAsync( PipePair pipePair, CancellationToken cancellationToken) + { + return CompleteGatewayHandshakeAsync(pipePair, maxFrameBytes: 0, cancellationToken); + } + + private static async Task CompleteGatewayHandshakeAsync( + PipePair pipePair, + uint maxFrameBytes, + CancellationToken cancellationToken) { await pipePair.GatewayWriter - .WriteAsync(CreateGatewayHelloEnvelope(), cancellationToken) + .WriteAsync(CreateGatewayHelloEnvelope(maxFrameBytes: maxFrameBytes), cancellationToken) .ConfigureAwait(false); WorkerEnvelope hello = await pipePair.GatewayReader.ReadAsync(cancellationToken).ConfigureAwait(false); 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 84f8e1d..a405e7f 100644 --- a/src/ZB.MOM.WW.MxGateway.Worker.Tests/MxAccess/MxAccessEventQueueTests.cs +++ b/src/ZB.MOM.WW.MxGateway.Worker.Tests/MxAccess/MxAccessEventQueueTests.cs @@ -98,6 +98,103 @@ public sealed class MxAccessEventQueueTests Assert.Equal(0, queue.Count); } + /// + /// Verifies the byte-budgeted drain stops before the budget is exceeded, leaves the + /// remainder queued in order, and reports the exact remaining count (WRK-21). Events that + /// do not fit must never be dequeued — dequeuing them is how the pre-fix drain lost events + /// when the reply frame was rejected. + /// + [Fact] + public void Drain_ByteBudget_StopsBeforeBudgetAndLeavesRemainderQueued() + { + MxAccessEventQueue queue = new(capacity: 8); + for (int itemHandle = 0; itemHandle < 5; itemHandle++) + { + queue.Enqueue(CreateEventWithPayload(itemHandle, payloadLength: 512)); + } + + int perEventCost = MeasureDrainCost(payloadLength: 512); + + // Budget for exactly two events (plus a sliver too small for a third). + IReadOnlyList drained = + queue.Drain(maxEvents: 0, maxTotalBytes: (perEventCost * 2) + (perEventCost / 2)).Events; + + Assert.Equal(2, drained.Count); + Assert.Equal(0, drained[0].Event.ItemHandle); + Assert.Equal(1, drained[1].Event.ItemHandle); + Assert.Equal(3, queue.Count); + + // The undrained remainder is still present, still in order. + IReadOnlyList rest = queue.Drain(maxEvents: 0); + Assert.Equal(new[] { 2, 3, 4 }, new[] { rest[0].Event.ItemHandle, rest[1].Event.ItemHandle, rest[2].Event.ItemHandle }); + } + + /// + /// Verifies the byte-budgeted drain reports truncation and the exact remaining count so the + /// DrainEvents reply can tell the caller to drain again. + /// + [Fact] + public void Drain_ByteBudget_ReportsTruncationAndRemainingCount() + { + MxAccessEventQueue queue = new(capacity: 8); + for (int itemHandle = 0; itemHandle < 4; itemHandle++) + { + queue.Enqueue(CreateEventWithPayload(itemHandle, payloadLength: 256)); + } + + WorkerEventDrainResult result = queue.Drain( + maxEvents: 0, + maxTotalBytes: MeasureDrainCost(payloadLength: 256)); + + Assert.Single(result.Events); + Assert.True(result.TruncatedBySize); + Assert.Equal(3, result.RemainingCount); + Assert.Equal(0UL, result.OversizedHeadSequence); + } + + /// + /// Verifies the degenerate case: a head event whose own serialized size exceeds the whole + /// budget is not drained (draining it would build an oversized reply or lose the event) and + /// its worker sequence is reported so an operator can find the offending tag. + /// + [Fact] + public void Drain_ByteBudget_OversizedHead_DrainsNothingAndReportsHeadSequence() + { + MxAccessEventQueue queue = new(capacity: 8); + queue.Enqueue(CreateEventWithPayload(itemHandle: 0, payloadLength: 4096)); + queue.Enqueue(CreateEventWithPayload(itemHandle: 1, payloadLength: 8)); + + WorkerEventDrainResult result = queue.Drain(maxEvents: 0, maxTotalBytes: 1024); + + Assert.Empty(result.Events); + Assert.True(result.TruncatedBySize); + Assert.Equal(2, result.RemainingCount); + Assert.Equal(1UL, result.OversizedHeadSequence); + + // The blocked event is still queued — it was never removed. + Assert.Equal(2, queue.Count); + } + + /// + /// Verifies the count cap still binds when the byte budget is generous: the byte cap is an + /// additional bound, not a replacement. + /// + [Fact] + public void Drain_ByteBudget_CountCapStillBinds() + { + MxAccessEventQueue queue = new(capacity: 8); + for (int itemHandle = 0; itemHandle < 5; itemHandle++) + { + queue.Enqueue(CreateEventWithPayload(itemHandle, payloadLength: 16)); + } + + WorkerEventDrainResult result = queue.Drain(maxEvents: 2, maxTotalBytes: 1024 * 1024); + + Assert.Equal(2, result.Events.Count); + Assert.False(result.TruncatedBySize); + Assert.Equal(3, result.RemainingCount); + } + /// Verifies that Enqueue is rejected after a fault is recorded manually. [Fact] public void Enqueue_AfterRecordFault_ThrowsInvalidOperationException() @@ -149,6 +246,40 @@ public sealed class MxAccessEventQueueTests Assert.Equal(WorkerFaultCategory.MxaccessEventConversionFailed, queue.Fault?.Category); } + // Mirrors MxAccessEventQueue's per-event repeated-field allowance. Kept local (and asserted + // through the budgets below) so a change to the queue's charge shows up as a failing bound + // rather than silently loosening these tests. + private const int RepeatedFieldOverheadBytes = 8; + + /// + /// 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. + /// + /// Length of the event's raw-status payload string. + /// The per-event byte cost. + private static int MeasureDrainCost(int payloadLength) + { + MxAccessEventQueue probe = new(capacity: 1); + probe.Enqueue(CreateEventWithPayload(0, payloadLength)); + Assert.True(probe.TryDequeue(out WorkerEvent? probeEvent)); + return probeEvent!.CalculateSize() + RepeatedFieldOverheadBytes; + } + + /// + /// Builds a byte-heavy event: a large string field is the cheapest stand-in for the array/string + /// payloads that make a count-capped drain overshoot the frame maximum. + /// + /// Item handle identifying the event in assertions. + /// Length of the raw-status payload string. + /// The constructed event. + private static MxEvent CreateEventWithPayload(int itemHandle, int payloadLength) + { + MxEvent mxEvent = CreateEvent(MxEventFamily.OnDataChange, itemHandle); + mxEvent.RawStatus = new string('x', payloadLength); + return mxEvent; + } + private static MxEvent CreateEvent( MxEventFamily family, int itemHandle) diff --git a/src/ZB.MOM.WW.MxGateway.Worker.Tests/TestSupport/FakeRuntimeSession.cs b/src/ZB.MOM.WW.MxGateway.Worker.Tests/TestSupport/FakeRuntimeSession.cs index fb0d611..142551f 100644 --- a/src/ZB.MOM.WW.MxGateway.Worker.Tests/TestSupport/FakeRuntimeSession.cs +++ b/src/ZB.MOM.WW.MxGateway.Worker.Tests/TestSupport/FakeRuntimeSession.cs @@ -43,6 +43,13 @@ internal sealed class FakeRuntimeSession : IWorkerRuntimeSession /// Gets or sets whether ShutdownGracefullyAsync throws a TimeoutException. public bool ThrowTimeoutOnShutdown { get; set; } + /// + /// Optional diagnostic message stuffed into every dispatched command reply. A long value + /// pushes the STA command reply past a small negotiated frame maximum, which is how a test + /// drives the ProcessCommandAsync reply-size backstop. + /// + public string? DispatchReplyDiagnosticMessage { get; set; } + /// Gets a value indicating whether Dispose was called. public bool Disposed { get; private set; } @@ -92,7 +99,7 @@ internal sealed class FakeRuntimeSession : IWorkerRuntimeSession throw new InvalidOperationException("Command failed after shutdown started."); } - return new MxCommandReply + MxCommandReply reply = new() { SessionId = command.SessionId, CorrelationId = command.CorrelationId, @@ -103,6 +110,13 @@ internal sealed class FakeRuntimeSession : IWorkerRuntimeSession Message = "OK", }, }; + + if (DispatchReplyDiagnosticMessage is not null) + { + reply.DiagnosticMessage = DispatchReplyDiagnosticMessage; + } + + return reply; }); } @@ -133,6 +147,27 @@ internal sealed class FakeRuntimeSession : IWorkerRuntimeSession /// public uint? LastDrainMaxEvents { get; private set; } + /// + /// Optional real event queue backing the drain paths. When set, both + /// and delegate to it + /// so a test can exercise the production byte-budgeting logic behind the fake session. + /// + public MxAccessEventQueue? BackingQueue { get; set; } + + /// + /// When set, ignores the byte budget and drains purely + /// by count. Simulates the "sizing bug or future command" case the control-reply size + /// backstop exists for, so a test can drive an oversized reply without a real budgeting + /// defect. + /// + public bool IgnoreDrainByteBudget { get; set; } + + /// + /// Records the maxTotalBytes argument of the most recent byte-budgeted + /// call. + /// + public int? LastDrainMaxTotalBytes { get; private set; } + /// public IReadOnlyList DrainEvents(uint maxEvents) { @@ -143,6 +178,76 @@ internal sealed class FakeRuntimeSession : IWorkerRuntimeSession LastDrainMaxEvents = maxEvents; + if (BackingQueue is not null) + { + return BackingQueue.Drain(maxEvents); + } + + lock (gate) + { + int drainCount = maxEvents == 0 + ? events.Count + : Math.Min(events.Count, checked((int)Math.Min(maxEvents, int.MaxValue))); + List drained = new(drainCount); + for (int index = 0; index < drainCount; index++) + { + drained.Add(events.Dequeue()); + } + + return drained; + } + } + + /// + public WorkerEventDrainResult DrainEvents(uint maxEvents, int maxTotalBytes) + { + if (SuppressDrainForBatchSize is uint suppressed && maxEvents == suppressed) + { + return new WorkerEventDrainResult( + Array.Empty(), + truncatedBySize: false, + remainingCount: PendingEventCount, + oversizedHeadSequence: 0); + } + + LastDrainMaxEvents = maxEvents; + LastDrainMaxTotalBytes = maxTotalBytes; + + if (BackingQueue is not null && !IgnoreDrainByteBudget) + { + return BackingQueue.Drain(maxEvents, maxTotalBytes); + } + + // Count-only drain: either no backing queue (the simple fakes) or a deliberately + // budget-blind drain used to exercise the reply-size backstop. + IReadOnlyList drained = BackingQueue is not null + ? BackingQueue.Drain(maxEvents) + : DrainByCount(maxEvents); + return new WorkerEventDrainResult( + drained, + truncatedBySize: false, + remainingCount: PendingEventCount, + oversizedHeadSequence: 0); + } + + private int PendingEventCount + { + get + { + if (BackingQueue is not null) + { + return BackingQueue.Count; + } + + lock (gate) + { + return events.Count; + } + } + } + + private IReadOnlyList DrainByCount(uint maxEvents) + { lock (gate) { int drainCount = maxEvents == 0 diff --git a/src/ZB.MOM.WW.MxGateway.Worker/Ipc/WorkerFrameWriter.cs b/src/ZB.MOM.WW.MxGateway.Worker/Ipc/WorkerFrameWriter.cs index dbe101c..6a377a7 100644 --- a/src/ZB.MOM.WW.MxGateway.Worker/Ipc/WorkerFrameWriter.cs +++ b/src/ZB.MOM.WW.MxGateway.Worker/Ipc/WorkerFrameWriter.cs @@ -43,8 +43,9 @@ public sealed class WorkerFrameWriter private readonly Queue _eventFrames = new Queue(); // Only ever read/written by the current write-lock holder while draining, so no interlock is - // needed. Starts at 0 and is pre-incremented, so the first written frame carries sequence 1 - // (matching the previous behaviour). + // needed. Starts at 0 and is committed only immediately before the stream write, so the first + // written frame carries sequence 1 and a per-frame rejection leaves the counter untouched — + // the next accepted frame reuses the number and the wire sequence stays contiguous. private ulong _nextSequence; /// Initializes a new instance of the WorkerFrameWriter class. @@ -237,7 +238,14 @@ public sealed class WorkerFrameWriter // Stamp the sequence at the actual point of writing, under the write lock, so the wire order // and the stamped sequence agree regardless of caller concurrency or priority. - envelope.Sequence = unchecked(++_nextSequence); + // + // Peek-stamp-commit (WRK-23): the sequence participates in CalculateSize() (varint width), + // so it must be stamped before the size checks — but a per-frame rejection must not burn a + // number, or the wire shows phantom gaps that an operator reads as lost frames. Stamp a + // candidate, validate the stamped envelope, and commit the counter only once the frame is + // certain to be written. + ulong candidateSequence = unchecked(_nextSequence + 1); + envelope.Sequence = candidateSequence; int payloadLength = envelope.CalculateSize(); if (payloadLength == 0) @@ -254,6 +262,8 @@ public sealed class WorkerFrameWriter $"Worker envelope payload length {payloadLength} exceeds the configured maximum of {_options.MaxMessageBytes} bytes."); } + _nextSequence = candidateSequence; + // Serialize once into a single buffer that carries the 4-byte length prefix followed by the // payload, then issue one stream write. This avoids a second serialization pass, a separate // prefix array, and a separate prefix write. The flush is deferred to the end of the drained diff --git a/src/ZB.MOM.WW.MxGateway.Worker/Ipc/WorkerPipeSession.cs b/src/ZB.MOM.WW.MxGateway.Worker/Ipc/WorkerPipeSession.cs index b9629ab..e584ae5 100644 --- a/src/ZB.MOM.WW.MxGateway.Worker/Ipc/WorkerPipeSession.cs +++ b/src/ZB.MOM.WW.MxGateway.Worker/Ipc/WorkerPipeSession.cs @@ -5,6 +5,7 @@ using System.IO; using System.Threading; using System.Threading.Tasks; using Google.Protobuf.WellKnownTypes; +using ZB.MOM.WW.MxGateway.Contracts; using ZB.MOM.WW.MxGateway.Contracts.Proto; using ZB.MOM.WW.MxGateway.Worker.Bootstrap; using ZB.MOM.WW.MxGateway.Worker.MxAccess; @@ -18,12 +19,11 @@ public sealed class WorkerPipeSession private static readonly TimeSpan BackgroundTaskStopTimeout = TimeSpan.FromSeconds(1); private const uint EventDrainBatchSize = 128; - // Hard cap on how many events a single DrainEvents diagnostic reply may carry. DrainEvents is a - // non-streaming control command, so an unbounded drain (including the max_events = 0 "as many as - // available" request) could pack the whole queue into one session-killing reply frame. - // The gateway request validator rejects requests above its public ceiling; this worker-side cap is - // the backstop and defines the effective per-reply maximum. Kept in step with that public ceiling. - private const uint MaxDrainEventsPerReply = 10_000; + // Headroom subtracted from the negotiated frame maximum when budgeting a DrainEvents reply. It + // covers the WorkerEnvelope/WorkerCommandReply/MxCommandReply wrapper the drained events are + // packed into — the same envelope-overhead reserve rationale docs/WorkerFrameProtocol.md + // records for the frame max itself. + private const int DrainReplyFrameHeadroomBytes = 64 * 1024; private readonly WorkerFrameProtocolOptions _options; private readonly Func _processIdProvider; @@ -376,13 +376,83 @@ public sealed class WorkerPipeSession // Events are the low-priority frame class: the writer holds them behind any pending // control frame (reply, fault, heartbeat, shutdown ack) so those are not delayed // behind an event backlog. - await _writer - .WriteAsync(CreateEnvelope(workerEvent), WorkerFrameWritePriority.Event, cancellationToken) - .ConfigureAwait(false); + try + { + await _writer + .WriteAsync(CreateEnvelope(workerEvent), WorkerFrameWritePriority.Event, cancellationToken) + .ConfigureAwait(false); + } + catch (WorkerFrameProtocolException exception) + when (exception.ErrorCode == WorkerFrameProtocolErrorCode.MessageTooLarge) + { + await FaultOnOversizedEventAsync(workerEvent, exception, cancellationToken) + .ConfigureAwait(false); + } } } } + /// + /// Ends the session on an event that cannot be framed, but deliberately and diagnosably + /// (IPC-30). An event above the negotiated frame maximum is undeliverable end to end — the + /// pipe maximum sits only an envelope reserve above the public gRPC cap — so dropping it + /// would silently make the event stream unfaithful, and synthesizing a placeholder is barred + /// by the no-synthesized-events rule. Instead the worker records which event blocked (never + /// its value: the redaction rule), writes a structured fault the gateway and dashboard can + /// surface, and then exits as it did before. Remediation is configuration: + /// MxGateway:Worker:MaxMessageBytes. Other per-frame rejection codes keep the previous + /// behavior — they indicate worker bugs, not workload size. + /// + private async Task FaultOnOversizedEventAsync( + WorkerEvent workerEvent, + WorkerFrameProtocolException exception, + CancellationToken cancellationToken) + { + MxEvent? mxEvent = workerEvent.Event; + string family = (mxEvent?.Family ?? MxEventFamily.Unspecified).ToString(); + ulong workerSequence = mxEvent?.WorkerSequence ?? 0; + int serverHandle = mxEvent?.ServerHandle ?? 0; + int itemHandle = mxEvent?.ItemHandle ?? 0; + + _logger?.Error( + "WorkerEventFrameTooLarge", + new Dictionary + { + ["session_id"] = _options.SessionId, + ["event_family"] = family, + ["worker_sequence"] = workerSequence, + ["server_handle"] = serverHandle, + ["item_handle"] = itemHandle, + ["max_message_bytes"] = _options.MaxMessageBytes, + // Sizes only — the event value never reaches the log. + ["reason"] = exception.Message, + }); + + string diagnosticMessage = + $"{family} event for server handle {serverHandle}, item handle {itemHandle} " + + $"(worker sequence {workerSequence}) exceeds the negotiated frame maximum of " + + $"{_options.MaxMessageBytes} bytes and cannot be delivered; raise " + + "MxGateway:Worker:MaxMessageBytes for this workload."; + + _state = WorkerState.Faulted; + await TryWriteFaultAsync( + new WorkerFault + { + Category = WorkerFaultCategory.ProtocolViolation, + CommandMethod = "EventDrain", + ExceptionType = exception.GetType().FullName ?? string.Empty, + DiagnosticMessage = diagnosticMessage, + ProtocolStatus = new ProtocolStatus + { + Code = ProtocolStatusCode.ProtocolViolation, + Message = diagnosticMessage, + }, + }, + cancellationToken).ConfigureAwait(false); + + throw new InvalidOperationException(diagnosticMessage, exception); + } + private async Task DispatchGatewayEnvelopeAsync( WorkerEnvelope envelope, CancellationToken cancellationToken) @@ -478,7 +548,8 @@ public sealed class WorkerPipeSession _ => CreateControlOkReply(correlationId, command.Kind), }; - await WriteControlReplyAsync(reply, cancellationToken).ConfigureAwait(false); + await WriteControlReplyWithSizeBackstopAsync(reply, correlationId, command.Kind, cancellationToken) + .ConfigureAwait(false); return true; } @@ -495,6 +566,70 @@ public sealed class WorkerPipeSession cancellationToken); } + /// + /// Writes a control reply, answering the correlation with a small error reply instead of + /// unwinding the session if the reply does not fit the negotiated frame maximum. Reply + /// builders already size their payloads (see ), so this + /// is a backstop against a future command or a sizing bug — but without it a single + /// oversized diagnostic reply is session-fatal, which no diagnostics command may be. + /// + private async Task WriteControlReplyWithSizeBackstopAsync( + MxCommandReply reply, + string correlationId, + MxCommandKind kind, + CancellationToken cancellationToken) + { + try + { + await WriteControlReplyAsync(reply, cancellationToken).ConfigureAwait(false); + } + catch (WorkerFrameProtocolException exception) + when (exception.ErrorCode == WorkerFrameProtocolErrorCode.MessageTooLarge) + { + LogControlReplyTooLarge(correlationId, kind, exception); + await WriteControlReplyAsync( + CreateReplyTooLargeReply(correlationId, kind), + cancellationToken).ConfigureAwait(false); + } + } + + private void LogControlReplyTooLarge( + string correlationId, + MxCommandKind kind, + WorkerFrameProtocolException exception) + { + _logger?.Error( + "WorkerControlReplyTooLarge", + new Dictionary + { + ["correlation_id"] = correlationId, + ["command_kind"] = kind.ToString(), + ["max_message_bytes"] = _options.MaxMessageBytes, + // The writer's message carries the rejected payload length; it names sizes only, + // never reply content. + ["reason"] = exception.Message, + }); + } + + private MxCommandReply CreateReplyTooLargeReply(string correlationId, MxCommandKind kind) + { + const string message = + "Worker reply exceeded the negotiated frame maximum; retry with a smaller request."; + return new MxCommandReply + { + SessionId = _options.SessionId, + CorrelationId = correlationId, + Kind = kind, + Hresult = 0, + DiagnosticMessage = message, + ProtocolStatus = new ProtocolStatus + { + Code = ProtocolStatusCode.InvalidRequest, + Message = message, + }, + }; + } + private MxCommandReply CreatePingReply(string correlationId, MxCommand command) { MxCommandReply reply = CreateControlOkReply(correlationId, command.Kind); @@ -543,24 +678,71 @@ public sealed class WorkerPipeSession if (runtimeSession is not null) { // Bound the diagnostic drain so max_events = 0 ("as many as available") or an over-large - // request cannot pack the whole queue into one session-killing reply frame. + // request cannot pack the whole queue into one session-killing reply frame. The count cap + // alone is not enough: byte-heavy events overshoot the negotiated frame maximum long + // before the count ceiling, so the drain is also byte-budgeted and sizes the reply while + // draining — an event that does not fit is left queued rather than dequeued and lost. uint requested = command.DrainEvents?.MaxEvents ?? 0; - uint maxEvents = requested == 0 || requested > MaxDrainEventsPerReply - ? MaxDrainEventsPerReply + uint maxEvents = requested == 0 || requested > GatewayContractInfo.MaxDrainEventsPerCommand + ? GatewayContractInfo.MaxDrainEventsPerCommand : requested; - foreach (WorkerEvent workerEvent in runtimeSession.DrainEvents(maxEvents)) + WorkerEventDrainResult drainResult = runtimeSession.DrainEvents( + maxEvents, + ResolveDrainReplyByteBudget()); + foreach (WorkerEvent workerEvent in drainResult.Events) { if (workerEvent.Event is not null) { drainReply.Events.Add(workerEvent.Event); } } + + if (drainResult.TruncatedBySize) + { + // DrainEventsReply has no truncation field, and adding one would regenerate every + // language client for a diagnostic nicety. The reply's existing DiagnosticMessage + // carries the same information at zero contract cost; the caller contract is to + // repeat DrainEvents until it comes back empty. + reply.DiagnosticMessage = CreateDrainTruncationMessage( + drainReply.Events.Count, + drainResult); + } } reply.DrainEvents = drainReply; return reply; } + /// + /// 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. + /// + private int ResolveDrainReplyByteBudget() + { + int budget = _options.MaxMessageBytes - DrainReplyFrameHeadroomBytes; + return budget > 0 ? budget : _options.MaxMessageBytes / 2; + } + + private static string CreateDrainTruncationMessage( + int returnedCount, + WorkerEventDrainResult drainResult) + { + string message = + $"{returnedCount} events returned, {drainResult.RemainingCount} remain; " + + "repeat DrainEvents for the rest."; + if (drainResult.OversizedHeadSequence != 0) + { + message += + $" The next event (worker sequence {drainResult.OversizedHeadSequence}) alone exceeds " + + "the negotiated frame maximum and cannot be drained; raise MxGateway:Worker:MaxMessageBytes."; + } + + return message; + } + private MxCommandReply CreateControlOkReply(string correlationId, MxCommandKind kind) { return new MxCommandReply @@ -627,15 +809,29 @@ public sealed class WorkerPipeSession return; } - await _writer - .WriteAsync( - CreateEnvelope(new WorkerCommandReply - { - Reply = reply, - CompletedTimestamp = Timestamp.FromDateTime(DateTime.UtcNow), - }), - cancellationToken) - .ConfigureAwait(false); + try + { + await _writer + .WriteAsync( + CreateEnvelope(new WorkerCommandReply + { + Reply = reply, + CompletedTimestamp = Timestamp.FromDateTime(DateTime.UtcNow), + }), + cancellationToken) + .ConfigureAwait(false); + } + catch (WorkerFrameProtocolException sizeException) + when (sizeException.ErrorCode == WorkerFrameProtocolErrorCode.MessageTooLarge) + { + // 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. + LogControlReplyTooLarge(envelope.CorrelationId, command.Kind, sizeException); + await WriteControlReplyAsync( + CreateReplyTooLargeReply(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/IWorkerRuntimeSession.cs b/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/IWorkerRuntimeSession.cs index e3b68e8..bf1f252 100644 --- a/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/IWorkerRuntimeSession.cs +++ b/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/IWorkerRuntimeSession.cs @@ -44,6 +44,19 @@ public interface IWorkerRuntimeSession : IDisposable /// List of drained events. IReadOnlyList DrainEvents(uint maxEvents); + /// + /// Drains pending events bounded by both a count cap and a byte budget, so a caller building a + /// single reply frame never removes an event it cannot ship. + /// + /// + /// Declared as a second method rather than a default interface method: the worker targets + /// .NET Framework 4.8, which has no runtime support for default interface members. + /// + /// Maximum number of events to drain; 0 means "no count limit". + /// Byte budget for the drained events' estimated serialized size. + /// The drained events and the truncation facts describing what stayed queued. + WorkerEventDrainResult DrainEvents(uint maxEvents, int maxTotalBytes); + /// /// Drains a pending fault from the queue, if any. /// diff --git a/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/MxAccessEventQueue.cs b/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/MxAccessEventQueue.cs index fe09401..e8d65f4 100644 --- a/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/MxAccessEventQueue.cs +++ b/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/MxAccessEventQueue.cs @@ -26,6 +26,11 @@ 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. + private const int RepeatedFieldOverheadBytes = 8; + private readonly int capacity; private readonly Queue events; private readonly object syncRoot = new(); @@ -209,6 +214,63 @@ public sealed class MxAccessEventQueue } } + /// + /// Drains from the head while both the count cap and a byte budget allow it, so the caller can + /// build a reply frame that is guaranteed to fit the negotiated frame maximum. + /// + /// + /// 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. + /// + /// Maximum number of events to drain; 0 means "no count limit". + /// Byte budget for the drained events' estimated serialized size. + /// The drained events plus the truncation facts the caller reports to the gateway. + public WorkerEventDrainResult Drain(uint maxEvents, int maxTotalBytes) + { + lock (syncRoot) + { + int countLimit = maxEvents == 0 + ? int.MaxValue + : checked((int)Math.Min(maxEvents, int.MaxValue)); + List drained = new(); + int remainingBudget = maxTotalBytes; + bool truncatedBySize = false; + ulong oversizedHeadSequence = 0; + + while (drained.Count < countLimit && events.Count > 0) + { + WorkerEvent head = events.Peek(); + int cost = head.CalculateSize() + RepeatedFieldOverheadBytes; + if (cost > remainingBudget) + { + truncatedBySize = true; + if (cost > maxTotalBytes) + { + // The head alone cannot fit this budget, so repeating the call will not + // move it either. Report its sequence instead of silently stalling; the + // events that did fit are still returned. + oversizedHeadSequence = head.Event?.WorkerSequence ?? 0; + } + + break; + } + + remainingBudget -= cost; + drained.Add(events.Dequeue()); + } + + return new WorkerEventDrainResult( + drained, + truncatedBySize, + events.Count, + oversizedHeadSequence); + } + } + /// /// Records a fault if one has not already been recorded. /// diff --git a/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/MxAccessStaSession.cs b/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/MxAccessStaSession.cs index 5353bc3..44d2b80 100644 --- a/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/MxAccessStaSession.cs +++ b/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/MxAccessStaSession.cs @@ -392,6 +392,12 @@ public sealed class MxAccessStaSession : IWorkerRuntimeSession return eventQueue.Drain(maxEvents); } + /// + public WorkerEventDrainResult DrainEvents(uint maxEvents, int maxTotalBytes) + { + return eventQueue.Drain(maxEvents, maxTotalBytes); + } + /// public WorkerFault? DrainFault() { diff --git a/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/WorkerEventDrainResult.cs b/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/WorkerEventDrainResult.cs new file mode 100644 index 0000000..4639469 --- /dev/null +++ b/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/WorkerEventDrainResult.cs @@ -0,0 +1,56 @@ +using System.Collections.Generic; +using ZB.MOM.WW.MxGateway.Contracts.Proto; + +namespace ZB.MOM.WW.MxGateway.Worker.MxAccess; + +/// +/// Outcome of a byte-budgeted drain from the MXAccess outbound event queue. +/// +/// +/// A count cap alone cannot keep a DrainEvents reply inside the negotiated frame +/// maximum: byte-heavy events (large string or array MxValues) overshoot the frame max +/// long before the count ceiling is reached, and the writer's per-frame rejection then +/// destroys events that were already removed from the queue. The byte-budgeted drain sizes +/// the reply while draining, so an event that does not fit is never dequeued (WRK-21), and +/// this result carries the truncation facts the reply's DiagnosticMessage reports — +/// no contract change is needed to express them. +/// Plain constructor and get-only properties: the worker targets .NET Framework 4.8, which +/// has no init-only members or positional records. +/// +public sealed class WorkerEventDrainResult +{ + /// Initializes a new instance of the class. + /// Events removed from the queue, in enqueue order. + /// Whether the byte budget, not the count cap, ended the drain. + /// Number of events still queued after the drain. + /// + /// Worker sequence of a head event whose own serialized size exceeds the whole budget, so + /// no future call of the same budget can ship it; 0 when there is no such event. + /// + public WorkerEventDrainResult( + IReadOnlyList events, + bool truncatedBySize, + int remainingCount, + ulong oversizedHeadSequence) + { + Events = events; + TruncatedBySize = truncatedBySize; + RemainingCount = remainingCount; + OversizedHeadSequence = oversizedHeadSequence; + } + + /// Gets the events removed from the queue, in enqueue order. + public IReadOnlyList Events { get; } + + /// Gets a value indicating whether the byte budget ended the drain early. + public bool TruncatedBySize { get; } + + /// Gets the number of events still queued after the drain. + public int RemainingCount { get; } + + /// + /// Gets the worker sequence of the head event that alone exceeds the byte budget, or 0 when + /// no single event blocks the drain. Naming it lets an operator find the offending tag. + /// + public ulong OversizedHeadSequence { get; } +}