mxaccessgw/src/ZB.MOM.WW.MxGateway.Worker.Tests/Ipc/WorkerFrameProtocolTests.cs

using System;
using System.IO;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using ZB.MOM.WW.MxGateway.Contracts;
using ZB.MOM.WW.MxGateway.Contracts.Proto;
using ZB.MOM.WW.MxGateway.Worker.Ipc;
using ZB.MOM.WW.MxGateway.Worker.Tests.TestSupport;

namespace ZB.MOM.WW.MxGateway.Worker.Tests.Ipc;

public sealed class WorkerFrameProtocolTests
{
    private const string SessionId = "session-1";
    private const string Nonce = "nonce-secret";

    /// <summary>Verifies that valid envelopes round-trip through write and read.</summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task WriteAndReadAsync_WithValidEnvelope_RoundTripsFrame()
    {
        WorkerFrameProtocolOptions options = CreateOptions();
        using MemoryStream stream = new();
        WorkerEnvelope original = CreateGatewayHelloEnvelope();

        WorkerFrameWriter writer = new(stream, options);
        await writer.WriteAsync(original);
        stream.Position = 0;

        WorkerFrameReader reader = new(stream, options);
        WorkerEnvelope parsed = await reader.ReadAsync();

        Assert.Equal(original, parsed);
    }

    /// <summary>Verifies that wrong protocol version throws mismatch error.</summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task ReadAsync_WithWrongProtocolVersion_ThrowsProtocolVersionMismatch()
    {
        WorkerFrameProtocolOptions options = CreateOptions();
        WorkerEnvelope envelope = CreateGatewayHelloEnvelope();
        envelope.ProtocolVersion++;
        using MemoryStream stream = new(WorkerFrameTestHelpers.CreateFrame(envelope));

        WorkerFrameReader reader = new(stream, options);
        WorkerFrameProtocolException exception =
            await Assert.ThrowsAsync<WorkerFrameProtocolException>(
                async () => await reader.ReadAsync());

        Assert.Equal(WorkerFrameProtocolErrorCode.ProtocolVersionMismatch, exception.ErrorCode);
    }

    /// <summary>Verifies that wrong session ID throws mismatch error.</summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task ReadAsync_WithWrongSessionId_ThrowsSessionMismatch()
    {
        WorkerFrameProtocolOptions options = CreateOptions();
        WorkerEnvelope envelope = CreateGatewayHelloEnvelope();
        envelope.SessionId = "different-session";
        using MemoryStream stream = new(WorkerFrameTestHelpers.CreateFrame(envelope));

        WorkerFrameReader reader = new(stream, options);
        WorkerFrameProtocolException exception =
            await Assert.ThrowsAsync<WorkerFrameProtocolException>(
                async () => await reader.ReadAsync());

        Assert.Equal(WorkerFrameProtocolErrorCode.SessionMismatch, exception.ErrorCode);
    }

    /// <summary>
    /// Verifies that a frame whose length prefix is zero is rejected before the
    /// payload buffer is allocated. <c>docs/WorkerFrameProtocol.md</c> states the
    /// reader rejects zero-length payloads as a malformed-length error. The
    /// length prefix is the leading four bytes of the stream, so a four-zero-byte
    /// stream is exactly a frame declaring a zero-length payload.
    /// </summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task ReadAsync_WithZeroLengthPayload_ThrowsMalformedLength()
    {
        WorkerFrameProtocolOptions options = CreateOptions();
        using MemoryStream stream = new(new byte[sizeof(uint)]);

        WorkerFrameReader reader = new(stream, options);
        WorkerFrameProtocolException exception =
            await Assert.ThrowsAsync<WorkerFrameProtocolException>(
                async () => await reader.ReadAsync());

        Assert.Equal(WorkerFrameProtocolErrorCode.MalformedLength, exception.ErrorCode);
    }

    /// <summary>
    /// Verifies that a frame whose length prefix exceeds the configured maximum
    /// is rejected before the payload buffer is allocated. <c>docs/WorkerFrameProtocol.md</c>
    /// states the reader rejects oversized payloads as a message-too-large error.
    /// A small maximum is configured so the rejection is asserted without
    /// allocating a multi-megabyte buffer.
    /// </summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task ReadAsync_WithPayloadAboveConfiguredMaximum_ThrowsMessageTooLarge()
    {
        const int maxMessageBytes = 64;
        WorkerFrameProtocolOptions options = new(
            SessionId,
            GatewayContractInfo.WorkerProtocolVersion,
            Nonce,
            maxMessageBytes);
        byte[] frame = new byte[sizeof(uint)];
        WorkerFrameTestHelpers.WriteUInt32LittleEndian(frame, maxMessageBytes + 1);
        using MemoryStream stream = new(frame);

        WorkerFrameReader reader = new(stream, options);
        WorkerFrameProtocolException exception =
            await Assert.ThrowsAsync<WorkerFrameProtocolException>(
                async () => await reader.ReadAsync());

        Assert.Equal(WorkerFrameProtocolErrorCode.MessageTooLarge, exception.ErrorCode);
    }

