diff --git a/archreview/2026-07-12/remediation/00-tracking.md b/archreview/2026-07-12/remediation/00-tracking.md index 11cca5a..a2619a4 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) | Done | Empty-ring ReplayGap sentinel carries `oldest_available_sequence = 0`, dead-streaming a compliant client | | CLI-35 | Medium | S | GWC-25 (coord) | Done | Python CLI `stream-events` crashes on a ReplayGap | | CLI-36 | Medium | S | GWC-25 (coord) | Done | 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 | — | Done | Failure limiter partitions on attacker-controlled key id and blocks before verification (lockout DoS) | | SEC-32 | Low | S | SEC-31 | Done | 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 | @@ -172,3 +172,5 @@ Sequence these together rather than piecemeal — several are one change set spa | 2026-08-07 | **SEC-31 + SEC-32 → `Done`** (branch `fix/sec-31-32-limiter`, one change set as planned). `ApiKeyFailureLimiter` reworked from `IsBlocked/RecordFailure/Reset(string peer)` to a partition-pair API (`Check/RecordFailure/Reset(ApiKeyThrottlePartition)` returning `ApiKeyThrottleDecision`): layer 1 is the composite `(transport peer, key id)` partition, layer 2 a per-key-id aggregate across peers (`ApiKeyFailureAggregateLimit`, default 30), and an over-limit state now admits one probe per `ApiKeyFailureProbeIntervalSeconds` (default 5) instead of blocking absolutely — so a success can reset the state while throttled, killing the 10-packets-per-minute lockout. SEC-32 rides along: the interceptor validates token shape (`mxgw` prefix, ≥3 non-empty `_` segments, key id ≤ 64 chars) before minting a key-id partition, each peer may mint at most 32 of them (overflow collapses to its fallback partition), and eviction prefers expired windows, never dropping an over-limit partition below a 2× transient overshoot ceiling. New counter `mxgateway.auth.throttled` tagged `stage=peer\|aggregate` only (no key material — `/metrics` is still unauthenticated per open SEC-14). Docs updated in the same commit (`docs/GatewayConfiguration.md` limiter rows + two new keys, `docs/Authentication.md` hot-path paragraph, `docs/Authorization.md` SEC-11 section, limiter/`SecurityOptions` XML remarks). Evidence: `dotnet build …Server` clean; `--filter ~ApiKeyFailureLimiter` 11/11 passed (new `ApiKeyFailureLimiterTests`), `--filter ~GatewayGrpcAuthorizationInterceptor` 20/20 passed (incl. the four SEC-31 contract tests and `NonMxgwToken_FallsBackToTransportPeerPartition`), `--filter ~GatewayOptionsValidator` 66/66 passed. Full suite on macOS: 804 passed / 44 failed — all 44 are the pre-existing named-pipe fake-worker classes (`WorkerClientTests`, `FakeWorkerHarnessTests`, `SessionWorkerClientFactoryFakeWorkerTests`, `GatewayEndToEnd*`), verified identical (44) on the unmodified tree. Follow-up unchanged: the new `MxGateway:Security` keys belong in old **SEC-24**'s effective-config projection when that is picked up. **Code review of the branch found two defects in the first pass, both fixed before merge:** (1) probe admission was check-then-act across two lock scopes, so a burst arriving at an interval boundary could all observe "due" and all be admitted — the claim is now a single critical section (`TryConsumeProbe`), and because the two layers are claimed one at a time, a slot claimed on the partition is compensated (`ReleaseProbe`) when the aggregate then refuses; (2) `Reset` on a success whose key id had been collapsed into the address's shared fallback partition removed that shared partition, letting one authentication wipe an in-progress spray from the same address — it is now left to decay by window expiry, while the key's aggregate is still cleared. Tests added: `ProbeAdmission_UnderConcurrentArrivals_GrantsExactlyOneSlot`, `ProbeAdmission_WhenAggregateRefuses_ReturnsTheClaimedPeerSlot`, `Reset_WithOverCapKeyId_DoesNotClearSharedFallbackPartition`. **A second review pass found a residual defect in that compensation path:** the release identified its own reservation by comparing `NextProbeAtTicks` to `now + interval`, the identical expression a failure re-arm writes — so a concurrent `RecordFailure` on the same state sharing a clock tick (routine at ~1 ms resolution) was mistaken for the caller's own claim and stomped back to the stale, already-due value, prematurely reopening the probe slot. Replaced with a monotonic per-state `ProbeVersion` bumped by every writer of `NextProbeAtTicks` (claim and re-arm alike); the release restores only when the version still matches the one its claim stamped, and bumps it again on restore so no other stale release can match. Covered by `ProbeSlotRestore_DoesNotStompConcurrentRearmAtSameTick`, made deterministic by a new `internal ProbeReleaseInterleaveHook` test seam (null in production, one null check on the refused path) because the claim-to-release window is nanoseconds wide and racing threads cannot hit it reliably — verified as a genuine red against the timestamp guard (`Expected: ThrottledByPeer / Actual: ProbeAdmitted`). Limiter suite 11 → 15. | | 2026-08-07 | **CLI-40 + CLI-41 + CLI-44 → `Done`** (branch `fix/cli-40-41-44`), one change set; two fast-follow riders from the CLI-37/38 review landed alongside. **CLI-40** (exact-secret credential scrub, ported to Rust/Java/.NET; Go/Python already did it): every credential-bearing helper (`AuthenticateUser` password, `WriteSecured`/`WriteSecured2` string payloads) now scrubs the **exact** caller-supplied secret from any surfaced error text, on top of the by-construction guarantee — Rust `MxAccessError` gained a `secrets: Vec` field whose `Display` scrubs exact-then-pattern (and a **hand-written redacting `Debug`**, since the derived `Debug` would have leaked the reply verbatim — caught by the existing Debug regression test); Java added `MxGatewaySecrets.redactExact` + a private `invokeCommandRedacted(command, secrets…)` that rebuilds the same exception type with the redacted message and **does not chain the secret-bearing original as cause**; .NET added an internal `MxGatewaySecretRedaction` (rebuilds the same concrete `MxGateway*Exception` type via a type switch) wired into the three credential helpers — and it carries the original's **inner** cause forward rather than the secret-bearing original, so `ToString()` (what loggers emit) is scrubbed too, not just `Message` (locked by a `ToString()` assertion). **CLI-41** (uniform malformed-reply contract for `AuthenticateUser`/`ArchestrAUserToId`/`AddBufferedItem` across all five): typed payload → present `return_value` with the int32 variant → else a typed malformed-reply error (`MalformedReplyError` Go/Python, `MxGatewayMalformedReplyException` Java/.NET, existing `Error::MalformedReply` Rust) — never a proto3 default `0`, never an NRE (fixes the Go/Java silent-`0`, .NET NRE, and Rust's own internal inconsistency by giving `authenticate_user_id`/`archestra_user_id` the same `return_value` fallback `add_buffered_item_handle` already had). **CLI-44** (Go): the event goroutine's Recv-error path now uses a new non-blocking `sendTerminalEventResult` on the reserved slot instead of `sendEventResult`, so a genuine terminal gRPC error is reported as itself even when the 16 data slots are full, rather than being mislabeled `ErrSlowConsumer`; test `TestEventsFullBufferTerminalErrorKeepsRootCause` was confirmed red-first (a 250 ms settle after `streamDone` is required to make the buffer genuinely full at error time). Three shared fixtures added under `clients/proto/fixtures/behavior/command-replies/` (`authenticate-user.{echoed-credential,missing-payload,return-value-only}.reply.json`; the echoed-credential reply uses an OK envelope + negative HRESULT + the credential in `protocolStatus.message` / `statuses[0].diagnosticText` / `diagnosticMessage` so all five clients route it to their MXAccess error uniformly) + manifest + `docs/ClientBehaviorFixtures.md` + `docs/ClientLibrariesDesign.md`. **Rider (a):** .NET `ToDiagnosticSummary` and Python `_mxaccess_message` now surface the raw `success` member (Rust already did), for diagnostics-only parity. **Rider (b):** the status-conversion fixture gained an independent `wantSuccess` boolean per case; the Go `TestStatusConversionFixtures` and .NET `FixtureStatuses_ProjectSuccessAndPreserveRawFields` now assert against it instead of recomputing `category == OK` (the formula under test). **Deviation:** the `` marker is not universal — Go/Rust/Java use ``, Python and the .NET CLI use `[redacted]`; each suite asserts its own client's marker plus the exact-secret absence (marker unification was out of scope). Verification (all five, on macOS): Go `gofmt -l` clean + `go build ./...` + `go test ./...` ok; Python `python -m pytest` 162 passed / 1 skipped; .NET `dotnet build …Client.slnx` 0 warnings + client tests 120 passed / 1 skipped; Rust `cargo fmt` + `cargo check --workspace` + `cargo test --workspace` (all targets pass) + `cargo clippy --all-targets -- -D warnings` clean; Java `JAVA_HOME=/opt/homebrew/opt/openjdk@17 gradle test` BUILD SUCCESSFUL with **no** generated-tree churn to revert (no `.proto` changed). **Code review of the branch found three defects, all fixed before merge:** (1, all five, critical) the message-only scrub left the server-echoed credential exposed on the redacted error's **structured reply accessor** (`.NET MxAccessException.Reply`/`Statuses`, Java `reply()`/`protocolStatus()`, Go `MxAccessError.Reply` via `errors.As`, Rust `reply()`/`into_reply()`, Python `raw_reply`) — the redacted error now carries a **scrubbed clone** of the reply (`protocol_status.message`, `diagnostic_message`, `statuses[].diagnostic_text` all redacted), with per-language tests asserting the reply accessor is clean; docs/ClientLibrariesDesign.md updated to make the "never reaches exception text" claim true for structured fields too. (2, Rust, critical) `ensure_command_success` routed `PROTOCOL_STATUS_CODE_MXACCESS_FAILURE` to `Error::Command` (unlike the other four clients), where `attach_secrets` did not patch it and its derived `Debug`/`Display` leaked the secret — now `MxaccessFailure` routes to `Error::MxAccess` (fixing a real cross-client inconsistency; an existing test flipped from `Error::Command` to `Error::MxAccess`). (3, Go, important) the CLI-44 `sendTerminalEventResult` was unconditionally non-blocking, so on the never-drop `SubscribeEvents`/`SubscribeEventsAfter` path (`cancelWhenResultBufferFull=false`) a genuine terminal error under a full buffer hit the `default:` and was silently dropped — now the terminal send is reserved-slot-non-blocking only for the cancel-on-overflow path and **blocking** for the never-drop path. A new shared fixture `authenticate-user.echoed-credential-mxaccess-failure.reply.json` (the echo under `MXACCESS_FAILURE`) is wired into all five suites. Minors also landed: whitespace-only-secret guard on the .NET/Java redact helpers; Java `invokeCommandRedacted` now preserves the exception subtype (mirroring .NET's type switch) instead of collapsing to the base type; dedicated redaction-helper unit tests (multiple occurrences, substring-overlap, empty/blank secrets) in Go/Java/.NET. Re-verified all five green (Go `go test ./...` ok + gofmt clean; Python 163 passed/1 skipped; .NET build 0 warnings + 128 passed/1 skipped; Rust fmt/check/`test --workspace`/clippy `-D warnings` all clean; Java gradle BUILD SUCCESSFUL, no generated churn). | | 2026-08-07 | **GWC-28, GWC-29, GWC-30, TST-28 → `Done`** (branch `fix/gwc-28-29-30-polish`). GWC-28: `WorkerClient.WriteLoopAsync` now stamps `envelope.Sequence = unchecked(++_nextSequence)` immediately before `_writer.WriteAsync`, and `CreateEnvelope` leaves it unset; `_nextSequence` dropped from `long` + `Interlocked` to a plain `ulong` touched only by the write loop (the channel's single consumer, `SingleReader = true`), so wire order and sequence order are the same thing by construction. Mirrors the worker's WRK-04 stamping, which the gateway half had never received; `gateway.md`'s envelope-sequence rule now states that both sides stamp at write inside their single write path and that inbound enforcement (still open, old **GWC-10**) would rely on it. New `WorkerClientTests.ConcurrentInvokesEmitStrictlyIncreasingSequencesOnTheWire` (32 parallel invokes, sequences asserted strictly increasing in wire order) failed 3/3 pre-fix. GWC-29: added `MxAccessGrpcMapper.MapCommand(MxCommand)`; `Invoke` no longer deep-clones the whole `MxCommandRequest` just to overwrite and discard its command. The one clone inside `MapCommand` stays and is documented as required — `commandToInvoke` may be the gRPC-owned `request.Command` and is read again after dispatch by `TrackCommandReply`, so it is what keeps `CreateCommandEnvelope`'s no-aliasing invariant true. New `MxAccessGrpcMapperTests.MapCommandFromCommandClonesPayload` (isolation + both overloads equal under a `FakeTimeProvider`). GWC-30: `WorkerFrameReader` reuses a per-instance `_lengthPrefix` scratch buffer instead of allocating 4 bytes per frame, with a class remark that `ReadAsync` is not reentrant (single read loop per `WorkerClient`; handshake reads complete before the loop starts); guarded by new `WorkerFrameProtocolTests.ReadAsync_WithMultipleFramesOnOneReader_ParsesEveryFrame` (5 frames, varying payload lengths, one reader). TST-28: new `[Theory] WorkerClientTests.StartAsync_SendsGatewayHelloWithConfiguredMaxFrameBytes` over the default and a 2 MiB override via `FakeWorkerHarness.CreateConnectedPairAsync(maxMessageBytes:)` — test-only, and the mutation check (hard-code `MaxFrameBytes = 0`) failed both cases before being reverted. Verification: `NonWindows.slnx` 0 warnings/0 errors; `WorkerClientTests` 25 passed, `WorkerFrameProtocolTests` 11 passed, `MxAccessGrpcMapperTests` 6 passed, `MxAccessGatewayService*` 29 passed, full gateway suite 844 passed / 0 failed (`TMPDIR=/tmp` on macOS). | +| 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.) | 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 558831a..b67c438 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. @@ -634,6 +671,13 @@ consumer that merely drains slower than this worker produces (the staging bound, metric `QueueOverflow("worker-event-staging")`). See `docs/GatewayProcessDesign.md`. A worker that outruns its consumer therefore dies loudly rather than growing gateway memory silently. +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 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 4280be2..460f120 100644 --- a/gateway.md +++ b/gateway.md @@ -474,10 +474,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..ff9020e 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,21 @@ 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 truncate. + // + // 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; + /// Verifies that valid gateway hello triggers worker hello and ready responses. /// A task that represents the asynchronous operation. [Fact] @@ -487,6 +502,361 @@ 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(RepeatedDrainEventCount, 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 < 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(RepeatedDrainEventCount, recovered.Count); + for (int index = 0; index < recovered.Count; index++) + { + Assert.Equal((ulong)(index + 1), recovered[index]); + } + + 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 + /// 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 +1611,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 +1628,8 @@ public sealed class WorkerPipeSessionTests options, () => 1234, sessionOptions, - () => runtime); + () => runtime, + logger); } private static WorkerFrameProtocolOptions CreateOptions() @@ -1270,7 +1650,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 +1663,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 +1777,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..1492ec0 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,157 @@ 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)); + } + + // 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) * 3 / 2); + + 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); + } + + /// + /// 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. + /// + [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 +300,42 @@ public sealed class MxAccessEventQueueTests Assert.Equal(WorkerFaultCategory.MxaccessEventConversionFailed, queue.Fault?.Category); } + // 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; + + /// + /// 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. 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. + 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..b8fdd5d 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,105 @@ 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 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( + 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 +713,77 @@ 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, 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() + { + return Math.Max( + _options.MaxMessageBytes - DrainReplyFrameHeadroomBytes, + _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 +850,30 @@ 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. 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 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/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..ed27a67 100644 --- a/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/MxAccessEventQueue.cs +++ b/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/MxAccessEventQueue.cs @@ -26,6 +26,15 @@ public sealed class MxAccessEventQueue /// public const int DefaultCapacity = 10000; + // 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; private readonly Queue events; private readonly object syncRoot = new(); @@ -209,6 +218,64 @@ 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() — 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. + /// 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; } +}