From 33ba612ddd2238f9cad143c2f3671e473be2ab93 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Fri, 7 Aug 2026 05:38:23 -0400 Subject: [PATCH 1/6] fix(WRK-21,WRK-28,WRK-23,IPC-30): byte-budget the DrainEvents reply, stop size errors from killing sessions MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit WRK-21 — DrainEvents was bounded by event count only, so a byte-heavy queue (large string/array MxValues) built a reply above the negotiated frame maximum: the writer rejected the frame, the exception unwound the session, and the events already dequeued were destroyed. The drain is now byte-budgeted inside the queue lock, so an event is dequeued only once it is known to fit and one that does not stays at the head. Truncation is reported through the reply's existing DiagnosticMessage (no contract change); callers drain until an empty reply. Both reply-write seams — the control-command path and ProcessCommandAsync — now catch MessageTooLarge and answer the correlation with an InvalidRequest reply instead of unwinding or faulting the session. Satisfies IPC-23 R1-R3. WRK-28 — the 10,000 drain ceiling moves to GatewayContractInfo .MaxDrainEventsPerCommand, referenced by both the gateway request validator and the worker clamp, replacing a comment-only sync contract. C# const only; no .proto change. WRK-23 — WorkerFrameWriter now peek-stamps, validates, then commits the sequence counter immediately before the stream write, so a per-frame rejection leaves no phantom gap on the wire. IPC-30 — an oversized event frame stays session-fatal (it is undeliverable end to end and neither dropping nor synthesizing a replacement is allowed), but the death is structured: the event's identity and sizes are logged (never its value), a WorkerFault with category PROTOCOL_VIOLATION and command method EventDrain is written, then the session exits as before. Docs updated in the same change: MxAccessWorkerInstanceDesign.md (drain byte cap, truncation contract, oversized-head behavior, oversized-event policy, no control reply is session-fatal on size), WorkerFrameProtocol.md (reply pre-sizing, non-fatal reply-size rule, oversized-event policy, rejected frames do not consume sequence numbers), gateway.md (DrainEvents two-axis bound). --- .../2026-07-12/remediation/00-tracking.md | 16 +- .../2026-07-12/remediation/20-worker.md | 6 +- .../remediation/30-contracts-ipc.md | 4 +- docs/MxAccessWorkerInstanceDesign.md | 45 +++ docs/WorkerFrameProtocol.md | 23 ++ gateway.md | 12 +- .../GatewayContractInfo.cs | 15 + .../Grpc/MxAccessGrpcRequestValidator.cs | 20 +- .../Ipc/WorkerFrameProtocolTests.cs | 42 ++ .../Ipc/WorkerPipeSessionTests.cs | 359 +++++++++++++++++- .../MxAccess/MxAccessEventQueueTests.cs | 131 +++++++ .../TestSupport/FakeRuntimeSession.cs | 107 +++++- .../Ipc/WorkerFrameWriter.cs | 16 +- .../Ipc/WorkerPipeSession.cs | 242 ++++++++++-- .../MxAccess/IWorkerRuntimeSession.cs | 13 + .../MxAccess/MxAccessEventQueue.cs | 62 +++ .../MxAccess/MxAccessStaSession.cs | 6 + .../MxAccess/WorkerEventDrainResult.cs | 56 +++ 18 files changed, 1118 insertions(+), 57 deletions(-) create mode 100644 src/ZB.MOM.WW.MxGateway.Worker/MxAccess/WorkerEventDrainResult.cs 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; } +} From 7c2eaf09e2d5117b94089b8d92314b1e0f9bbad2 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Fri, 7 Aug 2026 06:25:15 -0400 Subject: [PATCH 2/6] test(WRK-21): size the byte-heavy drain fixture for the pipe harness MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit PipePair has no continuous read pump — the test thread drains the pipe only while it sits in ReadUntilAsync — so multi-megabyte DrainEvents frames interleaved with the heartbeat loop wedge both ends inside FlushFileBuffers, each waiting for the other to read. Negotiate a 128 KiB frame maximum instead: the 10,000 byte-heavy events still overflow it many times over, so every assertion (bounded reply, reported truncation, no event loss across repeated drains, surviving session) is unchanged. --- .../Ipc/WorkerPipeSessionTests.cs | 15 ++++++++++----- 1 file changed, 10 insertions(+), 5 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 65e8468..56fe252 100644 --- a/src/ZB.MOM.WW.MxGateway.Worker.Tests/Ipc/WorkerPipeSessionTests.cs +++ b/src/ZB.MOM.WW.MxGateway.Worker.Tests/Ipc/WorkerPipeSessionTests.cs @@ -20,12 +20,17 @@ public sealed class WorkerPipeSessionTests private const string Nonce = "nonce-secret"; // Byte-heavy drain fixture (WRK-21). 10,000 events at ~1.7 KiB each is ~17 MB of queue — far - // more than one frame — so DrainEvents must split across replies. 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. + // more than one frame — so DrainEvents must split across replies. + // + // The negotiated frame maximum is deliberately small. What is under test is the byte cap, and + // it behaves identically at any frame size, but this harness is not the production gateway: + // PipePair has no continuous read pump, so the test thread only drains the pipe while it sits + // in ReadUntilAsync. Multi-megabyte frames interleaved with the heartbeat loop can therefore + // wedge both ends inside FlushFileBuffers, each waiting for the other to read. A frame maximum + // well under the pipe buffer keeps the harness honest without weakening a single assertion. private const int ByteHeavyEventCount = 10_000; private const int ByteHeavyEventPayloadBytes = 1_800; - private const uint NegotiatedMaxFrameBytes = 2 * 1024 * 1024; + private const uint NegotiatedMaxFrameBytes = 128 * 1024; /// Verifies that valid gateway hello triggers worker hello and ready responses. /// A task that represents the asynchronous operation. @@ -611,7 +616,7 @@ public sealed class WorkerPipeSessionTests recovered.Add(drained.WorkerSequence); } - Assert.True(replyCount < 100, "DrainEvents made no progress across 100 replies."); + Assert.True(replyCount < 1_000, "DrainEvents made no progress across 1,000 replies."); } // More than one reply proves the drain really split; every event came back exactly once, in From a2565604df228e039830ac84b9c3941225200c45 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Fri, 7 Aug 2026 06:50:17 -0400 Subject: [PATCH 3/6] test(WRK-21): keep the drain-to-empty walk inside the pipe harness envelope PipePair runs both ends of a duplex pipe in one process with blocking FlushFileBuffers under every frame write, so it wedges after roughly 85 large round trips. Drain the full 10,000 byte-heavy events to empty at the queue layer, where the no-loss property actually lives, and keep the pipe walk at 1,000 events (29 replies) so it still proves the split end to end. Also give the truncation test's budget slack: item handle 0 is a proto3 default and is not serialized, so the probe measurement is a lower bound on the fixture's per-event cost. --- .../Ipc/WorkerPipeSessionTests.cs | 22 +++---- .../MxAccess/MxAccessEventQueueTests.cs | 60 ++++++++++++++++++- 2 files changed, 70 insertions(+), 12 deletions(-) diff --git a/src/ZB.MOM.WW.MxGateway.Worker.Tests/Ipc/WorkerPipeSessionTests.cs b/src/ZB.MOM.WW.MxGateway.Worker.Tests/Ipc/WorkerPipeSessionTests.cs index 56fe252..504cae8 100644 --- a/src/ZB.MOM.WW.MxGateway.Worker.Tests/Ipc/WorkerPipeSessionTests.cs +++ b/src/ZB.MOM.WW.MxGateway.Worker.Tests/Ipc/WorkerPipeSessionTests.cs @@ -20,15 +20,17 @@ public sealed class WorkerPipeSessionTests private const string Nonce = "nonce-secret"; // Byte-heavy drain fixture (WRK-21). 10,000 events at ~1.7 KiB each is ~17 MB of queue — far - // more than one frame — so DrainEvents must split across replies. + // more than one frame — so DrainEvents must truncate. // - // The negotiated frame maximum is deliberately small. What is under test is the byte cap, and - // it behaves identically at any frame size, but this harness is not the production gateway: - // PipePair has no continuous read pump, so the test thread only drains the pipe while it sits - // in ReadUntilAsync. Multi-megabyte frames interleaved with the heartbeat loop can therefore - // wedge both ends inside FlushFileBuffers, each waiting for the other to read. A frame maximum - // well under the pipe buffer keeps the harness honest without weakening a single assertion. + // Two limits below are harness accommodations, not properties of the fix. PipePair runs both + // ends of a duplex pipe inside one process, with no continuous read pump and with blocking + // FlushFileBuffers under every frame write, so it tolerates neither multi-megabyte frames nor + // hundreds of large round trips before both ends wedge waiting on each other. Hence a small + // negotiated frame maximum, and a smaller queue for the drain-to-empty walk. The byte cap + // behaves identically at any frame size; exhaustive no-loss over the full 10,000 events is + // covered without a pipe by MxAccessEventQueueTests. private const int ByteHeavyEventCount = 10_000; + private const int RepeatedDrainEventCount = 1_000; private const int ByteHeavyEventPayloadBytes = 1_800; private const uint NegotiatedMaxFrameBytes = 128 * 1024; @@ -576,7 +578,7 @@ public sealed class WorkerPipeSessionTests FakeRuntimeSession runtime = new() { SuppressDrainForBatchSize = 128, - BackingQueue = CreateByteHeavyQueue(ByteHeavyEventCount, ByteHeavyEventPayloadBytes), + BackingQueue = CreateByteHeavyQueue(RepeatedDrainEventCount, ByteHeavyEventPayloadBytes), }; WorkerPipeSession session = CreatePipeSession(pipePair.WorkerStream, runtime); Task runTask = session.RunAsync(cancellation.Token); @@ -616,13 +618,13 @@ public sealed class WorkerPipeSessionTests recovered.Add(drained.WorkerSequence); } - Assert.True(replyCount < 1_000, "DrainEvents made no progress across 1,000 replies."); + Assert.True(replyCount < 200, "DrainEvents made no progress across 200 replies."); } // More than one reply proves the drain really split; every event came back exactly once, in // enqueue order. Assert.True(replyCount > 2, $"Expected the byte cap to split the drain, saw {replyCount} replies."); - Assert.Equal(ByteHeavyEventCount, recovered.Count); + Assert.Equal(RepeatedDrainEventCount, recovered.Count); for (int index = 0; index < recovered.Count; index++) { Assert.Equal((ulong)(index + 1), recovered[index]); diff --git a/src/ZB.MOM.WW.MxGateway.Worker.Tests/MxAccess/MxAccessEventQueueTests.cs b/src/ZB.MOM.WW.MxGateway.Worker.Tests/MxAccess/MxAccessEventQueueTests.cs index a405e7f..5c3d639 100644 --- a/src/ZB.MOM.WW.MxGateway.Worker.Tests/MxAccess/MxAccessEventQueueTests.cs +++ b/src/ZB.MOM.WW.MxGateway.Worker.Tests/MxAccess/MxAccessEventQueueTests.cs @@ -142,9 +142,12 @@ public sealed class MxAccessEventQueueTests queue.Enqueue(CreateEventWithPayload(itemHandle, payloadLength: 256)); } + // One-and-a-half events' worth of budget: the head fits, the next does not, and the next is + // comfortably smaller than the whole budget so it is a plain truncation rather than the + // oversized-head case. WorkerEventDrainResult result = queue.Drain( maxEvents: 0, - maxTotalBytes: MeasureDrainCost(payloadLength: 256)); + maxTotalBytes: MeasureDrainCost(payloadLength: 256) * 3 / 2); Assert.Single(result.Events); Assert.True(result.TruncatedBySize); @@ -175,6 +178,57 @@ public sealed class MxAccessEventQueueTests Assert.Equal(2, queue.Count); } + /// + /// The no-loss half of the WRK-21 acceptance criterion, at full scale. Draining the review's + /// 10,000 byte-heavy events under a budget that fits only a fraction of them per call must + /// return every event exactly once and in order: the pre-fix drain removed events from the + /// queue before the reply was sized, so a rejected frame destroyed them. This runs at the + /// queue layer because the property is the queue's, and because the pipe harness that covers + /// the same walk end to end cannot sustain hundreds of large round trips. + /// + [Fact] + public void Drain_ByteBudget_RepeatedCalls_RecoverAllEventsInOrderWithoutLoss() + { + const int eventCount = 10_000; + const int payloadLength = 1_800; + MxAccessEventQueue queue = new(eventCount); + for (int index = 0; index < eventCount; index++) + { + queue.Enqueue(CreateEventWithPayload(index, payloadLength)); + } + + // A budget that fits roughly 35 events, so the walk takes hundreds of calls. + int budget = MeasureDrainCost(payloadLength) * 35; + List recovered = new(); + int calls = 0; + while (true) + { + WorkerEventDrainResult result = queue.Drain(maxEvents: 0, maxTotalBytes: budget); + calls++; + if (result.Events.Count == 0) + { + break; + } + + foreach (WorkerEvent drained in result.Events) + { + recovered.Add(drained.Event.WorkerSequence); + } + + Assert.Equal(eventCount - recovered.Count, result.RemainingCount); + Assert.True(calls < eventCount, "Drain made no progress."); + } + + Assert.True(calls > 100, $"Expected the byte budget to split the drain, saw {calls} calls."); + Assert.Equal(eventCount, recovered.Count); + for (int index = 0; index < recovered.Count; index++) + { + Assert.Equal((ulong)(index + 1), recovered[index]); + } + + Assert.Equal(0, queue.Count); + } + /// /// Verifies the count cap still binds when the byte budget is generous: the byte cap is an /// additional bound, not a replacement. @@ -254,7 +308,9 @@ public sealed class MxAccessEventQueueTests /// /// Measures what the queue charges one event of the given payload size against the byte budget: /// the serialized as it exists after Enqueue (sequence and timestamp - /// stamped) plus the repeated-field allowance. + /// stamped) plus the repeated-field allowance. The probe uses item handle 0, a proto3 default + /// that is not serialized, so this is a lower bound on the fixtures' real per-event cost — the + /// budgets above carry slack rather than assuming byte equality. /// /// Length of the event's raw-status payload string. /// The per-event byte cost. From c9925688f539924d12f2117d9837af2d64a81a91 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Fri, 7 Aug 2026 06:54:24 -0400 Subject: [PATCH 4/6] docs(tracking): record the WRK-21 cluster as Done with windev evidence MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Change-log row for 2026-08-07: what landed for WRK-21/WRK-28/WRK-23/IPC-30, why IPC-23 stays In progress (proto-comment/doc wave pending), and the verification evidence — macOS NonWindows build + validator tests, and the documented windev path (scripts/ci/windev-worker-ci.ps1 -Mode test) at a256560: x86 Worker build clean, Worker.Tests 367 passed / 0 failed / 11 skipped. --- archreview/2026-07-12/remediation/00-tracking.md | 1 + .../MxAccess/MxAccessEventQueueTests.cs | 6 +++--- 2 files changed, 4 insertions(+), 3 deletions(-) diff --git a/archreview/2026-07-12/remediation/00-tracking.md b/archreview/2026-07-12/remediation/00-tracking.md index ac9af59..255ca64 100644 --- a/archreview/2026-07-12/remediation/00-tracking.md +++ b/archreview/2026-07-12/remediation/00-tracking.md @@ -161,3 +161,4 @@ Sequence these together rather than piecemeal — several are one change set spa | 2026-07-13 | Operator bring-up complete: dedicated CI ed25519 key installed in windev `administrators_authorized_keys` (authorized into `dohertj2`, which owns the working MXAccess/toolchain env — a fresh OS account would break the build; the key is independently revocable), Gitea secrets `WINDEV_SSH_KEY`/`WINDEV_SSH_KNOWN_HOSTS` + variable `WINDEV_SSH_USER=dohertj2` stored, runner→`10.100.0.48:22` egress verified on the `traefik` net, issue-write confirmed. **TST-25/TST-26 → `Done`:** credentialed `windows-x86` ran GREEN on `d769244` (Gitea run #37) — Linux runner SSHed windev, checked out the SHA in `C:\build\mxaccessgw-ci` under lock, ran the x86 Worker build + `Worker.Tests`, exit 0; `nightly-windev` correctly skipped on the push event. Branch merged to `main`. Follow-ups (old tracker): revisit **TST-05** (scheduled live smoke — now covered by `nightly-windev`) and **TST-24** (client wire tests) which this unlocks. | | 2026-07-13 | Ran the TST-25 acceptance checks (scripts/ci/README.md) — they caught **two real CI defects, both fixed** on `fix/tst-25-ci-key-log-leak`: (1) **CI SSH key leaked in cleartext** in the `windows-x86` step env echo (Gitea's line-oriented masker missed the multiline PEM) — rotated the CI key on windev (old pubkey revoked), stored the key **base64-encoded** so the masker redacts it to `***` (confirmed on run #38), taught `run-windev-ci.sh` to decode, dropped the redundant public known-hosts secret from the job env; (2) **bootstrap lock race** — `run-windev-ci.sh`'s pre-hand-off `git fetch`/`checkout` ran outside the worktree lock, so concurrent runs collided on `.git/index.lock`; the bootstrap now holds the lock (ps1 re-uses it via `MXGW_CI_LOCK_HELD`), retest confirmed clean serialization. Also **deflaked** `SessionManagerTests` fail-fast timing assertions (absolute `<100ms` wall-clock bound flaked under CI load; now anchored to the configured timeout / dropped for the zero-timeout case). Checks passed: unreachable-host fast-fail (exit 255/15s), deliberate-red propagation (Worker.Tests failure → exit 1), lock concurrency (2nd run waits), no-key-in-logs (masked). Merge target `df7e20d` verified GREEN via the local windev path (Worker build + 356 tests); merged to `main` `19cbf7b`. Check 6 (forced-failure nightly issue): issue endpoint+token proven live at bring-up (#124); in-CI forced-failure probe abandoned to shared-runner congestion (residual `if: failure()` gating is standard Actions). | | 2026-07-13 | New finding **TST-30** (`Low`/`P2`) added — surfaced during TST-25 acceptance: CI runs on a single shared `gitea-runner` (`maxParallel=1`, co-located `10.100.0.35`) interleaved with `dohertj2/lmxopcua`, and Gitea 1.26 exposes no run cancel/delete, so queue latency is unbounded under cross-repo contention and the runner is a single point of failure. Design: add a second/labelled runner + document the no-cancel reality and the `run-windev-ci.sh` bypass. Roll-ups updated (Testing Low 2→3, total 47→48; P2 9→10). | +| 2026-08-07 | **WRK-21 + WRK-28 + WRK-23 + IPC-30 → `Done`** (branch `fix/wrk-21-drain-cluster`, commits `33ba612` + test-fixture follow-ups `7c2eaf0`/`a256560`). WRK-21: `MxAccessEventQueue` gains a byte-budgeted `Drain(maxEvents, maxTotalBytes)` returning the new `WorkerEventDrainResult`, sizing inside the queue lock so an event that will not fit is never dequeued; `CreateDrainEventsReply` budgets against the negotiated frame max less a 64 KiB wrapper reserve and reports truncation through the existing `DiagnosticMessage` (no proto change), satisfying IPC-23 R1–R3; both reply-write seams (`HandleControlCommandAsync`, `ProcessCommandAsync`) now catch `MessageTooLarge` and answer the correlation with an `InvalidRequest` reply instead of unwinding/faulting the session. WRK-28: the 10,000 ceiling moved to `GatewayContractInfo.MaxDrainEventsPerCommand`, referenced by the gateway validator and the worker clamp (C# const, no `.proto` change). WRK-23: `WorkerFrameWriter` peek-stamps then commits `Sequence` only immediately before the stream write, so rejections leave no wire gap. IPC-30: an oversized event frame stays session-fatal but writes a `PROTOCOL_VIOLATION` `WorkerFault` with `command_method = EventDrain` naming family/handles/sequence/sizes (never the value) before exiting. Docs same commit: `MxAccessWorkerInstanceDesign.md`, `WorkerFrameProtocol.md`, `gateway.md`. **IPC-23 → `In progress`** — mechanics landed here; the proto-comment/doc wave (and its regen fan-out) is still pending and must not be folded into this branch. **Evidence** — macOS: `dotnet build src/ZB.MOM.WW.MxGateway.NonWindows.slnx` 0 warnings/0 errors, `dotnet test …MxGateway.Tests --filter FullyQualifiedName~MxAccessGrpcRequestValidator` 4/4 passed. windev (`scripts/ci/windev-worker-ci.ps1 -Sha a2565604 -Mode test`, 2026-08-07 06:47): x86 Worker build 0 warnings/0 errors, `Worker.Tests` **367 passed / 0 failed / 11 skipped** (skips are the live-MXAccess/dev-rig opt-ins), script exit 0. **Harness note:** `PipePair` runs both pipe ends in one process with blocking `FlushFileBuffers` per frame, so it wedges on multi-MB frames or after ~85 large round trips; the pipe tests therefore negotiate a 128 KiB frame maximum and walk 1,000 events to empty, while the full 10,000-event drain-to-empty no-loss proof runs at the queue layer (`MxAccessEventQueueTests`). | 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 5c3d639..1492ec0 100644 --- a/src/ZB.MOM.WW.MxGateway.Worker.Tests/MxAccess/MxAccessEventQueueTests.cs +++ b/src/ZB.MOM.WW.MxGateway.Worker.Tests/MxAccess/MxAccessEventQueueTests.cs @@ -300,9 +300,9 @@ 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. + // Mirrors MxAccessEventQueue's per-event repeated-field allowance. Kept local rather than made + // public on the queue: the byte-budget tests state their budgets in units of that charge, so a + // change to it should surface here as a failing bound instead of silently moving with the code. private const int RepeatedFieldOverheadBytes = 8; /// From 6bc3f9b9918cdaa7bedad2e32d0367bbde3b52cf Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Fri, 7 Aug 2026 07:09:13 -0400 Subject: [PATCH 5/6] 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. From 758277bc62f729cdba6b3dd764af19b80c6f4d23 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Fri, 7 Aug 2026 07:13:46 -0400 Subject: [PATCH 6/6] docs(tracking): record WRK-21 review follow-ups (monotonic budget, guarded fallback) with windev evidence --- archreview/2026-07-12/remediation/00-tracking.md | 1 + 1 file changed, 1 insertion(+) diff --git a/archreview/2026-07-12/remediation/00-tracking.md b/archreview/2026-07-12/remediation/00-tracking.md index 255ca64..c4828cd 100644 --- a/archreview/2026-07-12/remediation/00-tracking.md +++ b/archreview/2026-07-12/remediation/00-tracking.md @@ -162,3 +162,4 @@ Sequence these together rather than piecemeal — several are one change set spa | 2026-07-13 | Ran the TST-25 acceptance checks (scripts/ci/README.md) — they caught **two real CI defects, both fixed** on `fix/tst-25-ci-key-log-leak`: (1) **CI SSH key leaked in cleartext** in the `windows-x86` step env echo (Gitea's line-oriented masker missed the multiline PEM) — rotated the CI key on windev (old pubkey revoked), stored the key **base64-encoded** so the masker redacts it to `***` (confirmed on run #38), taught `run-windev-ci.sh` to decode, dropped the redundant public known-hosts secret from the job env; (2) **bootstrap lock race** — `run-windev-ci.sh`'s pre-hand-off `git fetch`/`checkout` ran outside the worktree lock, so concurrent runs collided on `.git/index.lock`; the bootstrap now holds the lock (ps1 re-uses it via `MXGW_CI_LOCK_HELD`), retest confirmed clean serialization. Also **deflaked** `SessionManagerTests` fail-fast timing assertions (absolute `<100ms` wall-clock bound flaked under CI load; now anchored to the configured timeout / dropped for the zero-timeout case). Checks passed: unreachable-host fast-fail (exit 255/15s), deliberate-red propagation (Worker.Tests failure → exit 1), lock concurrency (2nd run waits), no-key-in-logs (masked). Merge target `df7e20d` verified GREEN via the local windev path (Worker build + 356 tests); merged to `main` `19cbf7b`. Check 6 (forced-failure nightly issue): issue endpoint+token proven live at bring-up (#124); in-CI forced-failure probe abandoned to shared-runner congestion (residual `if: failure()` gating is standard Actions). | | 2026-07-13 | New finding **TST-30** (`Low`/`P2`) added — surfaced during TST-25 acceptance: CI runs on a single shared `gitea-runner` (`maxParallel=1`, co-located `10.100.0.35`) interleaved with `dohertj2/lmxopcua`, and Gitea 1.26 exposes no run cancel/delete, so queue latency is unbounded under cross-repo contention and the runner is a single point of failure. Design: add a second/labelled runner + document the no-cancel reality and the `run-windev-ci.sh` bypass. Roll-ups updated (Testing Low 2→3, total 47→48; P2 9→10). | | 2026-08-07 | **WRK-21 + WRK-28 + WRK-23 + IPC-30 → `Done`** (branch `fix/wrk-21-drain-cluster`, commits `33ba612` + test-fixture follow-ups `7c2eaf0`/`a256560`). WRK-21: `MxAccessEventQueue` gains a byte-budgeted `Drain(maxEvents, maxTotalBytes)` returning the new `WorkerEventDrainResult`, sizing inside the queue lock so an event that will not fit is never dequeued; `CreateDrainEventsReply` budgets against the negotiated frame max less a 64 KiB wrapper reserve and reports truncation through the existing `DiagnosticMessage` (no proto change), satisfying IPC-23 R1–R3; both reply-write seams (`HandleControlCommandAsync`, `ProcessCommandAsync`) now catch `MessageTooLarge` and answer the correlation with an `InvalidRequest` reply instead of unwinding/faulting the session. WRK-28: the 10,000 ceiling moved to `GatewayContractInfo.MaxDrainEventsPerCommand`, referenced by the gateway validator and the worker clamp (C# const, no `.proto` change). WRK-23: `WorkerFrameWriter` peek-stamps then commits `Sequence` only immediately before the stream write, so rejections leave no wire gap. IPC-30: an oversized event frame stays session-fatal but writes a `PROTOCOL_VIOLATION` `WorkerFault` with `command_method = EventDrain` naming family/handles/sequence/sizes (never the value) before exiting. Docs same commit: `MxAccessWorkerInstanceDesign.md`, `WorkerFrameProtocol.md`, `gateway.md`. **IPC-23 → `In progress`** — mechanics landed here; the proto-comment/doc wave (and its regen fan-out) is still pending and must not be folded into this branch. **Evidence** — macOS: `dotnet build src/ZB.MOM.WW.MxGateway.NonWindows.slnx` 0 warnings/0 errors, `dotnet test …MxGateway.Tests --filter FullyQualifiedName~MxAccessGrpcRequestValidator` 4/4 passed. windev (`scripts/ci/windev-worker-ci.ps1 -Sha a2565604 -Mode test`, 2026-08-07 06:47): x86 Worker build 0 warnings/0 errors, `Worker.Tests` **367 passed / 0 failed / 11 skipped** (skips are the live-MXAccess/dev-rig opt-ins), script exit 0. **Harness note:** `PipePair` runs both pipe ends in one process with blocking `FlushFileBuffers` per frame, so it wedges on multi-MB frames or after ~85 large round trips; the pipe tests therefore negotiate a 128 KiB frame maximum and walk 1,000 events to empty, while the full 10,000-event drain-to-empty no-loss proof runs at the queue layer (`MxAccessEventQueueTests`). | +| 2026-08-07 | Code-review follow-ups on the same branch (commit `6bc3f9b`). (1) **Important** — `ResolveDrainReplyByteBudget` was a step, not a floor: just above the 64 KiB reserve the budget collapsed to a few bytes (exactly 1024 at the validator floor `MaxMessageBytes = 1024 + 64 KiB`), so a byte-heavy `DrainEvents` truncated on every call and the drain-until-empty loop never terminated. Now `Math.Max(frameMax - reserve, frameMax / 2)` — monotonic, never below half the frame max. New test `WorkerPipeSessionTests.DrainEvents_AtValidatorFloorFrameMax_MakesProgressAndTerminates` drives a byte-heavy queue at the exact validator floor and asserts drain-to-empty with no head reported oversized. (2) **Hardening** — the reply-too-large fallback write is now itself size-guarded (`WriteReplyTooLargeFallbackAsync`, shared by the control and STA reply seams) so a pathologically tiny negotiated max below the gateway floor (the WRK-24 gap) cannot make even the backstop session-fatal; log-and-swallow, comment points at WRK-24. (3) **Comment** — corrected the `RepeatedFieldOverheadBytes` docs: `WorkerEvent.CalculateSize()` already includes the event's tag+length, so the 8 bytes is pure slack, not wrapper compensation. **Evidence** — macOS build 0/0, validator filter 4/4. windev (`windev-worker-ci.ps1 -Sha 6bc3f9b -Mode test`, 07:07): x86 Worker build 0/0, `Worker.Tests` **368 passed / 0 failed / 11 skipped**, script exit 0. (An earlier run of the same SHA flaked on the pre-existing `RunAsync_WhenStaActivityIsStale_WritesWatchdogFault` — a 5 s CTS timeout under first-run load, untouched by this change; it passed on the clean re-run and in both prior full runs.) |