    /// <summary>Verifies that malformed payload throws invalid envelope error.</summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task ReadAsync_WithMalformedPayload_ThrowsInvalidEnvelope()
    {
        WorkerFrameProtocolOptions options = CreateOptions();
        using MemoryStream stream = new(WorkerFrameTestHelpers.CreateFrame(new byte[] { 0x80 }));

        WorkerFrameReader reader = new(stream, options);
        WorkerFrameProtocolException exception =
            await Assert.ThrowsAsync<WorkerFrameProtocolException>(
                async () => await reader.ReadAsync());

        Assert.Equal(WorkerFrameProtocolErrorCode.InvalidEnvelope, exception.ErrorCode);
    }

    /// <summary>
    ///     Pins the <c>EndOfStream</c> branch of
    ///     <c>WorkerFrameReader.ReadExactlyOrThrowAsync</c>. The gateway
    ///     closing its end of the pipe during a partial-frame read is the
    ///     most common production transport failure; the reader must
    ///     surface this as <c>WorkerFrameProtocolErrorCode.EndOfStream</c>
    ///     so the worker session can fault deterministically rather than
    ///     spinning on a partial buffer. The stream here declares a 100-byte
    ///     payload but only supplies 50 bytes, so the inner read loop sees
    ///     <c>bytesRead == 0</c> mid-frame.
    /// </summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task ReadAsync_WhenStreamEndsMidFrame_ThrowsEndOfStream()
    {
        WorkerFrameProtocolOptions options = CreateOptions();
        byte[] frame = new byte[sizeof(uint) + 50];
        WorkerFrameTestHelpers.WriteUInt32LittleEndian(frame, 100);
        using MemoryStream stream = new(frame);

        WorkerFrameReader reader = new(stream, options);
        WorkerFrameProtocolException exception =
            await Assert.ThrowsAsync<WorkerFrameProtocolException>(
                async () => await reader.ReadAsync());

        Assert.Equal(WorkerFrameProtocolErrorCode.EndOfStream, exception.ErrorCode);
    }

    /// <summary>
    ///     Pins the writer-side <c>MessageTooLarge</c> branch. A session that
    ///     constructs an envelope whose serialised size exceeds
    ///     <c>MaxMessageBytes</c> must be rejected by the writer before any
    ///     bytes are sent down the pipe, so a misbehaving producer cannot
    ///     push the receiver past its bounds. A small <c>MaxMessageBytes</c>
    ///     is configured so a modest <c>GatewayHello</c> payload — with its
    ///     nonce padded out to several hundred bytes — exceeds the limit
    ///     without allocating anything large.
    /// </summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task WriteAsync_WithEnvelopeAboveConfiguredMaximum_ThrowsMessageTooLarge()
    {
        const int maxMessageBytes = 64;
        WorkerFrameProtocolOptions options = new(
            SessionId,
            GatewayContractInfo.WorkerProtocolVersion,
            Nonce,
            maxMessageBytes);
        using MemoryStream stream = new();
        WorkerFrameWriter writer = new(stream, options);

        WorkerEnvelope envelope = CreateGatewayHelloEnvelope();
        envelope.GatewayHello.GatewayVersion = new string('x', 1024);

        WorkerFrameProtocolException exception =
            await Assert.ThrowsAsync<WorkerFrameProtocolException>(
                async () => await writer.WriteAsync(envelope));

        Assert.Equal(WorkerFrameProtocolErrorCode.MessageTooLarge, exception.ErrorCode);
        Assert.Equal(0, stream.Length);
    }

    /// <summary>
    ///     Documents that the writer-side <c>InvalidEnvelope</c> branch
    ///     (raised when <c>WorkerEnvelope.CalculateSize()</c> returns 0) is
    ///     unreachable through public API. <c>WorkerEnvelopeValidator.Validate</c>
    ///     (run before the size check in <c>WorkerFrameWriter.WriteAsync</c>)
    ///     rejects any envelope whose <c>BodyCase</c> is <c>None</c> with
    ///     <c>InvalidEnvelope</c>; a body-less envelope is therefore
    ///     intercepted before the empty-payload branch can fire. Any
    ///     envelope carrying a typed body serialises at least the field
    ///     tag bytes, so <c>CalculateSize()</c> is strictly positive. This
    ///     test exercises the body-less path and asserts the same
    ///     <c>InvalidEnvelope</c> error code reaches the caller, pinning
    ///     the contract that "no body" is rejected before any size check.
    ///     The defensive zero-length branch in <c>WriteAsync</c> is left
    ///     in place because the cost is one comparison and removing it
    ///     would weaken the writer against future serialisation
    ///     regressions; this test makes its rationale visible.
    /// </summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task WriteAsync_WithEmptyEnvelope_ThrowsInvalidEnvelopeFromValidator()
    {
        WorkerFrameProtocolOptions options = CreateOptions();
        using MemoryStream stream = new();
        WorkerFrameWriter writer = new(stream, options);

        WorkerEnvelope envelope = new()
        {
            ProtocolVersion = GatewayContractInfo.WorkerProtocolVersion,
            SessionId = SessionId,
            Sequence = 1,
            // No body — BodyCase == None, validator rejects.
        };

        WorkerFrameProtocolException exception =
            await Assert.ThrowsAsync<WorkerFrameProtocolException>(
                async () => await writer.WriteAsync(envelope));

        Assert.Equal(WorkerFrameProtocolErrorCode.InvalidEnvelope, exception.ErrorCode);
        Assert.Equal(0, stream.Length);
    }

