fix(GWC-24): bound the worker event staging channel and unify the depth gauge

The GWC-04 remediation decoupled the read loop from event backpressure by staging events into an unbounded channel, so its TryWrite always succeeded and the only overflow fault was a single timed WriteAsync exceeding EventChannelFullModeTimeout. A consumer draining slower than the worker produces — each individual write still completing inside the window — therefore grew gateway memory without bound, without a fault, and without a metric: the queue-depth gauge counted only the bounded consumer channel, so staged events were invisible. Bound _eventStaging at 2 x EventChannelCapacity (Wait, single reader/writer, no synchronous continuations). A rejected staging TryWrite is the sustained slow-drain signal and faults the client ProtocolViolation with QueueOverflow("worker-event-staging"), guarded by IsTerminalState() so a completed channel during shutdown stays a silent drop. SetFaulted is non-blocking, so the read loop still never awaits behind events. The timed-write fault is unchanged and still catches the full-stall case earlier. Move the queue-depth increment from EnqueueWorkerEventAsync to StageWorkerEvent so the single counter reports total undelivered events (staged + queued); the decrement at consumer read was already correct. No new configuration key: the bound is derived, and gateway-side buffering per session is now at most 3 x MxGateway:Events:QueueCapacity. Coordination with still-open GWC-21 (EventChannelFullModeTimeout configurability) remains open and was not blocked on. Tests: StagingChannelOverflowFaultsWorkerWithoutWaitingForFullModeTimeout (5-min full-mode timeout so only the staging bound can fire; asserts an interleaved command reply still completes) and WorkerEventQueueDepthGaugeCountsStagedEvents. Docs updated in the same change: GatewayProcessDesign, MxAccessWorkerInstanceDesign, GatewayConfiguration, Metrics. GWC-24 flipped to Done in both trackers.
2026-08-07 05:35:07 -04:00
parent ead921cace
commit d4154e340c
8 changed files with 254 additions and 37 deletions
@@ -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,