Merge branch 'fix/gwc-24-staging-bound'
# Conflicts: # archreview/2026-07-12/remediation/00-tracking.md
This commit is contained in:
Joseph Doherty
@@ -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. |
|
||||
|
||||
@@ -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.
|
||||
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
|
||||
+3
-2
@@ -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
|
||||
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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<WorkerEvent> _eventStaging;
|
||||
private readonly int _eventStagingCapacity;
|
||||
private readonly ConcurrentDictionary<string, PendingCommand> _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<WorkerEvent>(
|
||||
new UnboundedChannelOptions
|
||||
_eventStagingCapacity = checked(2 * _options.EventChannelCapacity);
|
||||
_eventStaging = Channel.CreateBounded<WorkerEvent>(
|
||||
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
|
||||
|
||||
/// <summary>
|
||||
/// 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 <c>TryWrite</c> 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 <see cref="EventWriteLoopAsync"/> against the bounded consumer channel,
|
||||
/// off the read loop.
|
||||
/// <c>TryWrite</c> is non-blocking, so the read loop keeps dispatching replies, heartbeats and
|
||||
/// faults regardless of how backed up the event path is. It returns <c>false</c> 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 × <see cref="WorkerClientOptions.EventChannelCapacity"/> 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) —
|
||||
/// <see cref="SetFaulted"/> 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 <see cref="EnqueueWorkerEventAsync"/>.
|
||||
/// </summary>
|
||||
/// <param name="workerEvent">The event received from the worker.</param>
|
||||
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);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
@@ -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 <c>StreamEvents</c> or raise
|
||||
/// <c>MxGateway:Events:QueueCapacity</c>).
|
||||
/// <c>MxGateway:Events:QueueCapacity</c>). This is the full-stall half of
|
||||
/// the backpressure policy; a consumer that merely drains too slowly is
|
||||
/// caught by the staging bound in <see cref="StageWorkerEvent"/> (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.
|
||||
/// </summary>
|
||||
/// <param name="workerEvent">The event to enqueue.</param>
|
||||
/// <param name="cancellationToken">Cancellation token.</param>
|
||||
@@ -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);
|
||||
}
|
||||
|
||||
@@ -650,6 +650,147 @@ public sealed class WorkerClientTests
|
||||
Assert.Contains("MxGateway:Events:QueueCapacity", fault.Message);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// GWC-24: the staging channel between the read loop and the event writer is bounded at
|
||||
/// 2 × <see cref="WorkerClientOptions.EventChannelCapacity"/>, 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 <c>EnqueueWorkerEventAsync</c>. A command reply interleaved before the fault
|
||||
/// must still complete: the read loop never blocks behind events (the GWC-04 guarantee).
|
||||
/// </summary>
|
||||
/// <returns>A task that represents the asynchronous operation.</returns>
|
||||
[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<WorkerCommandReply> 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<WorkerClientException>(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);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 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.
|
||||
/// </summary>
|
||||
/// <returns>A task that represents the asynchronous operation.</returns>
|
||||
[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<WorkerEvent> 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,
|
||||
|
||||
Reference in New Issue
Block a user