    /// <summary>Verifies that concurrent writes produce complete serialized frames.</summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task WriteAsync_WithConcurrentCalls_SerializesCompleteFrames()
    {
        WorkerFrameProtocolOptions options = CreateOptions();
        using MemoryStream stream = new();
        WorkerFrameWriter writer = new(stream, options);

        await Task.WhenAll(
            writer.WriteAsync(CreateGatewayHelloEnvelope(sequence: 1)),
            writer.WriteAsync(CreateGatewayHelloEnvelope(sequence: 2)),
            writer.WriteAsync(CreateGatewayHelloEnvelope(sequence: 3)));

        stream.Position = 0;
        WorkerFrameReader reader = new(stream, options);

        WorkerEnvelope first = await reader.ReadAsync();
        WorkerEnvelope second = await reader.ReadAsync();
        WorkerEnvelope third = await reader.ReadAsync();

        Assert.Equal(new ulong[] { 1, 2, 3 }, new[] { first.Sequence, second.Sequence, third.Sequence }.OrderBy(sequence => sequence));
    }

    /// <summary>
    /// The reader rents its payload buffer from a shared pool, so a rented
    /// buffer can be larger than the current frame and may carry bytes from
    /// a previous, larger frame. Reading frames of differing sizes
    /// back-to-back through one reader must parse each frame using only its
    /// own payload length, never trailing pooled bytes.
    /// </summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task ReadAsync_WithVaryingFrameSizes_ParsesEachFrameExactly()
    {
        WorkerFrameProtocolOptions options = CreateOptions();
        using MemoryStream stream = new();
        WorkerFrameWriter writer = new(stream, options);

        // A large-payload frame followed by a small-payload frame: if the
        // reader reused a pooled buffer without honouring the second frame's
        // length, the small frame would parse with stale trailing bytes.
        WorkerEnvelope large = CreateGatewayHelloEnvelope(sequence: 1);
        large.GatewayHello.GatewayVersion = new string('x', 4096);
        WorkerEnvelope small = CreateGatewayHelloEnvelope(sequence: 2);

        await writer.WriteAsync(large);
        await writer.WriteAsync(small);
        stream.Position = 0;

        WorkerFrameReader reader = new(stream, options);
        WorkerEnvelope firstParsed = await reader.ReadAsync();
        WorkerEnvelope secondParsed = await reader.ReadAsync();

        Assert.Equal(large, firstParsed);
        Assert.Equal(small, secondParsed);
    }

    private static WorkerFrameProtocolOptions CreateOptions()
    {
        return new WorkerFrameProtocolOptions(
            SessionId,
            GatewayContractInfo.WorkerProtocolVersion,
            Nonce);
    }

    /// <summary>
    /// Verifies that under concurrent writers every frame receives a distinct, gap-free sequence in
    /// strictly increasing on-wire order — the sequence is stamped by the writer at write time, so the
    /// wire order and the stamped sequence always agree.
    /// </summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task WriteAsync_UnderConcurrentCalls_StampsGapFreeMonotonicSequence()
    {
        const int frameCount = 50;
        WorkerFrameProtocolOptions options = CreateOptions();
        using MemoryStream stream = new();
        WorkerFrameWriter writer = new(stream, options);

        await Task.WhenAll(
            Enumerable.Range(0, frameCount).Select(_ => writer.WriteAsync(CreateEventEnvelope())));

        stream.Position = 0;
        WorkerFrameReader reader = new(stream, options);
        ulong[] sequences = new ulong[frameCount];
        for (int index = 0; index < frameCount; index++)
        {
            sequences[index] = (await reader.ReadAsync()).Sequence;
        }

        // On-wire order is strictly increasing 1..frameCount with no gaps or duplicates.
        Assert.Equal(Enumerable.Range(1, frameCount).Select(value => (ulong)value), sequences);
    }

    /// <summary>
    /// Verifies that when a control frame and an event frame are both queued behind an in-progress
    /// write, the draining lock-holder writes the control frame first even though the event was queued
    /// earlier.
    /// </summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task WriteAsync_WhenControlAndEventQueuedTogether_WritesControlFirst()
    {
        WorkerFrameProtocolOptions options = CreateOptions();
        using GatedWriteStream stream = new();
        WorkerFrameWriter writer = new(stream, options);

        // First write 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 first, then a control frame, while the writer is blocked. Both wait for the lock.
        Task eventWrite = writer.WriteAsync(CreateEventEnvelope(), WorkerFrameWritePriority.Event);
        Task controlWrite = writer.WriteAsync(CreateGatewayHelloEnvelope(), WorkerFrameWritePriority.Control);
        await Task.Delay(50);

        stream.ReleaseFirstWrite();
        await AwaitWithTimeoutAsync(Task.WhenAll(firstWrite, eventWrite, controlWrite));

        stream.Position = 0;
        WorkerFrameReader reader = new(stream, options);
        WorkerEnvelope frame1 = await reader.ReadAsync();
        WorkerEnvelope frame2 = await reader.ReadAsync();
        WorkerEnvelope frame3 = await reader.ReadAsync();

        Assert.Equal(WorkerEnvelope.BodyOneofCase.GatewayHello, frame1.BodyCase);
        // The control frame jumped ahead of the earlier-queued event.
        Assert.Equal(WorkerEnvelope.BodyOneofCase.GatewayHello, frame2.BodyCase);
        Assert.Equal(WorkerEnvelope.BodyOneofCase.WorkerEvent, frame3.BodyCase);
    }

