diff --git a/archreview/2026-07-12/remediation/00-tracking.md b/archreview/2026-07-12/remediation/00-tracking.md index cb94a4b..105884a 100644 --- a/archreview/2026-07-12/remediation/00-tracking.md +++ b/archreview/2026-07-12/remediation/00-tracking.md @@ -59,7 +59,7 @@ Full design + implementation for each row lives in the linked domain doc under i | ID | Sev | Tier | Eff | Dep | Status | Title | |---|---|:-:|:-:|---|---|---| -| GWC-24 | Medium | P1 | M | GWC-21 (coord, old tracker) | Not started | Unbounded event staging channel: sustained slow drain grows memory silently and invisibly | +| GWC-24 | Medium | P1 | M | GWC-21 (coord, old tracker) | Done | Unbounded event staging channel: sustained slow drain grows memory silently and invisibly | | GWC-25 | Medium | P0 | S | CLI-35/36 (coord) | Not started | Empty-ring ReplayGap sentinel carries `oldest_available_sequence = 0` | | GWC-26 | Low | P2 | M | GWC-27 | Not started | Alarm monitor attaches its subscriber after SubscribeAlarms; window transitions bypass the feed | | GWC-27 | Low | P2 | S | GWC-26 | Not started | `AttachInternalEventSubscriber` bypasses the readiness gate; premature attach poisons the distributor | @@ -144,7 +144,7 @@ Sequence these together rather than piecemeal — several are one change set spa - **Windows-tier automation (P1):** TST-25 + TST-26 (same commit). Unlocks old TST-05/TST-24 and provides CI evidence for every windev-verified cluster above; until it lands, record windev runs in this tracker's change log. - **Client conformance + release train (P1):** CLI-37 + CLI-38 co-land (one conformance commit; closes old CLI-08), then CLI-45, with CLI-40/41 fixtures as follow-ups; **CLI-39 lands last** so published 0.2.0 carries the conformant behavior. Shared fixtures under `clients/proto/fixtures/behavior/`; update CrossLanguageSmokeMatrix.md/ClientLibrariesDesign.md same-commit. Do not republish regenerated bindings (IPC-25) before CLI-39 resolves. - **Doc-drift batch (P1):** TST-27 + WRK-26 (discharges IPC-29) + CLI-42 + IPC-28 + SEC-35's doc note — one sweep commit is fine. -- **Backpressure follow-on (P1):** GWC-24, coordinating with still-open old GWC-21 (`EventChannelFullModeTimeout` configurability). +- **Backpressure follow-on (P1):** GWC-24 — `Done` 2026-08-07. Landed without a new config key (the staging bound is derived as `2 × EventChannelCapacity`), so still-open old GWC-21 (`EventChannelFullModeTimeout` configurability) remains the natural companion knob but was not a blocker. ## Old-tracker actions ([`../../remediation/00-tracking.md`](../../remediation/00-tracking.md)) @@ -162,3 +162,4 @@ Sequence these together rather than piecemeal — several are one change set spa | 2026-07-13 | Ran the TST-25 acceptance checks (scripts/ci/README.md) — they caught **two real CI defects, both fixed** on `fix/tst-25-ci-key-log-leak`: (1) **CI SSH key leaked in cleartext** in the `windows-x86` step env echo (Gitea's line-oriented masker missed the multiline PEM) — rotated the CI key on windev (old pubkey revoked), stored the key **base64-encoded** so the masker redacts it to `***` (confirmed on run #38), taught `run-windev-ci.sh` to decode, dropped the redundant public known-hosts secret from the job env; (2) **bootstrap lock race** — `run-windev-ci.sh`'s pre-hand-off `git fetch`/`checkout` ran outside the worktree lock, so concurrent runs collided on `.git/index.lock`; the bootstrap now holds the lock (ps1 re-uses it via `MXGW_CI_LOCK_HELD`), retest confirmed clean serialization. Also **deflaked** `SessionManagerTests` fail-fast timing assertions (absolute `<100ms` wall-clock bound flaked under CI load; now anchored to the configured timeout / dropped for the zero-timeout case). Checks passed: unreachable-host fast-fail (exit 255/15s), deliberate-red propagation (Worker.Tests failure → exit 1), lock concurrency (2nd run waits), no-key-in-logs (masked). Merge target `df7e20d` verified GREEN via the local windev path (Worker build + 356 tests); merged to `main` `19cbf7b`. Check 6 (forced-failure nightly issue): issue endpoint+token proven live at bring-up (#124); in-CI forced-failure probe abandoned to shared-runner congestion (residual `if: failure()` gating is standard Actions). | | 2026-07-13 | New finding **TST-30** (`Low`/`P2`) added — surfaced during TST-25 acceptance: CI runs on a single shared `gitea-runner` (`maxParallel=1`, co-located `10.100.0.35`) interleaved with `dohertj2/lmxopcua`, and Gitea 1.26 exposes no run cancel/delete, so queue latency is unbounded under cross-repo contention and the runner is a single point of failure. Design: add a second/labelled runner + document the no-cancel reality and the `run-windev-ci.sh` bypass. Roll-ups updated (Testing Low 2→3, total 47→48; P2 9→10). | | 2026-08-07 | **TST-29 → `Done`:** migrated the Phase-5 (orphan-worker reattach) deferred-not-planned governance record and the settled Phase-4 Viewer-default decision from `oldtasks.md` into a new "Session-Resilience Epic Scope" entry in `docs/DesignDecisions.md`; repointed CLAUDE.md and `stillpending.md:7,165` from `oldtasks.md` to `docs/DesignDecisions.md` / `docs/plans/2026-06-15-session-resilience.md.tasks.json`; `git rm oldtasks.md`. The five untracked root docs-review artifacts (`MxAccessGateway-docs-{issues,fixed,final}.md`, `MxGatewayClient-docs-{issues,fixed}.md`) were absent from this worktree — delete from the main working tree separately. | +| 2026-08-07 | **GWC-24 → `Done`** (branch `fix/gwc-24-staging-bound`). `WorkerClient._eventStaging` is now `Channel.CreateBounded` at `2 × EventChannelCapacity` (`Wait`, single reader/writer, no sync continuations); a rejected staging `TryWrite` faults the client `ProtocolViolation` with `QueueOverflow("worker-event-staging")` unless `IsTerminalState()` (shutdown stays a silent drop), so a consumer draining slower than its worker produces dies at a fixed ceiling instead of growing gateway memory. Queue-depth accounting moved from `EnqueueWorkerEventAsync` to `StageWorkerEvent`, so the single gauge reports staged + queued; the timed-write fault (`EventChannelFullModeTimeout` / `QueueOverflow("worker-events")`) is unchanged and still catches the full-stall case first. No new config key — total gateway-side buffering is `3 × MxGateway:Events:QueueCapacity`, derived; coordination with still-open old **GWC-21** (`EventChannelFullModeTimeout` configurability) remains open and was not blocked on. Docs same commit: `GatewayProcessDesign.md` (two overflow faults), `MxAccessWorkerInstanceDesign.md`, `GatewayConfiguration.md`, `Metrics.md`. Tests: `WorkerClientTests.StagingChannelOverflowFaultsWorkerWithoutWaitingForFullModeTimeout` and `.WorkerEventQueueDepthGaugeCountsStagedEvents`; `WorkerClientTests` 22/22 green, `NonWindows.slnx` builds with 0 warnings. | diff --git a/archreview/2026-07-12/remediation/10-gateway-core.md b/archreview/2026-07-12/remediation/10-gateway-core.md index 7d98ccb..151e9e0 100644 --- a/archreview/2026-07-12/remediation/10-gateway-core.md +++ b/archreview/2026-07-12/remediation/10-gateway-core.md @@ -8,7 +8,7 @@ This document turns the 2026-07-12 re-review's **new** Gateway Server Core findi | ID | Sev | Tier | Eff | Dep | Status | Title | |----|-----|------|-----|-----|--------|-------| -| GWC-24 | Medium | P1 | M | GWC-21 (coord) | Not started | Unbounded event staging channel: sustained slow drain grows memory silently and invisibly | +| GWC-24 | Medium | P1 | M | GWC-21 (coord) | Done | Unbounded event staging channel: sustained slow drain grows memory silently and invisibly | | GWC-25 | Medium | P0 | S | CLI-35/36 (coord) | Not started | Empty-ring ReplayGap sentinel carries `oldest_available_sequence = 0`, dead-streaming a compliant client | | GWC-26 | Low | P2 | M | GWC-27 | Not started | Alarm monitor attaches its subscriber after SubscribeAlarms; window transitions bypass the feed, missed Acknowledge never repaired | | GWC-27 | Low | P2 | S | GWC-26 | Not started | `AttachInternalEventSubscriber` bypasses the readiness gate; premature attach poisons the distributor permanently | @@ -20,7 +20,7 @@ Dependency notes: GWC-26 and GWC-27 both change the internal-subscriber attach p --- -## GWC-24 — Unbounded event staging channel: sustained slow drain grows memory silently and invisibly `Medium` · `P1` +## GWC-24 — Unbounded event staging channel: sustained slow drain grows memory silently and invisibly `Medium` · `P1` · **Done (2026-08-07)** **Finding.** The GWC-04 remediation decoupled the read loop from event backpressure by staging events into `_eventStaging`, an **unbounded** channel (`Workers/WorkerClient.cs:93-100`). `StageWorkerEvent`'s `TryWrite` therefore always succeeds (`:565-573`), and the sustained-overflow `ProtocolViolation` fault fires only when a *single* timed `WriteAsync` against the bounded `_events` exceeds `EventChannelFullModeTimeout` (default 5 s, `:610-647`). The queue-depth gauge counts only `_events` — `_eventQueueDepth` is incremented in `EnqueueWorkerEventAsync` (`:616`, `:627`) and decremented in `ReadEventsCoreAsync` (`:289`), so staged-but-unqueued events are invisible to `SetWorkerEventQueueDepth`. The field comment (`:28-33`) claims staging "only fills during the bounded EventChannelFullModeTimeout window", which is true only for a full consumer stall, not for a consumer that drains slower than the worker produces while each individual write still completes inside the window. diff --git a/docs/GatewayConfiguration.md b/docs/GatewayConfiguration.md index e26406a..cb293c2 100644 --- a/docs/GatewayConfiguration.md +++ b/docs/GatewayConfiguration.md @@ -149,17 +149,19 @@ All numeric session options must be greater than zero. | Option | Default | Description | |--------|---------|-------------| -| `MxGateway:Events:QueueCapacity` | `10000` | Capacity for bounded per-session event queues used by the gateway worker event channel and the public gRPC event stream queue. | +| `MxGateway:Events:QueueCapacity` | `10000` | Capacity for bounded per-session event queues used by the gateway worker event channel and the public gRPC event stream queue. Gateway-side buffering per session is at most `3 ×` this value: the bounded worker event channel plus the read loop's staging channel, which is bounded at `2 ×` it. Overflow of either bound faults the session with `ProtocolViolation` and kills its worker. | | `MxGateway:Events:BackpressurePolicy` | `FailFast` | Per-subscriber event backpressure behavior when a subscriber's bounded event channel overflows. Overflow is isolated to the offending subscriber: it is always disconnected with an `EventQueueOverflow` fault while the session pump and other subscribers keep running. `FailFast` additionally faults the whole session only in the legacy single-subscriber case (the current default mode); with multiple subscribers it degrades to a per-subscriber disconnect so one slow consumer never faults a shared session. `DisconnectSubscriber` disconnects only the slow subscriber in all cases. | | `MxGateway:Events:ReplayBufferCapacity` | `1024` | Maximum number of events retained per session in the replay ring buffer, used to re-deliver events a returning subscriber missed (reconnect/reattach). The oldest retained event is evicted once this count is exceeded. `0` disables replay retention. | | `MxGateway:Events:ReplayRetentionSeconds` | `300` | Maximum age, in seconds, of an event retained in the replay ring buffer. Entries older than this are evicted regardless of capacity. `0` disables age-based eviction. | | `MxGateway:Events:MaxSparseArrayLength` | `1000000` | Maximum `total_length` a sparse-array write (`MxSparseArray`) may declare. A write above this cap is rejected with `InvalidArgument` before the full array is materialized, guarding against a single write forcing a multi-GB allocation. Must be between `1` and `Array.MaxLength`. | `QueueCapacity` must be greater than zero; it bounds each per-subscriber event -channel fed by the session's single event pump. A slow subscriber overflows only -its own channel and is always disconnected with an `EventQueueOverflow` fault -rather than silently dropping MXAccess events — the pump, the session, and other -subscribers are unaffected. With `FailFast` in the single-subscriber case (the +channel fed by the session's single event pump, and — at `2 ×` — the worker +read loop's event staging channel, so a consumer that drains slower than its +worker produces faults the session at a fixed ceiling instead of growing gateway +memory (GWC-24). A slow subscriber overflows only its own channel and is always +disconnected with an `EventQueueOverflow` fault rather than silently dropping +MXAccess events — the pump, the session, and other subscribers are unaffected. With `FailFast` in the single-subscriber case (the default mode), that overflow additionally faults the whole session; with multiple subscribers `FailFast` degrades to a per-subscriber disconnect, matching `DisconnectSubscriber`, so one slow consumer cannot fault a session shared by diff --git a/docs/GatewayProcessDesign.md b/docs/GatewayProcessDesign.md index 0e5da17..4a1d3af 100644 --- a/docs/GatewayProcessDesign.md +++ b/docs/GatewayProcessDesign.md @@ -483,7 +483,7 @@ Internally it owns: - write loop, - event write loop, - outbound command/control channel serialized by the write loop, -- unbounded event staging channel drained by the event write loop, +- bounded event staging channel drained by the event write loop, - bounded inbound event channel, - pending command dictionary keyed by correlation id, - heartbeat monitor, @@ -513,16 +513,36 @@ If the pipe closes while the session is not closing, fault the session. The read loop never awaits event enqueue. Events are staged to the event write loop with a non-blocking write, so a full inbound event channel (a slow or absent `StreamEvents` consumer) cannot stall the read loop behind an event and -delay a command reply or heartbeat (GWC-04). The bounded-channel backpressure -window (`EventChannelFullModeTimeout`) and the sustained-overflow fault are -applied by the event write loop, not the read loop. +delay a command reply or heartbeat (GWC-04). The timed backpressure window +(`EventChannelFullModeTimeout`) is applied by the event write loop, not the read +loop; the read loop's only event-path fault is the staging-bound rejection +described below, which uses the non-blocking `SetFaulted`. -### Event write loop +### Event write loop and the two overflow faults -The event write loop drains the staging channel and performs the timed write -into the bounded inbound event channel. When the inbound channel stays full past -`EventChannelFullModeTimeout` it faults the session (`ProtocolViolation`) — the -same overflow contract as before, moved off the read loop. +The staging channel is bounded at `2 ×` the inbound event channel capacity, so +gateway-side buffering per session is at most `3 × MxGateway:Events:QueueCapacity` +— the fault fires as soon as *staging* is full, which is anywhere between `2 ×` +and `3 ×` depending on how much the event writer has already drained. +An unbounded staging channel would let a consumer that drains slower than the +worker produces grow gateway memory without limit and without any fault or +metric, because each individual timed write still completed inside the window +(GWC-24). Two distinct faults now bound the event path, both `ProtocolViolation` +and both killing the worker: + +- **Full stall** — the event write loop's timed write into the bounded inbound + channel stays blocked past `EventChannelFullModeTimeout`. Recorded as + `QueueOverflow("worker-events")`. Catches a consumer that stopped entirely, + earlier than the staging bound would. +- **Sustained slow drain** — the read loop's staging `TryWrite` is rejected + because staging is full at its `2 ×` bound, meaning the writer has been + saturated for as long as the worker took to emit that many further events. + Recorded as `QueueOverflow("worker-event-staging")`. A rejected `TryWrite` + during shutdown (the staging channel is completed) stays a silent drop. + +The worker event queue-depth gauge (`mxgateway.events.worker_queue.depth`) is +incremented at staging and decremented at consumer read, so it reports total +undelivered events across both channels rather than only the inbound channel. ### Write loop diff --git a/docs/Metrics.md b/docs/Metrics.md index 1ffea39..191c7a8 100644 --- a/docs/Metrics.md +++ b/docs/Metrics.md @@ -72,7 +72,7 @@ Observable gauges are pull-based; the `Meter` invokes the supplied callback when |------------|--------------|-------------| | `mxgateway.sessions.open` | `_openSessions` | Currently open sessions tracked by `SessionManager`. | | `mxgateway.workers.running` | `_workersRunning` | Worker clients in a running state. | -| `mxgateway.events.worker_queue.depth` | `_workerEventQueueDepth` | Last reported depth of the worker-side event queue. | +| `mxgateway.events.worker_queue.depth` | `_workerEventQueueDepth` | Undelivered worker events held by `WorkerClient` — staged *and* queued (GWC-24). Incremented when the read loop stages an event, decremented when the consumer reads it, so a backlog stuck in the staging channel is visible rather than invisible. | | `mxgateway.events.grpc_stream_queue.depth` | `_eventStreamBacklogSources` (summed on demand) | Live backlog buffered across every active `EventStreamService` subscriber, summed from the subscribers' channel `Count` at collection time. | ## Snapshot Shape @@ -146,8 +146,9 @@ _metrics.RemoveSessionEvents(session.SessionId); - `RecordWorkerStoppedOnce` calls `WorkerStopped(reason)` exactly once per worker, guarding against double-counting on simultaneous fault and exit signals. - `WorkerKilled(reason)` when the client forcibly terminates the worker. - `HeartbeatFailed(SessionId)` per missed heartbeat. -- `SetWorkerEventQueueDepth(queueDepth)` after each event ingest. +- `SetWorkerEventQueueDepth(queueDepth)` when the read loop stages an event and when the consumer reads one, so the gauge tracks staged + queued events. - `EventReceived(SessionId, workerEvent.Event.Family.ToString())` for each worker event. +- `QueueOverflow("worker-events")` when the timed write into the bounded consumer channel exceeds `EventChannelFullModeTimeout`, and `QueueOverflow("worker-event-staging")` when the staging channel is full at its `2 × EventChannelCapacity` bound. The two labels distinguish a stalled consumer from one that merely drains too slowly; both fault the session with `ProtocolViolation`. ### Worker process launcher diff --git a/docs/MxAccessWorkerInstanceDesign.md b/docs/MxAccessWorkerInstanceDesign.md index 936601d..558831a 100644 --- a/docs/MxAccessWorkerInstanceDesign.md +++ b/docs/MxAccessWorkerInstanceDesign.md @@ -623,6 +623,18 @@ queue fills: Production coalescing may be added later, but it must be explicit and tested. Do not drop or coalesce events in v1. +The gateway side of the event path is bounded to match. `WorkerClient` buffers +inbound events in a bounded consumer channel plus a staging channel bounded at +`2 ×` that capacity, so a session holds at most three times +`MxGateway:Events:QueueCapacity` undelivered events before it faults with +`ProtocolViolation` and kills this worker. Two faults cover the two failure +shapes — a consumer that stops entirely (the timed write past +`EventChannelFullModeTimeout`, metric `QueueOverflow("worker-events")`) and a +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. + ## Heartbeat And Watchdog `WorkerPipeSession` starts the heartbeat loop after the gateway validates diff --git a/src/ZB.MOM.WW.MxGateway.Server/Workers/WorkerClient.cs b/src/ZB.MOM.WW.MxGateway.Server/Workers/WorkerClient.cs index 841aca1..207574e 100644 --- a/src/ZB.MOM.WW.MxGateway.Server/Workers/WorkerClient.cs +++ b/src/ZB.MOM.WW.MxGateway.Server/Workers/WorkerClient.cs @@ -27,10 +27,16 @@ public sealed class WorkerClient : IWorkerClient // Staging hand-off between the read loop and the dedicated event writer. The read loop writes // here with a non-blocking TryWrite so a full consumer channel (_events) can never stall the read - // loop behind an event — replies and heartbeats keep flowing. Unbounded, but only fills - // during the bounded EventChannelFullModeTimeout window before EventWriteLoopAsync faults on a - // sustained backlog, after which the read loop stops. + // loop behind an event — replies and heartbeats keep flowing. Bounded at 2 × EventChannelCapacity + // (GWC-24): an unbounded staging channel let a consumer that drains slower than the worker + // produces grow gateway memory without limit and without a fault, because each individual timed + // write into _events still completed inside EventChannelFullModeTimeout. A rejected TryWrite here + // is the sustained-slow-drain signal and faults the client immediately (ProtocolViolation), which + // is the fail-fast backpressure policy in docs/DesignDecisions.md. Total gateway-side buffering + // per session is therefore 3 × MxGateway:Events:QueueCapacity, and the single _eventQueueDepth + // gauge covers staged + queued events so the whole backlog is observable. private readonly Channel _eventStaging; + private readonly int _eventStagingCapacity; private readonly ConcurrentDictionary _pendingCommands = new(StringComparer.Ordinal); private readonly SemaphoreSlim _pendingCommandSlots; private readonly CancellationTokenSource _stopCts = new(); @@ -90,12 +96,17 @@ public sealed class WorkerClient : IWorkerClient FullMode = BoundedChannelFullMode.Wait, AllowSynchronousContinuations = false, }); - _eventStaging = Channel.CreateUnbounded( - new UnboundedChannelOptions + _eventStagingCapacity = checked(2 * _options.EventChannelCapacity); + _eventStaging = Channel.CreateBounded( + new BoundedChannelOptions(_eventStagingCapacity) { // The read loop is the only writer; EventWriteLoopAsync is the only reader. SingleReader = true, SingleWriter = true, + // Wait (not Drop*) so the read loop's non-blocking TryWrite returns false exactly + // when the bound is reached — the same Wait+TryWrite overflow-detection idiom the + // session event distributor uses. The read loop never awaits this channel. + FullMode = BoundedChannelFullMode.Wait, AllowSynchronousContinuations = false, }); _lastHeartbeatAt = _timeProvider.GetUtcNow(); @@ -555,11 +566,16 @@ public sealed class WorkerClient : IWorkerClient /// /// Hands a received worker event to the dedicated event writer without blocking the read loop. - /// The staging channel is unbounded and this is the only writer, so TryWrite always - /// succeeds unless the channel has been completed during shutdown — in which case the event is - /// safely dropped because the client is closing. Backpressure and the sustained-overflow fault - /// are applied by against the bounded consumer channel, - /// off the read loop. + /// TryWrite is non-blocking, so the read loop keeps dispatching replies, heartbeats and + /// faults regardless of how backed up the event path is. It returns false in two cases: + /// the staging channel has been completed during shutdown (the event is safely dropped because + /// the client is already terminal), or the channel is full at its + /// 2 × bound. The latter means the + /// consumer has been draining slower than the worker produces for the whole time it took the + /// worker to emit that many further events, so the client is faulted immediately (GWC-24) — + /// is non-blocking, so the read loop still never awaits here. The + /// complementary full-stall case (a consumer that stops entirely) is caught earlier by the + /// timed write in . /// /// The event received from the worker. private void StageWorkerEvent(WorkerEvent workerEvent) @@ -569,7 +585,30 @@ public sealed class WorkerClient : IWorkerClient _metrics?.EventReceived(SessionId, workerEvent.Event.Family.ToString()); } - _eventStaging.Writer.TryWrite(workerEvent); + if (_eventStaging.Writer.TryWrite(workerEvent)) + { + // Counted here rather than at the _events write so the single gauge reports total + // undelivered events (staged + queued). ReadEventsCoreAsync decrements on consumer read. + int queueDepth = Interlocked.Increment(ref _eventQueueDepth); + _metrics?.SetWorkerEventQueueDepth(queueDepth); + return; + } + + if (IsTerminalState()) + { + // Shutdown completed the staging channel; dropping the event is the documented behavior. + return; + } + + _metrics?.QueueOverflow("worker-event-staging"); + int depthAtOverflow = Volatile.Read(ref _eventQueueDepth); + SetFaulted( + WorkerClientErrorCode.ProtocolViolation, + $"Worker event staging channel is full at its {_eventStagingCapacity}-event bound " + + $"(2 x EventChannelCapacity {_options.EventChannelCapacity}); undelivered depth is " + + $"{depthAtOverflow}. The event consumer is draining slower than the worker produces. " + + $"Attach or unblock the StreamEvents consumer or raise MxGateway:Events:QueueCapacity.", + null); } /// @@ -603,7 +642,11 @@ public sealed class WorkerClient : IWorkerClient /// missed slot even though the wait-mode channel would have absorbed /// the burst. The diagnostic now names the capacity, current depth, and /// the actionable fix (attach StreamEvents or raise - /// MxGateway:Events:QueueCapacity). + /// MxGateway:Events:QueueCapacity). This is the full-stall half of + /// the backpressure policy; a consumer that merely drains too slowly is + /// caught by the staging bound in (GWC-24). + /// Queue depth is not counted here — the event was already counted when it + /// was staged, so moving it between the two channels changes nothing. /// /// The event to enqueue. /// Cancellation token. @@ -613,8 +656,6 @@ public sealed class WorkerClient : IWorkerClient { if (_events.Writer.TryWrite(workerEvent)) { - int queueDepth = Interlocked.Increment(ref _eventQueueDepth); - _metrics?.SetWorkerEventQueueDepth(queueDepth); return; } @@ -624,8 +665,6 @@ public sealed class WorkerClient : IWorkerClient try { await _events.Writer.WriteAsync(workerEvent, fullModeCts.Token).ConfigureAwait(false); - int queueDepth = Interlocked.Increment(ref _eventQueueDepth); - _metrics?.SetWorkerEventQueueDepth(queueDepth); return; } catch (OperationCanceledException) when (!cancellationToken.IsCancellationRequested) @@ -641,7 +680,8 @@ public sealed class WorkerClient : IWorkerClient WorkerClientErrorCode.ProtocolViolation, $"Worker event channel rejected an event after waiting " + $"{_options.EventChannelFullModeTimeout.TotalMilliseconds:F0} ms; " - + $"channel depth is {depthAtOverflow} of {_options.EventChannelCapacity} capacity. " + + $"undelivered depth is {depthAtOverflow} against a consumer channel " + + $"of {_options.EventChannelCapacity} capacity. " + $"Attach a StreamEvents consumer or raise MxGateway:Events:QueueCapacity.", null); } diff --git a/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Workers/WorkerClientTests.cs b/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Workers/WorkerClientTests.cs index c4b112e..53a0867 100644 --- a/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Workers/WorkerClientTests.cs +++ b/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Workers/WorkerClientTests.cs @@ -650,6 +650,147 @@ public sealed class WorkerClientTests Assert.Contains("MxGateway:Events:QueueCapacity", fault.Message); } + /// + /// GWC-24: the staging channel between the read loop and the event writer is bounded at + /// 2 × , so a consumer that drains + /// slower than the worker produces faults the client at the bound instead of growing + /// gateway memory silently. The full-mode timeout here is far longer than the test could + /// ever wait, which proves the fault came from the staging bound and not from the timed + /// write in EnqueueWorkerEventAsync. A command reply interleaved before the fault + /// must still complete: the read loop never blocks behind events (the GWC-04 guarantee). + /// + /// A task that represents the asynchronous operation. + [Fact] + public async Task StagingChannelOverflowFaultsWorkerWithoutWaitingForFullModeTimeout() + { + const int capacity = 4; + const int stagingBound = 2 * capacity; + using GatewayMetrics metrics = new(); + await using PipePair pipePair = await PipePair.CreateAsync(); + await using WorkerClient client = CreateClient( + pipePair, + new WorkerClientOptions + { + EventChannelCapacity = capacity, + // Five minutes: the timed-write fault physically cannot be the trigger. + EventChannelFullModeTimeout = TimeSpan.FromMinutes(5), + HeartbeatGrace = TimeSpan.FromSeconds(30), + HeartbeatCheckInterval = TimeSpan.FromSeconds(30), + }, + metrics: metrics); + await CompleteHandshakeAsync(client, pipePair); + + // The staging channel alone holds 2 × capacity, and the event writer only ever removes + // from it, so this batch cannot fault however the writer happens to be scheduled. Waiting + // on the gauge (rather than a delay) proves the read loop consumed every one of them. + ulong sequence = 1; + for (; sequence <= (ulong)stagingBound; sequence++) + { + await pipePair.WorkerWriter.WriteAsync( + CreateEventEnvelope(sequence, MxEventFamily.OnDataChange)); + } + + await WaitUntilAsync( + () => metrics.GetSnapshot().WorkerEventQueueDepth == stagingBound, + TestTimeout); + Assert.Equal(WorkerClientState.Ready, client.State); + + // The event path is backed up with no consumer attached, yet a command still round-trips. + Task invokeTask = client.InvokeAsync( + CreateCommand(MxCommandKind.Ping), + TestTimeout, + CancellationToken.None); + WorkerEnvelope commandEnvelope = await pipePair.WorkerReader.ReadAsync().AsTask().WaitAsync(TestTimeout); + await pipePair.WorkerWriter.WriteAsync( + CreateCommandReplyEnvelope(commandEnvelope.CorrelationId, MxCommandKind.Ping)); + WorkerCommandReply reply = await invokeTask.WaitAsync(TestTimeout); + Assert.Equal(MxCommandKind.Ping, reply.Reply.Kind); + Assert.Equal(WorkerClientState.Ready, client.State); + + // The absolute ceiling is capacity (consumer channel) + 1 (in flight in the blocked event + // writer) + 2 × capacity (staging). Push comfortably past it. The events are tiny, so the + // ones the stopped read loop never drains stay in the OS pipe buffer instead of blocking. + for (int extra = 0; extra < 3 * capacity; extra++, sequence++) + { + await pipePair.WorkerWriter.WriteAsync( + CreateEventEnvelope(sequence, MxEventFamily.OnDataChange)); + } + + await WaitUntilAsync(() => client.State == WorkerClientState.Faulted, TestTimeout); + Assert.Equal(WorkerClientState.Faulted, client.State); + + using CancellationTokenSource drainTimeout = new(TestTimeout); + WorkerClientException fault = await Assert.ThrowsAsync(async () => + { + await foreach (WorkerEvent _ in client.ReadEventsAsync(drainTimeout.Token)) + { + } + }); + + Assert.Equal(WorkerClientErrorCode.ProtocolViolation, fault.ErrorCode); + Assert.Contains("staging", fault.Message, StringComparison.OrdinalIgnoreCase); + Assert.Contains($"{2 * capacity}", fault.Message, StringComparison.Ordinal); + Assert.Contains("StreamEvents", fault.Message, StringComparison.Ordinal); + Assert.Contains("MxGateway:Events:QueueCapacity", fault.Message, StringComparison.Ordinal); + + // The timed-write diagnostic must not be what fired. + Assert.DoesNotContain("Worker event channel rejected", fault.Message, StringComparison.Ordinal); + } + + /// + /// GWC-24: the worker event queue-depth gauge counts staged *and* queued events, so a + /// backlog held in the staging channel is visible rather than invisible. Depth is + /// incremented at staging and decremented when the consumer reads, so the single counter + /// reports total undelivered events and returns to zero once drained. + /// + /// A task that represents the asynchronous operation. + [Fact] + public async Task WorkerEventQueueDepthGaugeCountsStagedEvents() + { + const int capacity = 4; + const int eventCount = 8; + using GatewayMetrics metrics = new(); + await using PipePair pipePair = await PipePair.CreateAsync(); + await using WorkerClient client = CreateClient( + pipePair, + new WorkerClientOptions + { + EventChannelCapacity = capacity, + EventChannelFullModeTimeout = TimeSpan.FromMinutes(5), + HeartbeatGrace = TimeSpan.FromSeconds(30), + HeartbeatCheckInterval = TimeSpan.FromSeconds(30), + }, + metrics: metrics); + await CompleteHandshakeAsync(client, pipePair); + + // Above EventChannelCapacity but below the 2× staging bound: no consumer, no fault. + for (ulong sequence = 1; sequence <= eventCount; sequence++) + { + await pipePair.WorkerWriter.WriteAsync( + CreateEventEnvelope(sequence, MxEventFamily.OnDataChange)); + } + + await WaitUntilAsync( + () => metrics.GetSnapshot().WorkerEventQueueDepth == eventCount, + TestTimeout); + Assert.Equal(eventCount, metrics.GetSnapshot().WorkerEventQueueDepth); + Assert.Equal(WorkerClientState.Ready, client.State); + + using CancellationTokenSource drainTimeout = new(TestTimeout); + await using IAsyncEnumerator events = client + .ReadEventsAsync(drainTimeout.Token) + .GetAsyncEnumerator(drainTimeout.Token); + for (int read = 0; read < eventCount; read++) + { + Assert.True(await events.MoveNextAsync()); + } + + await WaitUntilAsync( + () => metrics.GetSnapshot().WorkerEventQueueDepth == 0, + TestTimeout); + Assert.Equal(0, metrics.GetSnapshot().WorkerEventQueueDepth); + } + private static WorkerClient CreateClient( PipePair pipePair, WorkerClientOptions? options = null,