From 8df35cd63ab5809d76095541f8ad5b6b5b48295e Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Fri, 7 Aug 2026 07:50:38 -0400 Subject: [PATCH 1/2] fix(WRK-22,WRK-24,WRK-25,WRK-27,IPC-26): worker write-seam hardening WRK-22/IPC-26: tombstone a WriteAsync/WriteBatchAsync cancelled while waiting for the write lock (PendingFrame.Claimed under _gate; DequeueNext skips cancelled, claims the frame it returns) so a cancelled write never reaches the wire unless already claimed mid-write (documented residual). WRK-25: add WriteBatchAsync; RunEventDrainLoopAsync submits the drained event batch through it, so a burst of N events costs one flush not N. IPC-30 oversized-event structured fault preserved via FindOversizedEvent. WRK-24: reject a below-1024 negotiated frame maximum at the handshake (MinNegotiableFrameBytes, matching GatewayOptionsValidator floor). WRK-27: alarm poll advertises StaCallInProgress on the heartbeat snapshot so the watchdog suppresses to the ceiling, not the grace. Docs (WorkerFrameProtocol.md, MxAccessWorkerInstanceDesign.md) and the 2026-07-12 remediation registers/change-log updated in the same commit. --- .../2026-07-12/remediation/00-tracking.md | 11 +- .../2026-07-12/remediation/20-worker.md | 8 +- .../remediation/30-contracts-ipc.md | 2 +- docs/MxAccessWorkerInstanceDesign.md | 17 ++ docs/WorkerFrameProtocol.md | 46 +++- .../Ipc/WorkerFrameProtocolTests.cs | 227 ++++++++++++++++ .../Ipc/WorkerPipeSessionTests.cs | 250 ++++++++++++++++++ .../MxAccess/MxAccessStaSessionTests.cs | 72 ++++- .../TestSupport/FakeRuntimeSession.cs | 17 ++ .../Ipc/WorkerFrameProtocolOptions.cs | 23 +- .../Ipc/WorkerFrameWriter.cs | 175 +++++++++++- .../Ipc/WorkerPipeSession.cs | 78 ++++-- .../MxAccess/MxAccessStaSession.cs | 27 +- .../WorkerRuntimeHeartbeatSnapshot.cs | 17 +- 14 files changed, 912 insertions(+), 58 deletions(-) diff --git a/archreview/2026-07-12/remediation/00-tracking.md b/archreview/2026-07-12/remediation/00-tracking.md index 6911b42..8c588d8 100644 --- a/archreview/2026-07-12/remediation/00-tracking.md +++ b/archreview/2026-07-12/remediation/00-tracking.md @@ -72,12 +72,12 @@ 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 | 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-22 | Low | — | S | IPC-26 (fix owned here) | Done | Cancelled `WriteAsync` leaves its frame queued; it is still written later | | 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-24 | Low | — | S | — | Done | `AdoptNegotiatedMaxMessageBytes` has no lower-bound sanity check | +| WRK-25 | Low | P2 | S | WRK-22 (shared seam) | Done | WRK-12 flush coalescing never engages on the event hot path | | WRK-26 | Low | P1 | S | WRK-23 (soft); discharges IPC-29 | Done | 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-27 | Low | — | S | — | Done | Alarm poll bypasses the watchdog's in-flight suppression (15 s vs 75 s) | | 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) @@ -87,7 +87,7 @@ Full design + implementation for each row lives in the linked domain doc under i | 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-26 | Low | P2 | S | WRK-22 (mechanics) | Done (mechanics in WRK-22) | 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 | — | Done | docs/Grpc.md missing CommandTooLarge → ResourceExhausted mapping | | IPC-29 | Low | — | S | WRK-26 (discharged by) | Done (discharged by WRK-26) | WorkerFrameProtocol.md missing write-scheduling/sequencing section | @@ -174,4 +174,5 @@ Sequence these together rather than piecemeal — several are one change set spa | 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.) | +| 2026-08-07 | **Worker-seam batch → `Done`: WRK-22 (mechanics for IPC-26), WRK-24, WRK-25, WRK-27** (branch `fix/wrk-22-25-seam`). **WRK-22/IPC-26**: `WorkerFrameWriter.PendingFrame` gained a `Claimed` field; a `WriteAsync`/`WriteBatchAsync` cancelled while waiting for the write lock tombstones its still-unclaimed frame (`TrySetCanceled` under `_gate`) and `DequeueNext` skips cancelled frames and marks the one it returns `Claimed`, so a cancelled write never reaches the wire — except the documented, by-design residual where a lock-holder claimed the frame first (mid-write, cannot be recalled; caller still observes cancellation). **WRK-25**: new `WriteBatchAsync(IReadOnlyList, priority, ct)` enqueues a whole batch under one `_gate` acquisition, takes the lock once, drains, then observes every completion (surfacing the first per-frame rejection); `RunEventDrainLoopAsync` now submits the drained event batch through it, so a burst of N events costs one flush not N — the WRK-12 coalescing now engages on the event hot path. IPC-30's oversized-event structured fault is preserved (`FindOversizedEvent` maps the batch rejection back to the offending event). **WRK-24**: `WorkerFrameProtocolOptions.MinNegotiableFrameBytes = 1024` (matches `GatewayOptionsValidator.MinimumMaxMessageBytes`); `AdoptNegotiatedMaxMessageBytes` now faults a below-floor negotiated value at the handshake, closing the [1024, 256 MiB] accepted range. **WRK-27**: alarm poll runs outside the dispatcher, so `MxAccessStaSession` sets a `volatile staAlarmPollInProgress` around the `PollOnce` COM call and surfaces it on the new `WorkerRuntimeHeartbeatSnapshot.StaCallInProgress`; `ReportWatchdogFaultIfNeededAsync` honors it alongside `CurrentCommandCorrelationId`, so a healthy-but-slow poll gets grace-to-ceiling suppression (not the 15 s grace) but still faults past the 75 s ceiling. Docs same commit: `docs/WorkerFrameProtocol.md` (accepted-range paragraph, flush-coalescing sentence flipped to coalesced-on-drain, cancellation-tombstone contract replacing the WRK-26 "pending" placeholder), `docs/MxAccessWorkerInstanceDesign.md` (watchdog alarm-poll paragraph). New tests: `WorkerFrameProtocolTests.{WriteAsync_CancelledWhileWaitingForLock_FrameIsNeverWritten, WriteAsync_CancelledEventFrame_DoesNotTrailShutdownAck, WriteBatchAsync_FlushesOnceAndPreservesOrder, WriteBatchAsync_ControlFrameQueuedDuringBatch_JumpsRemainingEvents, AdoptNegotiatedMaxMessageBytes_BelowFloor_ThrowsInvalidConfiguration}`, `WorkerPipeSessionTests.{Watchdog_StaCallInProgress_SuppressedUntilCeiling, EventBurst_DrainLoopCoalescesFlushes, Handshake_GatewayHelloWithTinyMaxFrameBytes_FaultsAtHandshake}`, `MxAccessStaSessionTests.CaptureHeartbeat_DuringAlarmPoll_ReportsStaCallInProgress` (+ `FakeRuntimeSession.EnqueueEvents` bulk helper, `staCallInProgress` snapshot ctor param). **IPC-26 → `Done`** (mechanics owned here). **Evidence** — macOS: `dotnet build src/ZB.MOM.WW.MxGateway.NonWindows.slnx` . windev: . | | 2026-08-07 | **P1 doc-drift batch → `Done`: TST-27, WRK-26 (discharges IPC-29), CLI-42, CLI-43, IPC-28** (branch `fix/doc-drift-batch`). Doc-only; no source, proto, or test changes — cross-checked against HEAD in this worktree. **TST-27**: `docs/GatewayConfiguration.md`'s `ShowTagValues` row no longer says "Reserved" — it now states what `false` (default) does (`DashboardEventBroadcaster` blanks tag values from a deep-cloned `MxEvent` before the SignalR events-hub mirror, metadata still renders), the security relevance (the per-session hub ACL, SEC-25 roadmap item 12, still does not exist, so this redaction is the only thing between a low-trust Viewer and other sessions' tag values), and the honest scope limit (the flag does **not** cover `/browse`). **WRK-26** (discharges **IPC-29**): `docs/MxAccessWorkerInstanceDesign.md`'s "Outbound Queues" section rewritten from the stale five-level priority list to the two-class `Control`/`Event` scheduler actually shipped (`WorkerFrameWriter`/`WorkerFrameWritePriority.cs`), with the collapsed-decision rationale recorded, and the overflow paragraph rewritten to the implemented fail-fast (`WorkerFault` category `QueueOverflow` → fault frame written → `RunAsync` unwinds → generic `WorkerExitCode.UnexpectedFailure`, dedicated code still open). `docs/WorkerFrameProtocol.md` gained a new "Write Scheduling And Sequencing" section: the two priority classes, enqueue-then-contend/single-lock-holder-drains-all, write-time peek-stamp-commit sequencing, per-frame-rejection vs. stream-failure semantics, and flush coalescing — stated truthfully as landed (WRK-23's peek-stamp-commit is live at HEAD) or not (the drain loop still awaits each event `WriteAsync` individually, so WRK-25's N-events-one-flush batching has **not** landed and the section says so explicitly). Cancellation is deliberately **not** documented as a firm contract — a one-paragraph placeholder notes it is pending WRK-22, which has not landed (confirmed by reading `WorkerFrameWriter.cs`: no `Claimed`/tombstone machinery exists yet). `gateway.md:328-330` was cross-checked and left unchanged — its sequence prose (both sides stamp at write, per GWC-28) already reads true. **CLI-42**: `clients/rust/README.md` and `docs/ClientPackaging.md`'s Rust section now document the vendored proto layout matching `clients/rust/build.rs` exactly — repo-path-first resolution (`../../src/ZB.MOM.WW.MxGateway.Contracts/Protos`) falling back to `clients/rust/protos/` when the canonical path is absent (published-tarball case), the same-commit refresh rule enforced by `scripts/check-codegen.ps1` Check 3, and why `cargo package`/`cargo publish` run without `--no-verify` (matches `scripts/pack-clients.ps1:190-192`). **CLI-43**: `docs/style-guides/JavaStyleGuide.md` line 8 now says "Target Java 17 (the Ignition 8.3 baseline...)" mirroring the CLI-12 wording, matching the shipped `clients/java/build.gradle` toolchain-17 build. **IPC-28**: `docs/Grpc.md`'s exception-mapping prose gained `CommandTooLarge` → `ResourceExhausted` (verified against the live `switch` in `Grpc/MxAccessGatewayService.cs:950-960`), and the `Invoke` section gained one sentence on the oversized-payload path (`WorkerClient.InvokeAsync` rejects at the enqueue boundary per-correlation, session not faulted — verified against `WorkerClient.cs:220-234`), cross-referencing the headroom rule already documented in `docs/GatewayConfiguration.md:120-129`. Did not touch the DrainEvents-truncation row or the proto/`Generated/` trees — those belong to a parallel codegen task per the handoff note. **Source files cross-read for accuracy** (no edits): `src/ZB.MOM.WW.MxGateway.Worker/Ipc/WorkerFrameWriter.cs`, `.../WorkerFrameWritePriority.cs`, `.../WorkerPipeSession.cs` (confirmed two-class scheduler, WRK-21/23/28/30 landed, WRK-25/WRK-22 not landed), `src/ZB.MOM.WW.MxGateway.Worker/WorkerApplication.cs` (exit-code mapping), `src/ZB.MOM.WW.MxGateway.Worker/MxAccess/MxAccessEventQueue.cs` (overflow fault path), `src/ZB.MOM.WW.MxGateway.Server/Dashboard/Hubs/DashboardEventBroadcaster.cs` + `Configuration/DashboardOptions.cs` + `docs/GatewayDashboardDesign.md:170` (ShowTagValues), `src/ZB.MOM.WW.MxGateway.Server/Grpc/MxAccessGatewayService.cs:940-963` + `Workers/WorkerClient.cs:205-244` + `Workers/WorkerClientErrorCode.cs` (CommandTooLarge mapping), `clients/rust/build.rs`, `clients/rust/Cargo.toml`, `scripts/check-codegen.ps1`, `scripts/pack-clients.ps1` (Rust vendoring), `gateway.md:326-360` (sequence-prose cross-check). Verification (greps, doc-only — no build required): `grep -n 'Reserved' docs/GatewayConfiguration.md` no longer matches the `ShowTagValues` row; `grep -n 'faults' docs/MxAccessWorkerInstanceDesign.md` shows no remaining five-level list; `grep -n 'scheduling' docs/WorkerFrameProtocol.md` finds the new section; `grep -rn 'Java 21' docs/style-guides/` empty; `grep -i vendored docs/ClientPackaging.md clients/rust/README.md` non-empty in both; `grep -n 'CommandTooLarge' docs/Grpc.md` shows the mapping. | diff --git a/archreview/2026-07-12/remediation/20-worker.md b/archreview/2026-07-12/remediation/20-worker.md index 63f85ff..93ab57f 100644 --- a/archreview/2026-07-12/remediation/20-worker.md +++ b/archreview/2026-07-12/remediation/20-worker.md @@ -17,12 +17,12 @@ 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) | 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-22 | Low | — | S | IPC-26 (same defect, fix owned here) | Done | Cancelled `WriteAsync` leaves its frame queued; it is still written later | | 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-24 | Low | — | S | — | Done | `AdoptNegotiatedMaxMessageBytes` has no lower-bound sanity check | +| WRK-25 | Low | P2 | S | WRK-22 (both touch enqueue/dequeue) | Done | WRK-12 flush coalescing never engages on the event hot path | | WRK-26 | Low | P1 | S | WRK-23 (soft — sequence prose); discharges IPC-29 | Done | 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-27 | Low | — | S | — | Done | 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) | 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 ce367d9..4947e9e 100644 --- a/archreview/2026-07-12/remediation/30-contracts-ipc.md +++ b/archreview/2026-07-12/remediation/30-contracts-ipc.md @@ -15,7 +15,7 @@ All `path:line` citations were re-verified against the working tree at `4f5371f` | 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-26 | Low | P2 | S¹ | WRK-22 | Done (mechanics landed in WRK-22) | 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 | — | Done | `docs/Grpc.md` omits the `CommandTooLarge` → `ResourceExhausted` mapping | | IPC-29 | Low | — | S | — | Done (discharged by WRK-26) | Worker writer priority scheduling and write-time sequence stamping undocumented in the frame-protocol doc | diff --git a/docs/MxAccessWorkerInstanceDesign.md b/docs/MxAccessWorkerInstanceDesign.md index ccc4c42..29211b5 100644 --- a/docs/MxAccessWorkerInstanceDesign.md +++ b/docs/MxAccessWorkerInstanceDesign.md @@ -747,6 +747,23 @@ heartbeat fields until dedicated thresholds own those warnings. The worker reports stale STA activity, but the gateway owns the final kill decision through its existing heartbeat and worker lifecycle policy. +The alarm poll runs outside the command dispatcher — `RunAlarmPollLoopAsync` +invokes `PollOnce` directly on the STA rather than through +`StaCommandDispatcher`, so it does not inflate `PendingCommandCount` or perturb +dispatch ordering for real gateway commands. Because it is not a dispatched +command it has no `CurrentCommandCorrelationId`, so a healthy-but-slow poll (a +large `GetXmlCurrentAlarms2` against a busy provider) blocking the STA past +`HeartbeatGrace` would otherwise fault a healthy session at 15 s while a +dispatched command gets the 75 s ceiling. To close that asymmetry (WRK-27) the +poll advertises itself on the heartbeat snapshot's `StaCallInProgress` flag — +set on the STA thread for exactly the span of the COM call — and the watchdog +suppression honors that flag alongside `CurrentCommandCorrelationId`. The poll +therefore receives the same grace-to-ceiling treatment as a dispatched command: +suppressed up to `HeartbeatStuckCeiling`, faulted past it (a poll that blocks +the STA more than 75 s without pumping *should* fault — that is the ceiling's +contract). The flag is named generically so any future non-dispatcher STA work +reuses it. + The in-flight-command suppression itself is bounded by `WorkerPipeSessionOptions.HeartbeatStuckCeiling` (default 75 seconds = 5 × `HeartbeatGrace`). The motivating case for the suppression is a legitimately diff --git a/docs/WorkerFrameProtocol.md b/docs/WorkerFrameProtocol.md index 5bf4018..a23631b 100644 --- a/docs/WorkerFrameProtocol.md +++ b/docs/WorkerFrameProtocol.md @@ -29,6 +29,17 @@ 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. +The worker accepts a negotiated value in the closed range [1024, 256 MiB] +(`MinNegotiableFrameBytes` .. `MaxNegotiableFrameBytes`); 0 keeps the default. +A value outside that range is rejected at the handshake with a fault frame +rather than adopted, because a nonsensical maximum — a gateway bug or a +foreign/old peer — would otherwise leave a session that handshakes cleanly and +then fails every subsequent frame with per-frame size errors, the worst +diagnostic shape for an operator. The 1024-byte floor matches the gateway's own +`GatewayOptionsValidator.MinimumMaxMessageBytes`, so the worker never rejects a +value the gateway's validator accepts as legal configuration, and 1024 still +guarantees hellos, heartbeats, acks, and faults fit. + 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 @@ -112,20 +123,29 @@ runs after the whole batch, and only then does every successfully-written frame's completion resolve — so a caller's `WriteAsync` still does not complete until its bytes are both written *and* flushed, but a batch that happened to contain several queued frames pays one flush instead of one per -frame. In practice this coalescing currently engages only when multiple -frames are queued at the moment a lock-holder starts draining. The event -drain loop (`WorkerPipeSession.RunEventDrainLoopAsync`) awaits each drained -event's `WriteAsync` individually before writing the next, so today at most -one event frame is queued per drain pass and each event still costs its own -flush; a dedicated batch write entry point that submits a whole drained -event batch under one lock acquisition is designed but not yet landed, so a -burst of N events currently costs N flushes on the event hot path, not one. +frame. The event drain loop (`WorkerPipeSession.RunEventDrainLoopAsync`) +submits a whole drained event batch through `WriteBatchAsync`, which enqueues +every frame under one `_gate` acquisition, takes the write lock once, and +drains them together, so a burst of N events costs one flush rather than N — +the coalescing the batch machinery was built for now engages on the event hot +path, not only when independent producers happen to queue behind a blocked +write. Intra-batch order is preserved (FIFO enqueue under one lock), and a +concurrently queued control frame is still drained ahead of the batch. A +per-frame rejection inside a batch (for example one oversized event) surfaces +from the batch's awaited completions as that frame's +`WorkerFrameProtocolException`; the remaining completions are still observed +so none faults unobserved. -Cancellation semantics for a `WriteAsync` call that is still waiting for the -write lock when its token fires are not yet defined at this layer — pending -a fix that will tombstone the queued frame so a cancelled call is guaranteed -never to reach the wire. Until that lands, a cancelled caller may still see -its frame written by whichever caller next holds the lock. +Cancellation of a `WriteAsync`/`WriteBatchAsync` call that is still waiting +for the write lock when its token fires tombstones the queued frame: the +cancelled caller marks its frame under `_gate`, and the draining lock-holder's +`DequeueNext` skips any tombstoned frame, so a cancelled call is guaranteed +never to reach the wire — *unless* a lock-holder has already claimed the frame +to write it. Claiming and cancelling are interlocked under `_gate`, so exactly +one wins; a frame already claimed is mid-write and can no longer be recalled, +so the caller observes `OperationCanceledException` while that one frame still +reaches the wire. That residual window is by design: blocking the canceller +behind the very write it is abandoning would defeat the point of cancellation. ## Verification 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 6cd43d3..334e4a9 100644 --- a/src/ZB.MOM.WW.MxGateway.Worker.Tests/Ipc/WorkerFrameProtocolTests.cs +++ b/src/ZB.MOM.WW.MxGateway.Worker.Tests/Ipc/WorkerFrameProtocolTests.cs @@ -481,6 +481,192 @@ public sealed class WorkerFrameProtocolTests Assert.Equal(WorkerFrameProtocolErrorCode.InvalidConfiguration, exception.ErrorCode); } + /// + /// Verifies a negotiated frame maximum below the worker floor is rejected as + /// InvalidConfiguration (WRK-24), and that exactly the floor is adopted. The floor matches + /// the gateway's own GatewayOptionsValidator.MinimumMaxMessageBytes, so the worker never + /// rejects a value the gateway's validator accepts, yet a nonsensical tiny value faults at the + /// handshake instead of leaving a session that fails every later frame. + /// + [Fact] + public void AdoptNegotiatedMaxMessageBytes_BelowFloor_ThrowsInvalidConfiguration() + { + WorkerFrameProtocolOptions belowFloor = CreateOptions(); + WorkerFrameProtocolException exception = Assert.Throws( + () => belowFloor.AdoptNegotiatedMaxMessageBytes(512)); + Assert.Equal(WorkerFrameProtocolErrorCode.InvalidConfiguration, exception.ErrorCode); + + // Boundary: exactly the floor is accepted. + WorkerFrameProtocolOptions atFloor = CreateOptions(); + atFloor.AdoptNegotiatedMaxMessageBytes((uint)WorkerFrameProtocolOptions.MinNegotiableFrameBytes); + Assert.Equal(WorkerFrameProtocolOptions.MinNegotiableFrameBytes, atFloor.MaxMessageBytes); + } + + /// + /// WRK-22 / IPC-26. A WriteAsync cancelled while it waits for the write lock must never + /// have its frame written by the next lock-holder. Writer A holds the lock mid-write (blocked in + /// the stream); an event write is queued and then cancelled; when A is released and a later + /// control frame drains, the wire carries A's frame and the control frame only — the cancelled + /// event envelope is tombstoned and skipped. + /// + /// A task that represents the asynchronous operation. + [Fact] + public async Task WriteAsync_CancelledWhileWaitingForLock_FrameIsNeverWritten() + { + WorkerFrameProtocolOptions options = CreateOptions(); + using GatedWriteStream stream = new(); + WorkerFrameWriter writer = new(stream, options); + + // Writer A occupies the writer and blocks inside the stream, holding the write lock. + Task firstWrite = writer.WriteAsync(CreateGatewayHelloEnvelope(), WorkerFrameWritePriority.Control); + await AwaitWithTimeoutAsync(stream.FirstWriteStarted); + + // Queue an event write with its own CTS while the lock is held, then cancel it. + using CancellationTokenSource cts = new(); + Task cancelledWrite = writer.WriteAsync(CreateEventEnvelope(), WorkerFrameWritePriority.Event, cts.Token); + await Task.Delay(50); + cts.Cancel(); + await Assert.ThrowsAnyAsync(async () => await cancelledWrite); + + // Release A, then drive a fresh control write. + stream.ReleaseFirstWrite(); + await AwaitWithTimeoutAsync(firstWrite); + await writer.WriteAsync(CreateGatewayHelloEnvelope(), WorkerFrameWritePriority.Control); + + stream.Position = 0; + WorkerFrameReader reader = new(stream, options); + WorkerEnvelope frame1 = await reader.ReadAsync(); + WorkerEnvelope frame2 = await reader.ReadAsync(); + Assert.Equal(WorkerEnvelope.BodyOneofCase.GatewayHello, frame1.BodyCase); + Assert.Equal(WorkerEnvelope.BodyOneofCase.GatewayHello, frame2.BodyCase); + // The cancelled event never reached the wire — no third frame, and sequences stay contiguous. + Assert.Equal(stream.Length, stream.Position); + Assert.Equal(1UL, frame1.Sequence); + Assert.Equal(2UL, frame2.Sequence); + } + + /// + /// WRK-22 / IPC-26, the review's shutdown scenario. A cancelled event frame queued before a + /// shutdown-ack control frame must not trail the ack on the wire: the tombstone rule plus the + /// control-before-event scheduler keeps the ack the last frame written. + /// + /// A task that represents the asynchronous operation. + [Fact] + public async Task WriteAsync_CancelledEventFrame_DoesNotTrailShutdownAck() + { + WorkerFrameProtocolOptions options = CreateOptions(); + using GatedWriteStream stream = new(); + WorkerFrameWriter writer = new(stream, options); + + Task firstWrite = writer.WriteAsync(CreateGatewayHelloEnvelope(), WorkerFrameWritePriority.Control); + await AwaitWithTimeoutAsync(stream.FirstWriteStarted); + + using CancellationTokenSource cts = new(); + Task cancelledEvent = writer.WriteAsync(CreateEventEnvelope(), WorkerFrameWritePriority.Event, cts.Token); + await Task.Delay(50); + cts.Cancel(); + await Assert.ThrowsAnyAsync(async () => await cancelledEvent); + + // The shutdown ack (a control frame) is queued behind the still-blocked first write. + Task ackWrite = writer.WriteAsync(CreateShutdownAckEnvelope(), WorkerFrameWritePriority.Control); + await Task.Delay(50); + + stream.ReleaseFirstWrite(); + await AwaitWithTimeoutAsync(Task.WhenAll(firstWrite, ackWrite)); + + stream.Position = 0; + WorkerFrameReader reader = new(stream, options); + WorkerEnvelope frame1 = await reader.ReadAsync(); + WorkerEnvelope frame2 = await reader.ReadAsync(); + Assert.Equal(WorkerEnvelope.BodyOneofCase.GatewayHello, frame1.BodyCase); + // The ack is the last frame — the cancelled event did not trail it. + Assert.Equal(WorkerEnvelope.BodyOneofCase.WorkerShutdownAck, frame2.BodyCase); + Assert.Equal(stream.Length, stream.Position); + } + + /// + /// WRK-25. The batch entry point enqueues a whole event burst under one lock acquisition and + /// drains it together, so N events cost exactly one flush and reach the wire in batch order with + /// monotonic sequences — the coalescing WRK-12 shipped, now on the event hot path. + /// + /// A task that represents the asynchronous operation. + [Fact] + public async Task WriteBatchAsync_FlushesOnceAndPreservesOrder() + { + const int count = 8; + WorkerFrameProtocolOptions options = CreateOptions(); + using FlushCountingStream stream = new(); + WorkerFrameWriter writer = new(stream, options); + + WorkerEnvelope[] batch = new WorkerEnvelope[count]; + for (int index = 0; index < count; index++) + { + batch[index] = CreateEventEnvelope(workerSequence: (ulong)(100 + index)); + } + + await writer.WriteBatchAsync(batch, WorkerFrameWritePriority.Event); + + // The whole batch was queued before the single lock wait, so it drained in one pass => one flush. + Assert.Equal(1, stream.FlushCount); + + stream.Position = 0; + WorkerFrameReader reader = new(stream, options); + for (int index = 0; index < count; index++) + { + WorkerEnvelope frame = await reader.ReadAsync(); + Assert.Equal(WorkerEnvelope.BodyOneofCase.WorkerEvent, frame.BodyCase); + // Wire order matches batch order. + Assert.Equal((ulong)(100 + index), frame.WorkerEvent.Event.WorkerSequence); + // Write-time stamped sequence is monotonic 1..count. + Assert.Equal((ulong)(index + 1), frame.Sequence); + } + } + + /// + /// WRK-25. Control-before-event still holds mid-batch: a control frame queued while a batch is + /// draining jumps ahead of the batch's remaining events. + /// + /// A task that represents the asynchronous operation. + [Fact] + public async Task WriteBatchAsync_ControlFrameQueuedDuringBatch_JumpsRemainingEvents() + { + WorkerFrameProtocolOptions options = CreateOptions(); + using GatedWriteStream stream = new(); + WorkerFrameWriter writer = new(stream, options); + + WorkerEnvelope[] batch = new[] + { + CreateEventEnvelope(), + CreateEventEnvelope(), + CreateEventEnvelope(), + }; + + // The batch takes the lock and blocks writing its first event frame inside the stream. + Task batchWrite = writer.WriteBatchAsync(batch, WorkerFrameWritePriority.Event); + await AwaitWithTimeoutAsync(stream.FirstWriteStarted); + + // A control frame queued mid-drain must jump the batch's remaining events. + Task controlWrite = writer.WriteAsync(CreateGatewayHelloEnvelope(), WorkerFrameWritePriority.Control); + await Task.Delay(50); + + stream.ReleaseFirstWrite(); + await AwaitWithTimeoutAsync(Task.WhenAll(batchWrite, controlWrite)); + + stream.Position = 0; + WorkerFrameReader reader = new(stream, options); + WorkerEnvelope f1 = await reader.ReadAsync(); + WorkerEnvelope f2 = await reader.ReadAsync(); + WorkerEnvelope f3 = await reader.ReadAsync(); + WorkerEnvelope f4 = await reader.ReadAsync(); + + // First event was already writing when the control frame queued; the control frame then jumps + // ahead of the two remaining events. + Assert.Equal(WorkerEnvelope.BodyOneofCase.WorkerEvent, f1.BodyCase); + Assert.Equal(WorkerEnvelope.BodyOneofCase.GatewayHello, f2.BodyCase); + Assert.Equal(WorkerEnvelope.BodyOneofCase.WorkerEvent, f3.BodyCase); + Assert.Equal(WorkerEnvelope.BodyOneofCase.WorkerEvent, f4.BodyCase); + } + private static WorkerEnvelope CreateGatewayHelloEnvelope(ulong sequence = 1) { return new WorkerEnvelope @@ -522,6 +708,47 @@ public sealed class WorkerFrameProtocolTests }; } + private static WorkerEnvelope CreateEventEnvelope(ulong workerSequence) + { + WorkerEnvelope envelope = CreateEventEnvelope(); + envelope.WorkerEvent.Event.WorkerSequence = workerSequence; + return envelope; + } + + private static WorkerEnvelope CreateShutdownAckEnvelope() + { + return new WorkerEnvelope + { + ProtocolVersion = GatewayContractInfo.WorkerProtocolVersion, + SessionId = SessionId, + WorkerShutdownAck = new WorkerShutdownAck + { + Status = new ProtocolStatus + { + Code = ProtocolStatusCode.Ok, + Message = "OK", + }, + }, + }; + } + + // A MemoryStream that counts FlushAsync calls without gating any write, so a batch write can be + // asserted to flush exactly once. + private sealed class FlushCountingStream : MemoryStream + { + private int _flushCount; + + /// Gets the number of calls observed so far. + public int FlushCount => Volatile.Read(ref _flushCount); + + /// + public override Task FlushAsync(CancellationToken cancellationToken) + { + Interlocked.Increment(ref _flushCount); + return base.FlushAsync(cancellationToken); + } + } + // A MemoryStream whose first WriteAsync blocks until released, so a test can queue additional frames // behind an in-progress write and observe the writer's priority ordering. private sealed class GatedWriteStream : MemoryStream 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 ff9020e..208c54a 100644 --- a/src/ZB.MOM.WW.MxGateway.Worker.Tests/Ipc/WorkerPipeSessionTests.cs +++ b/src/ZB.MOM.WW.MxGateway.Worker.Tests/Ipc/WorkerPipeSessionTests.cs @@ -1112,6 +1112,184 @@ public sealed class WorkerPipeSessionTests await SendShutdownAndWaitAsync(pipePair, runTask, cancellation.Token); } + /// + /// WRK-27. An STA call outside the command dispatcher (the alarm poll) advertises itself on + /// the heartbeat snapshot's StaCallInProgress flag, and the watchdog grants it the same + /// grace-to-ceiling suppression as a dispatched command: stale STA activity within the ceiling + /// does not fault while the flag is set, but stale activity beyond the ceiling faults anyway. + /// This closes the 15 s-vs-75 s asymmetry between polls and commands. + /// + /// A task that represents the asynchronous operation. + [Fact] + public async Task Watchdog_StaCallInProgress_SuppressedUntilCeiling() + { + using CancellationTokenSource cancellation = new(TimeSpan.FromSeconds(10)); + + // Phase 1 — within the ceiling: stale beyond grace, empty correlation id, StaCallInProgress + // set. The default 75 s ceiling is far beyond the 5 s staleness, so the watchdog must suppress. + using (PipePair pipePair = await PipePair.CreateAsync(cancellation.Token)) + { + FakeRuntimeSession runtime = new(); + runtime.SetSnapshot(new WorkerRuntimeHeartbeatSnapshot( + DateTimeOffset.UtcNow - TimeSpan.FromSeconds(5), + pendingCommandCount: 0, + outboundEventQueueDepth: 0, + lastEventSequence: 0, + currentCommandCorrelationId: string.Empty, + staCallInProgress: true)); + WorkerPipeSession session = CreatePipeSession( + pipePair.WorkerStream, + runtime, + new WorkerPipeSessionOptions + { + HeartbeatInterval = TimeSpan.FromMilliseconds(20), + HeartbeatGrace = TimeSpan.FromMilliseconds(50), + }); + Task runTask = session.RunAsync(cancellation.Token); + await CompleteGatewayHandshakeAsync(pipePair, cancellation.Token); + + const int framesToInspect = 6; + int heartbeatsObserved = 0; + for (int index = 0; index < framesToInspect; index++) + { + WorkerEnvelope envelope = await pipePair.GatewayReader.ReadAsync(cancellation.Token); + Assert.NotEqual(WorkerEnvelope.BodyOneofCase.WorkerFault, envelope.BodyCase); + if (envelope.BodyCase == WorkerEnvelope.BodyOneofCase.WorkerHeartbeat) + { + heartbeatsObserved++; + } + } + + Assert.True( + heartbeatsObserved >= 2, + $"Expected multiple heartbeats during the in-progress STA-call window; observed {heartbeatsObserved}."); + + await SendShutdownAndWaitAsync(pipePair, runTask, cancellation.Token); + } + + // Phase 2 — beyond the ceiling: same StaCallInProgress flag, but staleness (5 s) exceeds the + // 200 ms ceiling, so the watchdog must fire even with the poll in progress. + using (PipePair pipePair = await PipePair.CreateAsync(cancellation.Token)) + { + FakeRuntimeSession runtime = new(); + runtime.SetSnapshot(new WorkerRuntimeHeartbeatSnapshot( + DateTimeOffset.UtcNow - TimeSpan.FromSeconds(5), + pendingCommandCount: 0, + outboundEventQueueDepth: 0, + lastEventSequence: 0, + currentCommandCorrelationId: string.Empty, + staCallInProgress: true)); + WorkerPipeSession session = CreatePipeSession( + pipePair.WorkerStream, + runtime, + new WorkerPipeSessionOptions + { + HeartbeatInterval = TimeSpan.FromMilliseconds(20), + HeartbeatGrace = TimeSpan.FromMilliseconds(50), + HeartbeatStuckCeiling = TimeSpan.FromMilliseconds(200), + }); + Task runTask = session.RunAsync(cancellation.Token); + await CompleteGatewayHandshakeAsync(pipePair, cancellation.Token); + + WorkerEnvelope fault = await ReadUntilAsync( + pipePair.GatewayReader, + WorkerEnvelope.BodyOneofCase.WorkerFault, + cancellation.Token); + + Assert.Equal(WorkerFaultCategory.StaHung, fault.WorkerFault.Category); + + await SendShutdownAndWaitAsync(pipePair, runTask, cancellation.Token); + } + } + + /// + /// WRK-25. The event drain loop submits a whole drained batch through the writer's batch entry + /// point, so a burst of 128 events costs one flush, not 128 — the assertion the WRK-12 + /// tracking claim needed to actually hold on the event hot path. + /// + /// A task that represents the asynchronous operation. + [Fact] + public async Task EventBurst_DrainLoopCoalescesFlushes() + { + using CancellationTokenSource cancellation = new(TimeSpan.FromSeconds(15)); + using PipePair pipePair = await PipePair.CreateAsync(cancellation.Token); + FakeRuntimeSession runtime = new(); + // A far-off heartbeat interval keeps heartbeat flushes out of the measurement window. + FlushCountingPassthroughStream countingStream = new(pipePair.WorkerStream); + WorkerPipeSession session = CreatePipeSession( + countingStream, + runtime, + new WorkerPipeSessionOptions + { + HeartbeatInterval = TimeSpan.FromMinutes(5), + HeartbeatGrace = TimeSpan.FromSeconds(30), + }); + Task runTask = session.RunAsync(cancellation.Token); + await CompleteGatewayHandshakeAsync(pipePair, cancellation.Token); + + // Let the idle drain loop settle (no events yet → no flushes) and record the baseline. + await Task.Delay(100, cancellation.Token); + int baselineFlushes = countingStream.FlushCount; + + // Enqueue a full 128-event batch atomically so the drain loop sees it as one batch. + const int burst = 128; + List batch = new(burst); + for (int index = 0; index < burst; index++) + { + batch.Add(CreateWorkerEvent(sequence: (ulong)(index + 1))); + } + + runtime.EnqueueEvents(batch); + + // Drain all 128 events off the gateway side. + for (int index = 0; index < burst; index++) + { + await ReadUntilAsync( + pipePair.GatewayReader, + WorkerEnvelope.BodyOneofCase.WorkerEvent, + cancellation.Token); + } + + // The whole burst cost exactly one additional flush. + Assert.Equal(1, countingStream.FlushCount - baselineFlushes); + + await SendShutdownAndWaitAsync(pipePair, runTask, cancellation.Token); + } + + /// + /// WRK-24. A GatewayHello negotiating a frame maximum below the worker floor faults at the + /// handshake with a fault frame rather than being adopted — mirroring the above-ceiling + /// handshake behavior — so a nonsensical tiny value never leaves a session that fails every + /// later frame. No message loop is entered. + /// + /// A task that represents the asynchronous operation. + [Fact] + public async Task Handshake_GatewayHelloWithTinyMaxFrameBytes_FaultsAtHandshake() + { + WorkerFrameProtocolOptions options = CreateOptions(); + using MemoryStream inbound = new(); + await new WorkerFrameWriter(inbound, options) + .WriteAsync(CreateGatewayHelloEnvelope(maxFrameBytes: 512)); + inbound.Position = 0; + using MemoryStream outbound = new(); + WorkerPipeSession session = CreateSession(inbound, outbound, options); + bool initialized = false; + + WorkerFrameProtocolException exception = + await Assert.ThrowsAsync( + async () => await session.CompleteStartupHandshakeAsync( + _ => + { + initialized = true; + return Task.CompletedTask; + })); + + Assert.False(initialized); + Assert.Equal(WorkerFrameProtocolErrorCode.InvalidConfiguration, exception.ErrorCode); + WorkerEnvelope fault = Assert.Single(ReadWrittenFrames(outbound, options)); + Assert.Equal(WorkerEnvelope.BodyOneofCase.WorkerFault, fault.BodyCase); + } + /// /// Regression test: a long in-flight STA command that keeps pumping /// must NOT self-fault as StaHung, and its reply must still be @@ -1976,6 +2154,78 @@ public sealed class WorkerPipeSessionTests } } + // Wraps the worker side of the pipe and counts FlushAsync calls so a test can assert the event + // drain loop coalesces a burst into a single flush. Delegates every other operation to the inner + // stream; does not own the inner stream's lifetime (PipePair disposes it). + private sealed class FlushCountingPassthroughStream : Stream + { + private readonly Stream inner; + private int flushCount; + + /// Initializes the passthrough over the given inner stream. + /// The stream to delegate to. + public FlushCountingPassthroughStream(Stream inner) + { + this.inner = inner; + } + + /// Gets the number of calls observed so far. + public int FlushCount => Volatile.Read(ref flushCount); + + /// + public override bool CanRead => inner.CanRead; + + /// + public override bool CanSeek => inner.CanSeek; + + /// + public override bool CanWrite => inner.CanWrite; + + /// + public override long Length => inner.Length; + + /// + public override long Position + { + get => inner.Position; + set => inner.Position = value; + } + + /// + public override void Flush() + { + Interlocked.Increment(ref flushCount); + inner.Flush(); + } + + /// + public override Task FlushAsync(CancellationToken cancellationToken) + { + Interlocked.Increment(ref flushCount); + return inner.FlushAsync(cancellationToken); + } + + /// + public override int Read(byte[] buffer, int offset, int count) => inner.Read(buffer, offset, count); + + /// + public override Task ReadAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken) + => inner.ReadAsync(buffer, offset, count, cancellationToken); + + /// + public override long Seek(long offset, SeekOrigin origin) => inner.Seek(offset, origin); + + /// + public override void SetLength(long value) => inner.SetLength(value); + + /// + public override void Write(byte[] buffer, int offset, int count) => inner.Write(buffer, offset, count); + + /// + public override Task WriteAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken) + => inner.WriteAsync(buffer, offset, count, cancellationToken); + } + private sealed class PipePair : IDisposable { private readonly NamedPipeServerStream gatewayStream; diff --git a/src/ZB.MOM.WW.MxGateway.Worker.Tests/MxAccess/MxAccessStaSessionTests.cs b/src/ZB.MOM.WW.MxGateway.Worker.Tests/MxAccess/MxAccessStaSessionTests.cs index ffa47f9..a8002b3 100644 --- a/src/ZB.MOM.WW.MxGateway.Worker.Tests/MxAccess/MxAccessStaSessionTests.cs +++ b/src/ZB.MOM.WW.MxGateway.Worker.Tests/MxAccess/MxAccessStaSessionTests.cs @@ -438,6 +438,53 @@ public sealed class MxAccessStaSessionTests Assert.Contains("alarm poll failed", fault.DiagnosticMessage, StringComparison.OrdinalIgnoreCase); } + /// + /// WRK-27. While the alarm poll's PollOnce is executing on the STA, a heartbeat captured mid-poll + /// must report so the watchdog + /// grants the poll the same grace-to-ceiling suppression as a dispatched command; once the poll + /// returns the flag clears. PollOnce is blocked on a gate so the heartbeat can be captured while + /// the STA call is genuinely in flight. + /// + /// A task that represents the asynchronous operation. + [Fact] + public async Task CaptureHeartbeat_DuringAlarmPoll_ReportsStaCallInProgress() + { + FakeAlarmCommandHandler handler = new() { BlockPoll = true }; + FakeMxAccessComObjectFactory factory = new(); + FakeMxAccessEventSink eventSink = new(); + using StaRuntime runtime = CreateRuntime(); + using MxAccessStaSession session = new( + runtime, + factory, + eventSink, + new MxAccessEventQueue(), + (_eq, _affinity, _comFactory) => handler); + + await session.StartAsync("session-1", workerProcessId: 1); + + // Wait until PollOnce is blocked mid-call on the STA thread. + Assert.True( + handler.WaitForPollEntered(TimeSpan.FromSeconds(5)), + "Expected the alarm poll to start within 5 seconds."); + + // Captured mid-poll, the heartbeat advertises the in-progress STA call. + Assert.True(session.CaptureHeartbeat().StaCallInProgress); + + // Release the poll and stop blocking; the flag clears once the poll returns. + handler.BlockPoll = false; + handler.ReleasePoll(); + + using CancellationTokenSource timeout = new(TimeSpan.FromSeconds(5)); + while (session.CaptureHeartbeat().StaCallInProgress && !timeout.IsCancellationRequested) + { + await Task.Delay(25, CancellationToken.None); + } + + Assert.False( + session.CaptureHeartbeat().StaCallInProgress, + "Expected StaCallInProgress to clear once the alarm poll returned."); + } + /// /// The STA-affinity guard throws when an /// IMxAccessAlarmConsumer call is attempted off the thread that created @@ -472,6 +519,8 @@ public sealed class MxAccessStaSessionTests private sealed class FakeAlarmCommandHandler : IAlarmCommandHandler { private readonly object gate = new object(); + private readonly ManualResetEventSlim pollEntered = new(false); + private readonly ManualResetEventSlim releasePoll = new(false); private int pollCount; private int? lastPollThreadId; @@ -484,6 +533,17 @@ public sealed class MxAccessStaSessionTests /// Exception thrown by PollOnce; null to succeed. public Exception? PollException { get; set; } + /// When set, blocks until is called. + public bool BlockPoll { get; set; } + + /// Waits until a blocking has entered and is blocked. + /// Maximum time to wait. + /// True if a poll entered within the timeout. + public bool WaitForPollEntered(TimeSpan timeout) => pollEntered.Wait(timeout); + + /// Releases a blocked on the gate. + public void ReleasePoll() => releasePoll.Set(); + /// Gets the count of PollOnce calls. public int PollCount { @@ -533,6 +593,12 @@ public sealed class MxAccessStaSessionTests lastPollThreadId = Thread.CurrentThread.ManagedThreadId; } + if (BlockPoll) + { + pollEntered.Set(); + releasePoll.Wait(TimeSpan.FromSeconds(10)); + } + if (PollException is not null) { throw PollException; @@ -540,6 +606,10 @@ public sealed class MxAccessStaSessionTests } /// - public void Dispose() { } + public void Dispose() + { + pollEntered.Dispose(); + releasePoll.Dispose(); + } } } 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 142551f..a992b12 100644 --- a/src/ZB.MOM.WW.MxGateway.Worker.Tests/TestSupport/FakeRuntimeSession.cs +++ b/src/ZB.MOM.WW.MxGateway.Worker.Tests/TestSupport/FakeRuntimeSession.cs @@ -372,6 +372,23 @@ internal sealed class FakeRuntimeSession : IWorkerRuntimeSession } } + /// + /// Enqueues a batch of worker events atomically under one lock so the drain loop cannot + /// observe a partial batch. Lets a test assert the drain loop coalesces a whole batch into one + /// flush (WRK-25) without racing a mid-enqueue drain that would split the batch. + /// + /// The events to enqueue in order. + public void EnqueueEvents(IEnumerable workerEvents) + { + lock (gate) + { + foreach (WorkerEvent workerEvent in workerEvents) + { + events.Enqueue(workerEvent); + } + } + } + /// public void Dispose() { diff --git a/src/ZB.MOM.WW.MxGateway.Worker/Ipc/WorkerFrameProtocolOptions.cs b/src/ZB.MOM.WW.MxGateway.Worker/Ipc/WorkerFrameProtocolOptions.cs index 96fdcde..76e56b2 100644 --- a/src/ZB.MOM.WW.MxGateway.Worker/Ipc/WorkerFrameProtocolOptions.cs +++ b/src/ZB.MOM.WW.MxGateway.Worker/Ipc/WorkerFrameProtocolOptions.cs @@ -18,6 +18,17 @@ public sealed class WorkerFrameProtocolOptions /// public const int MaxNegotiableFrameBytes = 256 * 1024 * 1024; + /// + /// Lower floor the worker will accept for a gateway-negotiated frame maximum + /// (GatewayHello.max_frame_bytes). Matches the gateway's own + /// GatewayOptionsValidator.MinimumMaxMessageBytes validation floor so the worker never + /// rejects a value the gateway's own validator accepts as legal configuration, yet a nonsensical + /// tiny value (a gateway bug or a foreign/old peer) is rejected at the handshake rather than + /// leaving a session that handshakes cleanly and then fails every subsequent frame with + /// per-frame size errors. 1024 bytes still guarantees hellos, heartbeats, acks, and faults fit. + /// + public const int MinNegotiableFrameBytes = 1024; + /// Initializes a new instance of the WorkerFrameProtocolOptions class from WorkerOptions. /// Worker initialization options. public WorkerFrameProtocolOptions(WorkerOptions options) @@ -118,7 +129,9 @@ public sealed class WorkerFrameProtocolOptions /// /// Adopts the gateway-negotiated frame maximum conveyed in GatewayHello.max_frame_bytes. /// A value of 0 (an older gateway that never set the field) is ignored and the - /// constructor default is kept. A value above is rejected. + /// constructor default is kept. A value outside the accepted range + /// [, ] is rejected so a + /// nonsensical negotiated value faults at the handshake rather than mid-session. /// /// The gateway-negotiated maximum, or 0 for "keep default". internal void AdoptNegotiatedMaxMessageBytes(uint negotiatedMaxFrameBytes) @@ -128,6 +141,14 @@ public sealed class WorkerFrameProtocolOptions return; } + if (negotiatedMaxFrameBytes < MinNegotiableFrameBytes) + { + throw new WorkerFrameProtocolException( + WorkerFrameProtocolErrorCode.InvalidConfiguration, + $"GatewayHello negotiated frame maximum {negotiatedMaxFrameBytes} is below the worker floor " + + $"of {MinNegotiableFrameBytes} bytes."); + } + if (negotiatedMaxFrameBytes > MaxNegotiableFrameBytes) { throw new WorkerFrameProtocolException( diff --git a/src/ZB.MOM.WW.MxGateway.Worker/Ipc/WorkerFrameWriter.cs b/src/ZB.MOM.WW.MxGateway.Worker/Ipc/WorkerFrameWriter.cs index 6a377a7..6061f00 100644 --- a/src/ZB.MOM.WW.MxGateway.Worker/Ipc/WorkerFrameWriter.cs +++ b/src/ZB.MOM.WW.MxGateway.Worker/Ipc/WorkerFrameWriter.cs @@ -1,6 +1,7 @@ using System; using System.Collections.Generic; using System.IO; +using System.Runtime.ExceptionServices; using System.Threading; using System.Threading.Tasks; using Google.Protobuf; @@ -33,6 +34,16 @@ public sealed class WorkerFrameWriter /// Gets the completion source signaled once the frame has been written or has failed. public TaskCompletionSource Completion { get; } + + /// + /// Set to true by — under _gate — at the instant the + /// draining lock-holder takes ownership of this frame to write it. A cancelled caller + /// tombstones its frame only while it is still unclaimed, so a claim and a cancel can never + /// both win: the flag is the interlock between the two. A frame already claimed is mid-write + /// and can no longer be recalled (see ). + /// Mutated only under _gate. + /// + public bool Claimed; } private readonly WorkerFrameProtocolOptions _options; @@ -75,6 +86,13 @@ public sealed class WorkerFrameWriter /// Scheduling priority; control frames are written ahead of event frames. /// Token to cancel waiting for the write lock. /// A task that completes when the frame has been written and flushed. + /// + /// Cancellation contract (WRK-22): if the token fires while this call is waiting for the write + /// lock, the frame is tombstoned so it is never written — unless a draining lock-holder has + /// already claimed it, in which case the frame may still reach the wire even though this call + /// observes . That residual window is by design: blocking + /// the canceller behind the very write it is abandoning would defeat the point of cancellation. + /// public async Task WriteAsync( WorkerEnvelope envelope, WorkerFrameWritePriority priority, @@ -101,7 +119,19 @@ public sealed class WorkerFrameWriter // Contend for the single writer: whoever wins drains every currently-queued frame in priority // order, so this frame is written by this call or by a concurrent caller that got the lock // first. Either way it completes via its own TaskCompletionSource. - await _writeLock.WaitAsync(cancellationToken).ConfigureAwait(false); + try + { + await _writeLock.WaitAsync(cancellationToken).ConfigureAwait(false); + } + catch (OperationCanceledException) + { + // Tombstone the queued frame so DequeueNext skips it — but only if a draining lock-holder + // has not already claimed it. If it is claimed it is mid-write and cannot be recalled; the + // caller still observes cancellation while the frame reaches the wire (documented above). + TombstoneIfUnclaimed(frame, cancellationToken); + throw; + } + try { await DrainQueuedFramesAsync().ConfigureAwait(false); @@ -114,6 +144,123 @@ public sealed class WorkerFrameWriter await frame.Completion.Task.ConfigureAwait(false); } + /// + /// Queues a whole batch of envelopes at one priority under a single lock acquisition and drains + /// it, so a burst of frames — the event drain loop's hot path — pays one flush for the batch + /// rather than one per frame (WRK-25, realizing the WRK-12 coalescing on the path it was built + /// for). Intra-batch order is preserved because the enqueue is atomic under _gate and each + /// class queue is FIFO; the control-before-event guarantee still holds because any concurrently + /// queued control frame is drained ahead of this batch by . Every frame's + /// "written and flushed before completion" contract is unchanged. + /// + /// Envelopes to write, in order. + /// Scheduling priority for the whole batch. + /// Token to cancel waiting for the write lock. + /// A task that completes when every frame in the batch has been written and flushed. + /// + /// A per-frame rejection inside the batch (for example one oversized event) surfaces from the + /// awaited completions as its ; the remaining frames are + /// still observed so none faults unobserved. Cancellation while waiting for the lock tombstones + /// every still-unclaimed frame in the batch, per the WRK-22 contract on + /// . + /// + public async Task WriteBatchAsync( + IReadOnlyList envelopes, + WorkerFrameWritePriority priority, + CancellationToken cancellationToken = default) + { + if (envelopes is null) + { + throw new ArgumentNullException(nameof(envelopes)); + } + + if (envelopes.Count == 0) + { + return; + } + + PendingFrame[] frames = new PendingFrame[envelopes.Count]; + for (int index = 0; index < envelopes.Count; index++) + { + WorkerEnvelope envelope = envelopes[index] + ?? throw new ArgumentException("Batch envelopes must not contain null.", nameof(envelopes)); + frames[index] = new PendingFrame(envelope); + } + + lock (_gate) + { + Queue queue = priority == WorkerFrameWritePriority.Event ? _eventFrames : _controlFrames; + foreach (PendingFrame frame in frames) + { + queue.Enqueue(frame); + } + } + + try + { + await _writeLock.WaitAsync(cancellationToken).ConfigureAwait(false); + } + catch (OperationCanceledException) + { + TombstoneUnclaimed(frames, cancellationToken); + throw; + } + + try + { + await DrainQueuedFramesAsync().ConfigureAwait(false); + } + finally + { + _writeLock.Release(); + } + + // Await every completion so no per-frame rejection faults unobserved, but surface the first + // failure (in batch order) to the caller — the drain loop maps it back to the offending event. + Exception? firstFailure = null; + foreach (PendingFrame frame in frames) + { + try + { + await frame.Completion.Task.ConfigureAwait(false); + } + catch (Exception exception) + { + firstFailure ??= exception; + } + } + + if (firstFailure is not null) + { + ExceptionDispatchInfo.Capture(firstFailure).Throw(); + } + } + + private void TombstoneIfUnclaimed(PendingFrame frame, CancellationToken cancellationToken) + { + lock (_gate) + { + if (!frame.Claimed) + { + frame.Completion.TrySetCanceled(cancellationToken); + } + } + } + + private void TombstoneUnclaimed(PendingFrame[] frames, CancellationToken cancellationToken) + { + lock (_gate) + { + foreach (PendingFrame frame in frames) + { + if (!frame.Claimed) + { + frame.Completion.TrySetCanceled(cancellationToken); + } + } + } + } + // Runs only under _writeLock. Drains control frames before event frames, stamping and writing each. // The stream write itself is not cancellable: a frame is written atomically or fails, never left // half-written on the pipe because a caller gave up waiting. @@ -198,18 +345,36 @@ public sealed class WorkerFrameWriter or WorkerFrameProtocolErrorCode.SessionMismatch; } + // Returns the next frame to write, control frames first, skipping any frame a cancelled caller + // tombstoned while it waited for the lock (WRK-22). The frame actually returned is marked Claimed + // under _gate in the same critical section that checks the tombstone, so a claim and a concurrent + // cancel are mutually exclusive: whichever acquires _gate first wins. private PendingFrame? DequeueNext() { lock (_gate) { - if (_controlFrames.Count > 0) + while (_controlFrames.Count > 0) { - return _controlFrames.Dequeue(); + PendingFrame frame = _controlFrames.Dequeue(); + if (frame.Completion.Task.IsCanceled) + { + continue; + } + + frame.Claimed = true; + return frame; } - if (_eventFrames.Count > 0) + while (_eventFrames.Count > 0) { - return _eventFrames.Dequeue(); + PendingFrame frame = _eventFrames.Dequeue(); + if (frame.Completion.Task.IsCanceled) + { + continue; + } + + frame.Claimed = true; + return frame; } return null; diff --git a/src/ZB.MOM.WW.MxGateway.Worker/Ipc/WorkerPipeSession.cs b/src/ZB.MOM.WW.MxGateway.Worker/Ipc/WorkerPipeSession.cs index b8fdd5d..e91442a 100644 --- a/src/ZB.MOM.WW.MxGateway.Worker/Ipc/WorkerPipeSession.cs +++ b/src/ZB.MOM.WW.MxGateway.Worker/Ipc/WorkerPipeSession.cs @@ -371,27 +371,57 @@ public sealed class WorkerPipeSession continue; } - foreach (WorkerEvent workerEvent in events) + // Submit the whole drained batch through the writer's batch entry point under one lock + // acquisition so the burst pays a single flush instead of one per event (WRK-25). 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, + // and intra-batch order is preserved. + WorkerEnvelope[] envelopes = new WorkerEnvelope[events.Count]; + for (int index = 0; index < events.Count; index++) { - // 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. - 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); - } + envelopes[index] = CreateEnvelope(events[index]); + } + + try + { + await _writer + .WriteBatchAsync(envelopes, WorkerFrameWritePriority.Event, cancellationToken) + .ConfigureAwait(false); + } + catch (WorkerFrameProtocolException exception) + when (exception.ErrorCode == WorkerFrameProtocolErrorCode.MessageTooLarge) + { + // A single oversized event surfaces from the batch's awaited completions; the death is + // still IPC-30's structured, event-naming fault. Map the rejection back to the first + // event in batch order whose envelope overshoots the negotiated maximum — the same + // frame the writer rejected first. + await FaultOnOversizedEventAsync( + FindOversizedEvent(events, envelopes), + exception, + cancellationToken) + .ConfigureAwait(false); } } } + private WorkerEvent FindOversizedEvent( + IReadOnlyList events, + WorkerEnvelope[] envelopes) + { + for (int index = 0; index < envelopes.Length; index++) + { + if (envelopes[index].CalculateSize() > _options.MaxMessageBytes) + { + return events[index]; + } + } + + // Unreachable in practice: WriteBatchAsync surfaced MessageTooLarge, so at least one envelope + // exceeded the negotiated maximum. Fall back to the first event so the fault still names a + // concrete event rather than throwing a second, less useful exception from the fault path. + return events[0]; + } + /// /// 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 @@ -1085,16 +1115,16 @@ public sealed class WorkerPipeSession return; } - if (!string.IsNullOrEmpty(snapshot.CurrentCommandCorrelationId) + if ((!string.IsNullOrEmpty(snapshot.CurrentCommandCorrelationId) || snapshot.StaCallInProgress) && staleFor <= _sessionOptions.HeartbeatStuckCeiling) { - // A command is in flight and we are still within the defensive - // suppression ceiling — the STA is busy executing it, not - // hung. The next MarkActivity() in StaRuntime.ProcessQueuedCommands - // will refresh LastActivityUtc once the command returns, at which - // point this branch stops being taken. The heartbeat already - // surfaces the in-flight correlation id so the gateway can apply - // its own per-command timeout if it considers the command too slow. + // A command is in flight, or an STA call outside the dispatcher (the alarm poll, WRK-27) is + // executing, and we are still within the defensive suppression ceiling — the STA is busy + // doing that work, not hung. The next MarkActivity() in StaRuntime.ProcessQueuedCommands + // will refresh LastActivityUtc once the work returns, at which point this branch stops + // being taken. The heartbeat already surfaces the in-flight correlation id so the gateway + // can apply its own per-command timeout if it considers the command too slow; a poll that + // blocks the STA past the ceiling still faults, which is the ceiling's contract. return; } diff --git a/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/MxAccessStaSession.cs b/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/MxAccessStaSession.cs index 44d2b80..51ccbab 100644 --- a/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/MxAccessStaSession.cs +++ b/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/MxAccessStaSession.cs @@ -24,6 +24,14 @@ public sealed class MxAccessStaSession : IWorkerRuntimeSession private CancellationTokenSource? alarmPollCts; private Task? alarmPollTask; private int? alarmConsumerThreadId; + + // True on the STA thread exactly around the alarm PollOnce COM call. The alarm poll runs outside + // the StaCommandDispatcher (so it does not inflate PendingCommandCount or perturb command dispatch + // ordering), which means CaptureHeartbeat would otherwise see no in-flight activity during a long + // poll and the watchdog would fault the session at the 15 s grace instead of the 75 s ceiling + // granted to dispatched commands. Surfacing the poll on the heartbeat closes that asymmetry + // (WRK-27). Volatile: written on the STA thread, read on the heartbeat thread. + private volatile bool staAlarmPollInProgress; private bool disposed; /// @@ -247,8 +255,20 @@ public sealed class MxAccessStaSession : IWorkerRuntimeSession await staRuntime.InvokeAsync( () => { - EnsureOnAlarmConsumerThread(); - handler.PollOnce(); + // Advertise the poll to the watchdog for exactly the span of the COM call + // (WRK-27): set on the STA thread immediately before the affinity check and + // PollOnce, clear in the finally so a heartbeat captured mid-poll reports + // StaCallInProgress and one captured after does not. + staAlarmPollInProgress = true; + try + { + EnsureOnAlarmConsumerThread(); + handler.PollOnce(); + } + finally + { + staAlarmPollInProgress = false; + } }, cancellationToken).ConfigureAwait(false); } @@ -377,7 +397,8 @@ public sealed class MxAccessStaSession : IWorkerRuntimeSession pendingCommandCount, (uint)eventQueue.Count, eventQueue.LastEventSequence, - currentCommandCorrelationId); + currentCommandCorrelationId, + staAlarmPollInProgress); } /// diff --git a/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/WorkerRuntimeHeartbeatSnapshot.cs b/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/WorkerRuntimeHeartbeatSnapshot.cs index 0852166..2b8e05f 100644 --- a/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/WorkerRuntimeHeartbeatSnapshot.cs +++ b/src/ZB.MOM.WW.MxGateway.Worker/MxAccess/WorkerRuntimeHeartbeatSnapshot.cs @@ -10,18 +10,26 @@ public sealed class WorkerRuntimeHeartbeatSnapshot /// Current depth of the worker event queue. /// Sequence number of the most recent event. /// Correlation ID of the in-flight command. + /// + /// True while an STA call outside the command dispatcher is executing (currently the alarm poll, + /// WRK-27). The watchdog treats this like an in-flight command: it suppresses the stale-STA fault + /// up to the stuck ceiling instead of the shorter grace, so a healthy-but-slow poll does not fault + /// a healthy session. Named generically so any future non-dispatcher STA work reuses it. + /// public WorkerRuntimeHeartbeatSnapshot( DateTimeOffset lastStaActivityUtc, uint pendingCommandCount, uint outboundEventQueueDepth, ulong lastEventSequence, - string currentCommandCorrelationId) + string currentCommandCorrelationId, + bool staCallInProgress = false) { LastStaActivityUtc = lastStaActivityUtc; PendingCommandCount = pendingCommandCount; OutboundEventQueueDepth = outboundEventQueueDepth; LastEventSequence = lastEventSequence; CurrentCommandCorrelationId = currentCommandCorrelationId ?? string.Empty; + StaCallInProgress = staCallInProgress; } /// Gets the last STA activity timestamp in UTC. @@ -38,4 +46,11 @@ public sealed class WorkerRuntimeHeartbeatSnapshot /// Gets the correlation ID of the in-flight command. public string CurrentCommandCorrelationId { get; } + + /// + /// Gets a value indicating whether an STA call outside the command dispatcher (the alarm poll) is + /// executing. When true the watchdog grants the poll the same grace-to-ceiling suppression as a + /// dispatched command (WRK-27). + /// + public bool StaCallInProgress { get; } } From 815e58d28b556ecff44f65e5b7e130450d31b46e Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Fri, 7 Aug 2026 07:52:44 -0400 Subject: [PATCH 2/2] docs(tracking): record WRK-22/24/25/27 + IPC-26 windev evidence (377 pass) --- archreview/2026-07-12/remediation/00-tracking.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/archreview/2026-07-12/remediation/00-tracking.md b/archreview/2026-07-12/remediation/00-tracking.md index 8c588d8..72d76f1 100644 --- a/archreview/2026-07-12/remediation/00-tracking.md +++ b/archreview/2026-07-12/remediation/00-tracking.md @@ -174,5 +174,5 @@ Sequence these together rather than piecemeal — several are one change set spa | 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.) | -| 2026-08-07 | **Worker-seam batch → `Done`: WRK-22 (mechanics for IPC-26), WRK-24, WRK-25, WRK-27** (branch `fix/wrk-22-25-seam`). **WRK-22/IPC-26**: `WorkerFrameWriter.PendingFrame` gained a `Claimed` field; a `WriteAsync`/`WriteBatchAsync` cancelled while waiting for the write lock tombstones its still-unclaimed frame (`TrySetCanceled` under `_gate`) and `DequeueNext` skips cancelled frames and marks the one it returns `Claimed`, so a cancelled write never reaches the wire — except the documented, by-design residual where a lock-holder claimed the frame first (mid-write, cannot be recalled; caller still observes cancellation). **WRK-25**: new `WriteBatchAsync(IReadOnlyList, priority, ct)` enqueues a whole batch under one `_gate` acquisition, takes the lock once, drains, then observes every completion (surfacing the first per-frame rejection); `RunEventDrainLoopAsync` now submits the drained event batch through it, so a burst of N events costs one flush not N — the WRK-12 coalescing now engages on the event hot path. IPC-30's oversized-event structured fault is preserved (`FindOversizedEvent` maps the batch rejection back to the offending event). **WRK-24**: `WorkerFrameProtocolOptions.MinNegotiableFrameBytes = 1024` (matches `GatewayOptionsValidator.MinimumMaxMessageBytes`); `AdoptNegotiatedMaxMessageBytes` now faults a below-floor negotiated value at the handshake, closing the [1024, 256 MiB] accepted range. **WRK-27**: alarm poll runs outside the dispatcher, so `MxAccessStaSession` sets a `volatile staAlarmPollInProgress` around the `PollOnce` COM call and surfaces it on the new `WorkerRuntimeHeartbeatSnapshot.StaCallInProgress`; `ReportWatchdogFaultIfNeededAsync` honors it alongside `CurrentCommandCorrelationId`, so a healthy-but-slow poll gets grace-to-ceiling suppression (not the 15 s grace) but still faults past the 75 s ceiling. Docs same commit: `docs/WorkerFrameProtocol.md` (accepted-range paragraph, flush-coalescing sentence flipped to coalesced-on-drain, cancellation-tombstone contract replacing the WRK-26 "pending" placeholder), `docs/MxAccessWorkerInstanceDesign.md` (watchdog alarm-poll paragraph). New tests: `WorkerFrameProtocolTests.{WriteAsync_CancelledWhileWaitingForLock_FrameIsNeverWritten, WriteAsync_CancelledEventFrame_DoesNotTrailShutdownAck, WriteBatchAsync_FlushesOnceAndPreservesOrder, WriteBatchAsync_ControlFrameQueuedDuringBatch_JumpsRemainingEvents, AdoptNegotiatedMaxMessageBytes_BelowFloor_ThrowsInvalidConfiguration}`, `WorkerPipeSessionTests.{Watchdog_StaCallInProgress_SuppressedUntilCeiling, EventBurst_DrainLoopCoalescesFlushes, Handshake_GatewayHelloWithTinyMaxFrameBytes_FaultsAtHandshake}`, `MxAccessStaSessionTests.CaptureHeartbeat_DuringAlarmPoll_ReportsStaCallInProgress` (+ `FakeRuntimeSession.EnqueueEvents` bulk helper, `staCallInProgress` snapshot ctor param). **IPC-26 → `Done`** (mechanics owned here). **Evidence** — macOS: `dotnet build src/ZB.MOM.WW.MxGateway.NonWindows.slnx` . windev: . | +| 2026-08-07 | **Worker-seam batch → `Done`: WRK-22 (mechanics for IPC-26), WRK-24, WRK-25, WRK-27** (branch `fix/wrk-22-25-seam`). **WRK-22/IPC-26**: `WorkerFrameWriter.PendingFrame` gained a `Claimed` field; a `WriteAsync`/`WriteBatchAsync` cancelled while waiting for the write lock tombstones its still-unclaimed frame (`TrySetCanceled` under `_gate`) and `DequeueNext` skips cancelled frames and marks the one it returns `Claimed`, so a cancelled write never reaches the wire — except the documented, by-design residual where a lock-holder claimed the frame first (mid-write, cannot be recalled; caller still observes cancellation). **WRK-25**: new `WriteBatchAsync(IReadOnlyList, priority, ct)` enqueues a whole batch under one `_gate` acquisition, takes the lock once, drains, then observes every completion (surfacing the first per-frame rejection); `RunEventDrainLoopAsync` now submits the drained event batch through it, so a burst of N events costs one flush not N — the WRK-12 coalescing now engages on the event hot path. IPC-30's oversized-event structured fault is preserved (`FindOversizedEvent` maps the batch rejection back to the offending event). **WRK-24**: `WorkerFrameProtocolOptions.MinNegotiableFrameBytes = 1024` (matches `GatewayOptionsValidator.MinimumMaxMessageBytes`); `AdoptNegotiatedMaxMessageBytes` now faults a below-floor negotiated value at the handshake, closing the [1024, 256 MiB] accepted range. **WRK-27**: alarm poll runs outside the dispatcher, so `MxAccessStaSession` sets a `volatile staAlarmPollInProgress` around the `PollOnce` COM call and surfaces it on the new `WorkerRuntimeHeartbeatSnapshot.StaCallInProgress`; `ReportWatchdogFaultIfNeededAsync` honors it alongside `CurrentCommandCorrelationId`, so a healthy-but-slow poll gets grace-to-ceiling suppression (not the 15 s grace) but still faults past the 75 s ceiling. Docs same commit: `docs/WorkerFrameProtocol.md` (accepted-range paragraph, flush-coalescing sentence flipped to coalesced-on-drain, cancellation-tombstone contract replacing the WRK-26 "pending" placeholder), `docs/MxAccessWorkerInstanceDesign.md` (watchdog alarm-poll paragraph). New tests: `WorkerFrameProtocolTests.{WriteAsync_CancelledWhileWaitingForLock_FrameIsNeverWritten, WriteAsync_CancelledEventFrame_DoesNotTrailShutdownAck, WriteBatchAsync_FlushesOnceAndPreservesOrder, WriteBatchAsync_ControlFrameQueuedDuringBatch_JumpsRemainingEvents, AdoptNegotiatedMaxMessageBytes_BelowFloor_ThrowsInvalidConfiguration}`, `WorkerPipeSessionTests.{Watchdog_StaCallInProgress_SuppressedUntilCeiling, EventBurst_DrainLoopCoalescesFlushes, Handshake_GatewayHelloWithTinyMaxFrameBytes_FaultsAtHandshake}`, `MxAccessStaSessionTests.CaptureHeartbeat_DuringAlarmPoll_ReportsStaCallInProgress` (+ `FakeRuntimeSession.EnqueueEvents` bulk helper, `staCallInProgress` snapshot ctor param). **IPC-26 → `Done`** (mechanics owned here). **Evidence** — macOS: `dotnet build src/ZB.MOM.WW.MxGateway.NonWindows.slnx` 0 warnings / 0 errors (worker excluded there; proves the shared/server side is intact). windev (`scripts/ci/windev-worker-ci.ps1 -Sha 8df35cd -Mode test`, 2026-08-07 07:45): x86 Worker build 0 warnings / 0 errors, `Worker.Tests` **377 passed / 0 failed / 11 skipped** (skips are the live-MXAccess/dev-rig opt-ins; +9 over the prior 368 = the nine new tests, all green), script exit 0. | | 2026-08-07 | **P1 doc-drift batch → `Done`: TST-27, WRK-26 (discharges IPC-29), CLI-42, CLI-43, IPC-28** (branch `fix/doc-drift-batch`). Doc-only; no source, proto, or test changes — cross-checked against HEAD in this worktree. **TST-27**: `docs/GatewayConfiguration.md`'s `ShowTagValues` row no longer says "Reserved" — it now states what `false` (default) does (`DashboardEventBroadcaster` blanks tag values from a deep-cloned `MxEvent` before the SignalR events-hub mirror, metadata still renders), the security relevance (the per-session hub ACL, SEC-25 roadmap item 12, still does not exist, so this redaction is the only thing between a low-trust Viewer and other sessions' tag values), and the honest scope limit (the flag does **not** cover `/browse`). **WRK-26** (discharges **IPC-29**): `docs/MxAccessWorkerInstanceDesign.md`'s "Outbound Queues" section rewritten from the stale five-level priority list to the two-class `Control`/`Event` scheduler actually shipped (`WorkerFrameWriter`/`WorkerFrameWritePriority.cs`), with the collapsed-decision rationale recorded, and the overflow paragraph rewritten to the implemented fail-fast (`WorkerFault` category `QueueOverflow` → fault frame written → `RunAsync` unwinds → generic `WorkerExitCode.UnexpectedFailure`, dedicated code still open). `docs/WorkerFrameProtocol.md` gained a new "Write Scheduling And Sequencing" section: the two priority classes, enqueue-then-contend/single-lock-holder-drains-all, write-time peek-stamp-commit sequencing, per-frame-rejection vs. stream-failure semantics, and flush coalescing — stated truthfully as landed (WRK-23's peek-stamp-commit is live at HEAD) or not (the drain loop still awaits each event `WriteAsync` individually, so WRK-25's N-events-one-flush batching has **not** landed and the section says so explicitly). Cancellation is deliberately **not** documented as a firm contract — a one-paragraph placeholder notes it is pending WRK-22, which has not landed (confirmed by reading `WorkerFrameWriter.cs`: no `Claimed`/tombstone machinery exists yet). `gateway.md:328-330` was cross-checked and left unchanged — its sequence prose (both sides stamp at write, per GWC-28) already reads true. **CLI-42**: `clients/rust/README.md` and `docs/ClientPackaging.md`'s Rust section now document the vendored proto layout matching `clients/rust/build.rs` exactly — repo-path-first resolution (`../../src/ZB.MOM.WW.MxGateway.Contracts/Protos`) falling back to `clients/rust/protos/` when the canonical path is absent (published-tarball case), the same-commit refresh rule enforced by `scripts/check-codegen.ps1` Check 3, and why `cargo package`/`cargo publish` run without `--no-verify` (matches `scripts/pack-clients.ps1:190-192`). **CLI-43**: `docs/style-guides/JavaStyleGuide.md` line 8 now says "Target Java 17 (the Ignition 8.3 baseline...)" mirroring the CLI-12 wording, matching the shipped `clients/java/build.gradle` toolchain-17 build. **IPC-28**: `docs/Grpc.md`'s exception-mapping prose gained `CommandTooLarge` → `ResourceExhausted` (verified against the live `switch` in `Grpc/MxAccessGatewayService.cs:950-960`), and the `Invoke` section gained one sentence on the oversized-payload path (`WorkerClient.InvokeAsync` rejects at the enqueue boundary per-correlation, session not faulted — verified against `WorkerClient.cs:220-234`), cross-referencing the headroom rule already documented in `docs/GatewayConfiguration.md:120-129`. Did not touch the DrainEvents-truncation row or the proto/`Generated/` trees — those belong to a parallel codegen task per the handoff note. **Source files cross-read for accuracy** (no edits): `src/ZB.MOM.WW.MxGateway.Worker/Ipc/WorkerFrameWriter.cs`, `.../WorkerFrameWritePriority.cs`, `.../WorkerPipeSession.cs` (confirmed two-class scheduler, WRK-21/23/28/30 landed, WRK-25/WRK-22 not landed), `src/ZB.MOM.WW.MxGateway.Worker/WorkerApplication.cs` (exit-code mapping), `src/ZB.MOM.WW.MxGateway.Worker/MxAccess/MxAccessEventQueue.cs` (overflow fault path), `src/ZB.MOM.WW.MxGateway.Server/Dashboard/Hubs/DashboardEventBroadcaster.cs` + `Configuration/DashboardOptions.cs` + `docs/GatewayDashboardDesign.md:170` (ShowTagValues), `src/ZB.MOM.WW.MxGateway.Server/Grpc/MxAccessGatewayService.cs:940-963` + `Workers/WorkerClient.cs:205-244` + `Workers/WorkerClientErrorCode.cs` (CommandTooLarge mapping), `clients/rust/build.rs`, `clients/rust/Cargo.toml`, `scripts/check-codegen.ps1`, `scripts/pack-clients.ps1` (Rust vendoring), `gateway.md:326-360` (sequence-prose cross-check). Verification (greps, doc-only — no build required): `grep -n 'Reserved' docs/GatewayConfiguration.md` no longer matches the `ShowTagValues` row; `grep -n 'faults' docs/MxAccessWorkerInstanceDesign.md` shows no remaining five-level list; `grep -n 'scheduling' docs/WorkerFrameProtocol.md` finds the new section; `grep -rn 'Java 21' docs/style-guides/` empty; `grep -i vendored docs/ClientPackaging.md clients/rust/README.md` non-empty in both; `grep -n 'CommandTooLarge' docs/Grpc.md` shows the mapping. |