    /// <summary>
    /// Verifies the writer coalesces the flush across a batch of frames drained together: four frames
    /// queued behind an in-progress write drain in a single pass and share one FlushAsync, not four.
    /// Every frame still reaches the wire intact.
    /// <para>
    /// The burst is all-event on purpose. The control-frame completion decoupling made the drain flush
    /// at each control-to-event boundary, so a pass that mixes classes legitimately pays one flush per
    /// class run; the property
    /// worth pinning is that a run of same-class frames — the event hot path — still costs exactly one
    /// flush no matter how many frames drain together. The mixed shape has its own count assertion in
    /// <see cref="DrainPass_MixedClasses_FlushesControlRunBeforeWritingEvents"/>.
    /// </para>
    /// </summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task WriteAsync_WhenBatchDrainedTogether_FlushesOnce()
    {
        WorkerFrameProtocolOptions options = CreateOptions();
        using GatedWriteStream stream = new();
        WorkerFrameWriter writer = new(stream, options);

        // A blocked first write occupies the writer and holds the lock while more frames queue behind it.
        Task firstWrite = writer.WriteAsync(CreateEventEnvelope(), WorkerFrameWritePriority.Event);
        await AwaitWithTimeoutAsync(stream.FirstWriteStarted);

        Task eventWrite1 = writer.WriteAsync(CreateEventEnvelope(), WorkerFrameWritePriority.Event);
        Task eventWrite2 = writer.WriteAsync(CreateEventEnvelope(), WorkerFrameWritePriority.Event);
        Task eventWrite3 = writer.WriteAsync(CreateEventEnvelope(), WorkerFrameWritePriority.Event);
        await Task.Delay(50);

        stream.ReleaseFirstWrite();
        await AwaitWithTimeoutAsync(Task.WhenAll(firstWrite, eventWrite1, eventWrite2, eventWrite3));

        // Four frames written in one drain pass => exactly one flush.
        Assert.Equal(1, stream.FlushCount);

        stream.Position = 0;
        WorkerFrameReader reader = new(stream, options);
        for (int index = 0; index < 4; index++)
        {
            WorkerEnvelope frame = await reader.ReadAsync();
            Assert.NotEqual(WorkerEnvelope.BodyOneofCase.None, frame.BodyCase);
        }
    }

    /// <summary>
    /// Control-frame completion decoupling. A control frame's delivery point must not be charged for
    /// the event backlog behind it. The priority scheduler already wrote control <em>bytes</em> first,
    /// but a frame counts as
    /// delivered only once flushed, and the pass deferred its single flush — and every completion —
    /// until after the events. The drain now flushes at the control-to-event boundary: with two control
    /// frames written and the first event write blocked inside the stream, the flush that closes out the
    /// control run has already run, so the heartbeat or reply is on the pipe rather than waiting behind
    /// the batch. Exactly two flushes for the pass — one per class run, not one per control frame.
    /// <para>
    /// The assertion is on the flush, not on the queued callers' returned tasks, because those tasks are
    /// still gated by the write lock they lost to the drainer (see the latency contract on
    /// <c>WorkerFrameWriter.WriteAsync</c>): the completion resolves at the boundary flush, but a
    /// lock-race loser observes it only once the drainer releases the lock.
    /// </para>
    /// </summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task DrainPass_MixedClasses_FlushesControlRunBeforeWritingEvents()
    {
        WorkerFrameProtocolOptions options = CreateOptions();
        // Frame 1 (control) gates the pass open; frame 3 is the pass's first event write, which blocks
        // so the boundary flush can be observed with the event batch still unwritten.
        using GatedWriteStream stream = new(secondGateWriteIndex: 3);
        WorkerFrameWriter writer = new(stream, options);

        Task firstControl = writer.WriteAsync(CreateGatewayHelloEnvelope(), WorkerFrameWritePriority.Control);
        await AwaitWithTimeoutAsync(stream.FirstWriteStarted);

        Task secondControl = writer.WriteAsync(CreateShutdownAckEnvelope(), WorkerFrameWritePriority.Control);
        Task eventWrite1 = writer.WriteAsync(CreateEventEnvelope(), WorkerFrameWritePriority.Event);
        Task eventWrite2 = writer.WriteAsync(CreateEventEnvelope(), WorkerFrameWritePriority.Event);
        await Task.Delay(50);

        stream.ReleaseFirstWrite();

        // The drain writes both control frames and is now blocked on the first event write.
        await AwaitWithTimeoutAsync(stream.SecondGateWriteStarted);

        // The control run was flushed before the event batch was written — not after it.
        Assert.Equal(1, stream.FlushCount);
        Assert.False(eventWrite1.IsCompleted);
        Assert.False(eventWrite2.IsCompleted);

        stream.ReleaseSecondGateWrite();
        await AwaitWithTimeoutAsync(
            Task.WhenAll(firstControl, secondControl, eventWrite1, eventWrite2));

        // One flush per class run: the control run, then the event run at the end of the pass.
        Assert.Equal(2, stream.FlushCount);

        stream.Position = 0;
        WorkerFrameReader reader = new(stream, options);
        WorkerEnvelope frame1 = await reader.ReadAsync();
        WorkerEnvelope frame2 = await reader.ReadAsync();
        WorkerEnvelope frame3 = await reader.ReadAsync();
        WorkerEnvelope frame4 = await reader.ReadAsync();
        Assert.Equal(WorkerEnvelope.BodyOneofCase.GatewayHello, frame1.BodyCase);
        Assert.Equal(WorkerEnvelope.BodyOneofCase.WorkerShutdownAck, frame2.BodyCase);
        Assert.Equal(WorkerEnvelope.BodyOneofCase.WorkerEvent, frame3.BodyCase);
        Assert.Equal(WorkerEnvelope.BodyOneofCase.WorkerEvent, frame4.BodyCase);
        Assert.Equal(stream.Length, stream.Position);
    }

