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-08-07 07:50:38 -04:00
parent 10534ec906
commit 8df35cd63a
14 changed files with 912 additions and 58 deletions
@@ -1112,6 +1112,184 @@ public sealed class WorkerPipeSessionTests
        await SendShutdownAndWaitAsync(pipePair, runTask, cancellation.Token);
    }

+    /// <summary>
+    ///     WRK-27. An STA call outside the command dispatcher (the alarm poll) advertises itself on
+    ///     the heartbeat snapshot's <c>StaCallInProgress</c> 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.
+    /// </summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [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);
+        }
+    }
+
+    /// <summary>
+    ///     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.
+    /// </summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [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<WorkerEvent> 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);
+    }
+
+    /// <summary>
+    ///     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.
+    /// </summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [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<WorkerFrameProtocolException>(
+                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);
+    }
+
    /// <summary>
    ///     Regression test: a long in-flight STA command that keeps pumping
    ///     must NOT self-fault as <c>StaHung</c>, 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;
+
+        /// <summary>Initializes the passthrough over the given inner stream.</summary>
+        /// <param name="inner">The stream to delegate to.</param>
+        public FlushCountingPassthroughStream(Stream inner)
+        {
+            this.inner = inner;
+        }
+
+        /// <summary>Gets the number of <see cref="FlushAsync"/> calls observed so far.</summary>
+        public int FlushCount => Volatile.Read(ref flushCount);
+
+        /// <inheritdoc />
+        public override bool CanRead => inner.CanRead;
+
+        /// <inheritdoc />
+        public override bool CanSeek => inner.CanSeek;
+
+        /// <inheritdoc />
+        public override bool CanWrite => inner.CanWrite;
+
+        /// <inheritdoc />
+        public override long Length => inner.Length;
+
+        /// <inheritdoc />
+        public override long Position
+        {
+            get => inner.Position;
+            set => inner.Position = value;
+        }
+
+        /// <inheritdoc />
+        public override void Flush()
+        {
+            Interlocked.Increment(ref flushCount);
+            inner.Flush();
+        }
+
+        /// <inheritdoc />
+        public override Task FlushAsync(CancellationToken cancellationToken)
+        {
+            Interlocked.Increment(ref flushCount);
+            return inner.FlushAsync(cancellationToken);
+        }
+
+        /// <inheritdoc />
+        public override int Read(byte[] buffer, int offset, int count) => inner.Read(buffer, offset, count);
+
+        /// <inheritdoc />
+        public override Task<int> ReadAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
+            => inner.ReadAsync(buffer, offset, count, cancellationToken);
+
+        /// <inheritdoc />
+        public override long Seek(long offset, SeekOrigin origin) => inner.Seek(offset, origin);
+
+        /// <inheritdoc />
+        public override void SetLength(long value) => inner.SetLength(value);
+
+        /// <inheritdoc />
+        public override void Write(byte[] buffer, int offset, int count) => inner.Write(buffer, offset, count);
+
+        /// <inheritdoc />
+        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;