    /// <summary>
    /// Control-frame completion decoupling, the inverse guard. An event frame's completion boundary is
    /// still the end-of-pass flush: a pure-event pass takes no boundary flush, so with two events
    /// already written and the third
    /// blocked mid-write, nothing has been flushed and no event can have been reported delivered. Only
    /// a class transition may move a flush earlier — a plain event backlog may not.
    /// </summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task DrainPass_PureEventRun_DoesNotFlushBeforeThePassEnds()
    {
        WorkerFrameProtocolOptions options = CreateOptions();
        using GatedWriteStream stream = new(secondGateWriteIndex: 3);
        WorkerFrameWriter writer = new(stream, options);

        Task eventWrite1 = writer.WriteAsync(CreateEventEnvelope(), WorkerFrameWritePriority.Event);
        await AwaitWithTimeoutAsync(stream.FirstWriteStarted);

        Task eventWrite2 = writer.WriteAsync(CreateEventEnvelope(), WorkerFrameWritePriority.Event);
        Task eventWrite3 = writer.WriteAsync(CreateEventEnvelope(), WorkerFrameWritePriority.Event);
        await Task.Delay(50);

        stream.ReleaseFirstWrite();
        await AwaitWithTimeoutAsync(stream.SecondGateWriteStarted);

        // Two event frames written, none flushed: no event frame's delivery point has been reached.
        Assert.Equal(0, stream.FlushCount);
        Assert.False(eventWrite1.IsCompleted);
        Assert.False(eventWrite2.IsCompleted);

        stream.ReleaseSecondGateWrite();
        await AwaitWithTimeoutAsync(Task.WhenAll(eventWrite1, eventWrite2, eventWrite3));
        Assert.Equal(1, stream.FlushCount);
    }

    /// <summary>
    /// Control-frame completion decoupling. The boundary flush is charged per class run, not per
    /// control frame: a pass carrying nothing but control frames still pays exactly one flush. Flushing
    /// after every control frame
    /// would reinstate the syscall-per-heartbeat cost WRK-12 removed.
    /// </summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task DrainPass_PureControlRun_FlushesOnce()
    {
        WorkerFrameProtocolOptions options = CreateOptions();
        using GatedWriteStream stream = new();
        WorkerFrameWriter writer = new(stream, options);

        Task firstControl = writer.WriteAsync(CreateGatewayHelloEnvelope(), WorkerFrameWritePriority.Control);
        await AwaitWithTimeoutAsync(stream.FirstWriteStarted);

        Task secondControl = writer.WriteAsync(CreateGatewayHelloEnvelope(), WorkerFrameWritePriority.Control);
        Task thirdControl = writer.WriteAsync(CreateShutdownAckEnvelope(), WorkerFrameWritePriority.Control);
        await Task.Delay(50);

        stream.ReleaseFirstWrite();
        await AwaitWithTimeoutAsync(Task.WhenAll(firstControl, secondControl, thirdControl));

        Assert.Equal(1, stream.FlushCount);
    }

    /// <summary>
    /// Control-frame completion decoupling, the new failure window. The boundary flush is a new place
    /// the pipe can break with frames written but not yet delivered, so it must fail exactly like the
    /// end-of-pass flush: every written control
    /// frame fails, and so do the event frame the drain had already claimed off its queue (nothing else
    /// would ever complete it) and every frame still queued, so no caller waits forever on a stream that
    /// will not recover. The event bytes never reach the wire — the drain stops at the fault.
    /// </summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task DrainPass_WhenBoundaryFlushFails_FailsWrittenClaimedAndQueuedFrames()
    {
        const string faultMessage = "boundary flush failed";
        WorkerFrameProtocolOptions options = CreateOptions();
        using FlushFaultingGatedStream stream = new(faultMessage);
        WorkerFrameWriter writer = new(stream, options);

        Task firstControl = writer.WriteAsync(CreateGatewayHelloEnvelope(), WorkerFrameWritePriority.Control);
        await AwaitWithTimeoutAsync(stream.FirstWriteStarted);

        Task secondControl = writer.WriteAsync(CreateShutdownAckEnvelope(), WorkerFrameWritePriority.Control);
        Task claimedEvent = writer.WriteAsync(CreateEventEnvelope(), WorkerFrameWritePriority.Event);
        Task queuedEvent = writer.WriteAsync(CreateEventEnvelope(), WorkerFrameWritePriority.Event);
        await Task.Delay(50);

        // The drain writes both control frames, claims the first event, and faults on the boundary flush.
        stream.ReleaseFirstWrite();

        // AwaitWithTimeoutAsync turns a frame nobody ever completes into a TimeoutException — a failed
        // assertion rather than a hung test run.
        foreach (Task write in new[] { firstControl, secondControl, claimedEvent, queuedEvent })
        {
            IOException failure = await Assert.ThrowsAsync<IOException>(
                async () => await AwaitWithTimeoutAsync(write));
            Assert.Equal(faultMessage, failure.Message);
        }

        // Only the control frames reached the wire; the claimed event was never written.
        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.WorkerShutdownAck, frame2.BodyCase);
        Assert.Equal(stream.Length, stream.Position);
    }

    /// <summary>
    /// Verifies a per-frame rejection does not burn a sequence number (WRK-23). The sequence is a
    /// diagnostic counter, so a gap breaks nothing functionally — but an operator correlating a pipe
    /// capture reads a gap as a lost frame and chases a bug that does not exist, and the gap-free
    /// guarantee the concurrent-write test asserts would otherwise only hold until the first
    /// rejection.
    /// </summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task WriteAsync_PerFrameRejection_DoesNotConsumeSequence()
    {
        const int maxMessageBytes = 512;
        WorkerFrameProtocolOptions options = new(
            SessionId,
            GatewayContractInfo.WorkerProtocolVersion,
            Nonce,
            maxMessageBytes);
        using MemoryStream stream = new();
        WorkerFrameWriter writer = new(stream, options);

        await writer.WriteAsync(CreateEventEnvelope());

        WorkerEnvelope oversized = CreateGatewayHelloEnvelope();
        oversized.GatewayHello.GatewayVersion = new string('x', maxMessageBytes * 2);
        WorkerFrameProtocolException exception =
            await Assert.ThrowsAsync<WorkerFrameProtocolException>(
                async () => await writer.WriteAsync(oversized));
        Assert.Equal(WorkerFrameProtocolErrorCode.MessageTooLarge, exception.ErrorCode);

        await writer.WriteAsync(CreateEventEnvelope());

        stream.Position = 0;
        WorkerFrameReader reader = new(stream, options);
        WorkerEnvelope first = await reader.ReadAsync();
        WorkerEnvelope second = await reader.ReadAsync();

        // Two frames reached the wire; the rejected frame in between left no gap.
        Assert.Equal(1UL, first.Sequence);
        Assert.Equal(2UL, second.Sequence);
        Assert.Equal(stream.Length, stream.Position);
    }

    /// <summary>Verifies a zero negotiated frame maximum keeps the constructor default.</summary>
    [Fact]
    public void AdoptNegotiatedMaxMessageBytes_WithZero_KeepsDefault()
    {
        WorkerFrameProtocolOptions options = CreateOptions();
        int original = options.MaxMessageBytes;
        options.AdoptNegotiatedMaxMessageBytes(0);
        Assert.Equal(original, options.MaxMessageBytes);
    }

    /// <summary>Verifies an in-range negotiated frame maximum is adopted.</summary>
    [Fact]
    public void AdoptNegotiatedMaxMessageBytes_WithInRangeValue_Adopts()
    {
        WorkerFrameProtocolOptions options = CreateOptions();
        options.AdoptNegotiatedMaxMessageBytes(4 * 1024 * 1024);
        Assert.Equal(4 * 1024 * 1024, options.MaxMessageBytes);
    }

    /// <summary>Verifies a negotiated frame maximum above the worker ceiling is rejected.</summary>
    [Fact]
    public void AdoptNegotiatedMaxMessageBytes_AboveCeiling_Throws()
    {
        WorkerFrameProtocolOptions options = CreateOptions();
        WorkerFrameProtocolException exception = Assert.Throws<WorkerFrameProtocolException>(
            () => options.AdoptNegotiatedMaxMessageBytes((uint)WorkerFrameProtocolOptions.MaxNegotiableFrameBytes + 1));
        Assert.Equal(WorkerFrameProtocolErrorCode.InvalidConfiguration, exception.ErrorCode);
    }

    /// <summary>
    /// Verifies a negotiated frame maximum below the worker floor is rejected as
    /// <c>InvalidConfiguration</c> (WRK-24), and that exactly the floor is adopted. The floor matches
    /// the gateway's own <c>GatewayOptionsValidator.MinimumMaxMessageBytes</c>, 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.
    /// </summary>
    [Fact]
    public void AdoptNegotiatedMaxMessageBytes_BelowFloor_ThrowsInvalidConfiguration()
    {
        WorkerFrameProtocolOptions belowFloor = CreateOptions();
        WorkerFrameProtocolException exception = Assert.Throws<WorkerFrameProtocolException>(
            () => 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);
    }

    /// <summary>
    /// WRK-22 / IPC-26. A <c>WriteAsync</c> 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.
    /// </summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [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<OperationCanceledException>(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);
    }

    /// <summary>
    /// NEXT-04. A frame claimed by the draining lock-holder before its caller's cancellation lands
    /// is abandoned — the cancelled caller never awaits its completion. If the wire write then
    /// faults, the tombstone path's fault-observing continuation must still observe the exception
    /// so it never surfaces as <see cref="TaskScheduler.UnobservedTaskException"/>.
    /// </summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
    public async Task WriteAsync_ClaimedFrameAbandonedByCancellation_FaultIsObserved()
    {
        string marker = $"NEXT-04-{Guid.NewGuid():N}";
        WorkerFrameProtocolOptions options = CreateOptions();
        bool sawUnobservedMarkerFault = false;
        EventHandler<UnobservedTaskExceptionEventArgs> handler = (sender, args) =>
        {
            if (args.Exception.ToString().Contains(marker))
            {
                sawUnobservedMarkerFault = true;
            }
        };

        TaskScheduler.UnobservedTaskException += handler;
        try
        {
            using (SecondWriteFaultingGatedStream stream = new SecondWriteFaultingGatedStream(marker))
            {
                WorkerFrameWriter writer = new WorkerFrameWriter(stream, options);

                // Writer A holds the lock, blocked mid-write of its own frame.
                Task firstWrite = writer.WriteAsync(CreateGatewayHelloEnvelope(), WorkerFrameWritePriority.Control);
                await AwaitWithTimeoutAsync(stream.FirstWriteStarted);

                using (CancellationTokenSource cts = new CancellationTokenSource())
                {
                    // Queue the doomed event write behind A, release A so its drain claims the
                    // event frame and blocks mid-write of it, then cancel the queued caller —
                    // the frame is claimed, so the caller unwinds without an awaiter for it.
                    Task abandonedWrite = writer.WriteAsync(CreateEventEnvelope(), WorkerFrameWritePriority.Event, cts.Token);
                    stream.ReleaseFirstWrite();
                    await AwaitWithTimeoutAsync(stream.SecondWriteStarted);
                    cts.Cancel();
                    await Assert.ThrowsAnyAsync<OperationCanceledException>(async () => await abandonedWrite);
                }

                // Fault the abandoned frame's wire write; observe writer A's own outcome so only
                // the abandoned frame's completion could ever raise the marker unobserved.
                stream.ReleaseSecondWrite();
                _ = await Record.ExceptionAsync(async () => await firstWrite);
            }

            GC.Collect();
            GC.WaitForPendingFinalizers();
            GC.Collect();
        }
        finally
        {
            TaskScheduler.UnobservedTaskException -= handler;
        }

        Assert.False(
            sawUnobservedMarkerFault,
            "The abandoned frame's write fault surfaced as an unobserved-task exception.");
    }

    /// <summary>
    /// 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.
    /// </summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [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<OperationCanceledException>(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);
    }

    /// <summary>
    /// 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.
    /// </summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [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);
        }
    }

    /// <summary>
    /// 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.
    /// </summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [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
        {
            ProtocolVersion = GatewayContractInfo.WorkerProtocolVersion,
            SessionId = SessionId,
            Sequence = sequence,
            GatewayHello = new GatewayHello
            {
                SupportedProtocolVersion = GatewayContractInfo.WorkerProtocolVersion,
                Nonce = Nonce,
                GatewayVersion = "test-gateway",
            },
        };
    }

    // net48 has no Task.WaitAsync(TimeSpan); fail the test rather than hang if the writer misbehaves.
    private static async Task AwaitWithTimeoutAsync(Task task)
    {
        Task completed = await Task.WhenAny(task, Task.Delay(TimeSpan.FromSeconds(5)));
        if (completed != task)
        {
            throw new TimeoutException("Timed out waiting for the frame writer.");
        }

        await task;
    }

    private static WorkerEnvelope CreateEventEnvelope()
    {
        return new WorkerEnvelope
        {
            ProtocolVersion = GatewayContractInfo.WorkerProtocolVersion,
            SessionId = SessionId,
            WorkerEvent = new WorkerEvent
            {
                Event = new MxEvent { SessionId = SessionId },
            },
        };
    }

    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;

        /// <summary>Gets the number of <see cref="FlushAsync"/> calls observed so far.</summary>
        public int FlushCount => Volatile.Read(ref _flushCount);

        /// <inheritdoc />
        public override Task FlushAsync(CancellationToken cancellationToken)
        {
            Interlocked.Increment(ref _flushCount);
            return base.FlushAsync(cancellationToken);
        }
    }

    // A MemoryStream whose first write blocks until released and whose second write blocks until
    // released and then throws, so a test can abandon a claimed frame by cancellation and fault its
    // wire write afterwards (NEXT-04).
    private sealed class SecondWriteFaultingGatedStream : MemoryStream
    {
        private readonly SemaphoreSlim _firstRelease = new SemaphoreSlim(0);
        private readonly SemaphoreSlim _secondRelease = new SemaphoreSlim(0);
        private readonly TaskCompletionSource<bool> _firstWriteStarted =
            new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
        private readonly TaskCompletionSource<bool> _secondWriteStarted =
            new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
        private readonly string _faultMessage;
        private int _writeCount;

        public SecondWriteFaultingGatedStream(string faultMessage)
        {
            _faultMessage = faultMessage;
        }

        /// <summary>Gets a task that completes once the first <see cref="WriteAsync"/> call has started blocking.</summary>
        public Task FirstWriteStarted => _firstWriteStarted.Task;

        /// <summary>Gets a task that completes once the second <see cref="WriteAsync"/> call has started blocking.</summary>
        public Task SecondWriteStarted => _secondWriteStarted.Task;

        /// <summary>Releases the first blocked write so it can complete.</summary>
        public void ReleaseFirstWrite() => _firstRelease.Release();

        /// <summary>Releases the second blocked write so it can throw.</summary>
        public void ReleaseSecondWrite() => _secondRelease.Release();

        /// <inheritdoc />
        public override async Task WriteAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
        {
            int writeIndex = Interlocked.Increment(ref _writeCount);
            if (writeIndex == 1)
            {
                _firstWriteStarted.TrySetResult(true);
                await _firstRelease.WaitAsync(cancellationToken);
            }
            else if (writeIndex == 2)
            {
                _secondWriteStarted.TrySetResult(true);
                await _secondRelease.WaitAsync(cancellationToken);
                throw new IOException(_faultMessage);
            }

            await base.WriteAsync(buffer, offset, count, cancellationToken);
        }

        /// <inheritdoc />
        protected override void Dispose(bool disposing)
        {
            if (disposing)
            {
                _firstRelease.Dispose();
                _secondRelease.Dispose();
            }

            base.Dispose(disposing);
        }
    }

    // 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. A second, optional gate on
    // a chosen write index lets a test stop a drain pass mid-flight — at a class boundary, say — and
    // sample what the writer has already flushed while the rest of the pass is still unwritten.
    private sealed class GatedWriteStream : MemoryStream
    {
        private readonly SemaphoreSlim _release = new SemaphoreSlim(0);
        private readonly SemaphoreSlim _secondGateRelease = new SemaphoreSlim(0);
        private readonly TaskCompletionSource<bool> _firstWriteStarted =
            new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
        private readonly TaskCompletionSource<bool> _secondGateWriteStarted =
            new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
        private readonly int _secondGateWriteIndex;
        private int _writeCount;
        private int _flushCount;

        /// <summary>Initializes a new instance of the GatedWriteStream class.</summary>
        /// <param name="secondGateWriteIndex">
        /// One-based index of a later write to block as well, or 0 (the default) to gate only the first
        /// write. Write indexes start at 1, so 0 never matches.
        /// </param>
        public GatedWriteStream(int secondGateWriteIndex = 0)
        {
            _secondGateWriteIndex = secondGateWriteIndex;
        }

        /// <summary>Gets a task that completes once the first <see cref="WriteAsync"/> call has started blocking.</summary>
        public Task FirstWriteStarted => _firstWriteStarted.Task;

        /// <summary>Gets a task that completes once the second gated <see cref="WriteAsync"/> call has started blocking.</summary>
        public Task SecondGateWriteStarted => _secondGateWriteStarted.Task;

        /// <summary>Gets the number of <see cref="FlushAsync"/> calls observed so far.</summary>
        public int FlushCount => Volatile.Read(ref _flushCount);

        /// <summary>Releases the first blocked write so it can complete.</summary>
        public void ReleaseFirstWrite() => _release.Release();

        /// <summary>Releases the second gated write so it can complete.</summary>
        public void ReleaseSecondGateWrite() => _secondGateRelease.Release();

        /// <inheritdoc />
        public override async Task WriteAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
        {
            int writeIndex = Interlocked.Increment(ref _writeCount);
            if (writeIndex == 1)
            {
                _firstWriteStarted.TrySetResult(true);
                await _release.WaitAsync(cancellationToken);
            }
            else if (writeIndex == _secondGateWriteIndex)
            {
                _secondGateWriteStarted.TrySetResult(true);
                await _secondGateRelease.WaitAsync(cancellationToken);
            }

            await base.WriteAsync(buffer, offset, count, cancellationToken);
        }

        /// <inheritdoc />
        public override Task FlushAsync(CancellationToken cancellationToken)
        {
            Interlocked.Increment(ref _flushCount);
            return base.FlushAsync(cancellationToken);
        }

        /// <inheritdoc />
        protected override void Dispose(bool disposing)
        {
            if (disposing)
            {
                _release.Dispose();
                _secondGateRelease.Dispose();
            }

            base.Dispose(disposing);
        }
    }

    // A MemoryStream whose first write blocks until released and whose every FlushAsync throws, so a test
    // can fault the class-boundary flush with control frames already written and an event frame already
    // claimed off its queue.
    private sealed class FlushFaultingGatedStream : MemoryStream
    {
        private readonly SemaphoreSlim _release = new SemaphoreSlim(0);
        private readonly TaskCompletionSource<bool> _firstWriteStarted =
            new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
        private readonly string _faultMessage;
        private int _writeCount;

        /// <summary>Initializes a new instance of the FlushFaultingGatedStream class.</summary>
        /// <param name="faultMessage">Message carried by the <see cref="IOException"/> every flush throws.</param>
        public FlushFaultingGatedStream(string faultMessage)
        {
            _faultMessage = faultMessage;
        }

        /// <summary>Gets a task that completes once the first <see cref="WriteAsync"/> call has started blocking.</summary>
        public Task FirstWriteStarted => _firstWriteStarted.Task;

        /// <summary>Releases the first blocked write so it can complete.</summary>
        public void ReleaseFirstWrite() => _release.Release();

        /// <inheritdoc />
        public override async Task WriteAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
        {
            if (Interlocked.Increment(ref _writeCount) == 1)
            {
                _firstWriteStarted.TrySetResult(true);
                await _release.WaitAsync(cancellationToken);
            }

            await base.WriteAsync(buffer, offset, count, cancellationToken);
        }

        /// <inheritdoc />
        public override Task FlushAsync(CancellationToken cancellationToken)
        {
            return Task.FromException(new IOException(_faultMessage));
        }

        /// <inheritdoc />
        protected override void Dispose(bool disposing)
        {
            if (disposing)
            {
                _release.Dispose();
            }

            base.Dispose(disposing);
        }
    }
}