fix: more flaky test fixes

src/TurboHTTP.IntegrationTests.End2End/H11/WirePipeliningSpec.cs

@@ -53,7 +53,12 @@ private async Task<string> ReadUntilThreeResponsesAsync(NetworkStream stream, So
 
         try
         {
-            while (CountOccurrences(sb.ToString(), "HTTP/1.1 200") < 3)
+            // Read until all three response bodies have arrived, not just their status lines: the
+            // final response's header and body routinely land in separate TCP segments, so counting
+            // "HTTP/1.1 200" alone would stop before the last body and race the wire. A genuinely
+            // dropped or reordered body is still caught — its marker never arrives, the read blocks
+            // to the receive timeout, and the in-order assertion below then fails.
+            while (!HasAllThreeBodies(sb.ToString()))
             {
                 var read = await Task.Run(() => stream.Read(buffer, 0, buffer.Length));
                 if (read == 0)
@@ -70,6 +75,11 @@ private async Task<string> ReadUntilThreeResponsesAsync(NetworkStream stream, So
         return sb.ToString();
     }
 
+    private static bool HasAllThreeBodies(string raw)
+        => raw.Contains("RESP-1", StringComparison.Ordinal)
+           && raw.Contains("RESP-2", StringComparison.Ordinal)
+           && raw.Contains("RESP-3", StringComparison.Ordinal);
+
     private static int CountOccurrences(string haystack, string needle)
     {
         var count = 0;

src/TurboHTTP.Tests/Protocol/Body/QueuedBodyReaderSpec.cs

@@ -206,4 +206,63 @@ public async Task ReadAsync_should_succeed_when_data_arrives_before_cancellation
         Assert.Equal("hello"u8.ToArray(), result.Memory.ToArray());
         Assert.False(result.IsCompleted);
     }
+
+    [Fact(Timeout = 5000)]
+    public async Task Reset_should_not_return_a_checked_out_rental_to_the_pool()
+    {
+        // Regression: on connection teardown/abort, Reset/Dispose ran while a consumer was
+        // still reading the current chunk and returned its rental to the shared ArrayPool.
+        // A concurrent stream then re-rented and overwrote it — correct-length, wrong-content
+        // corruption. The checked-out rental must be returned only by the consumer's AdvanceTo.
+        var pool = new TrackingArrayPool();
+        var reader = new QueuedBodyReader(4, pool);
+
+        reader.TryEnqueue("payload"u8);
+        var result = await reader.ReadAsync(TestContext.Current.CancellationToken);
+        Assert.Equal("payload"u8.ToArray(), result.Memory.ToArray());
+
+        // Teardown while the consumer still holds result.Memory.
+        reader.Dispose();
+        Assert.Equal(0, pool.ReturnedCount);
+
+        // The consumer finishes reading and advances: the rental is returned exactly once.
+        reader.AdvanceTo();
+        Assert.Equal(1, pool.ReturnedCount);
+    }
+
+    [Fact(Timeout = 5000)]
+    public void Reset_should_return_queued_but_unread_rentals()
+    {
+        // Queued chunks were never handed to the consumer, so Reset must reclaim them
+        // (no leak) — only the published _current chunk is left for the consumer.
+        var pool = new TrackingArrayPool();
+        var reader = new QueuedBodyReader(4, pool);
+
+        reader.TryEnqueue("a"u8);
+        reader.TryEnqueue("b"u8);
+        Assert.Equal(2, pool.RentedCount);
+
+        reader.Dispose();
+        Assert.Equal(2, pool.ReturnedCount);
+    }
+
+    private sealed class TrackingArrayPool : System.Buffers.ArrayPool<byte>
+    {
+        private readonly System.Buffers.ArrayPool<byte> _inner = System.Buffers.ArrayPool<byte>.Shared;
+
+        public int RentedCount { get; private set; }
+        public int ReturnedCount { get; private set; }
+
+        public override byte[] Rent(int minimumLength)
+        {
+            RentedCount++;
+            return _inner.Rent(minimumLength);
+        }
+
+        public override void Return(byte[] array, bool clearArray = false)
+        {
+            ReturnedCount++;
+            _inner.Return(array, clearArray);
+        }
+    }
 }

src/TurboHTTP.Tests/Protocol/ProtocolNegotiatingStateMachineSpec.cs

@@ -142,6 +142,34 @@ public void DecodeClientData_should_stay_sniffing_for_insufficient_data()
         Assert.Empty(ops.ScheduledTimers);
     }
 
+    [Fact(Timeout = 5000)]
+    [Trait("RFC", "RFC9112-9.3.2")]
+    public void MaxConcurrentRequests_should_serialize_dispatch_for_negotiated_http11()
+    {
+        var ops = new FakeServerOps();
+        var sm = new ProtocolNegotiatingStateMachine(new TurboServerOptions(), ops);
+
+        sm.DecodeClientData(MakeConnected());
+        sm.DecodeClientData(MakeData("GET / HTTP/1.1\r\nHost: localhost\r\nContent-Length: 0\r\n\r\n"u8.ToArray()));
+
+        // HTTP/1.1 responses are positional on the wire, so the negotiator must forward the inner
+        // machine's one-at-a-time dispatch limit — otherwise the shared, completion-ordered bridge
+        // can reorder pipelined responses (RFC 9112 §9.3.2).
+        Assert.Equal(1, sm.MaxConcurrentRequests);
+    }
+
+    [Fact(Timeout = 5000)]
+    public void MaxConcurrentRequests_should_stay_unbounded_for_negotiated_http2()
+    {
+        var ops = new FakeServerOps();
+        var sm = new ProtocolNegotiatingStateMachine(new TurboServerOptions(), ops);
+
+        sm.DecodeClientData(MakeConnected(SslApplicationProtocol.Http2));
+
+        // HTTP/2 routes responses to streams by id, so concurrent dispatch must remain unbounded.
+        Assert.Equal(int.MaxValue, sm.MaxConcurrentRequests);
+    }
+
     [Fact(Timeout = 5000)]
     public void Cleanup_should_dispose_buffered_data()
     {

src/TurboHTTP.Tests/Streams/Stages/Server/Http11ServerConnectionStagePipeliningSpec.cs

@@ -0,0 +1,106 @@
+using System.Collections.Concurrent;
+using System.Net;
+using System.Text;
+using Akka.Actor;
+using Akka.Streams.Dsl;
+using Akka.Streams.TestKit;
+using Microsoft.AspNetCore.Hosting.Server;
+using Microsoft.AspNetCore.Http.Features;
+using Servus.Akka.Transport;
+using TurboHTTP.Server;
+using TurboHTTP.Streams;
+using TurboHTTP.Streams.Stages.Server;
+using TurboHTTP.Tests.Shared;
+
+namespace TurboHTTP.Tests.Streams.Stages.Server;
+
+/// <summary>
+/// RFC 9112 §9.3.2: a server MAY process pipelined requests in parallel but MUST send the
+/// corresponding responses in the same order the requests were received. TurboHTTP guarantees
+/// this by dispatching pipelined HTTP/1.1 requests to the application handler strictly
+/// one-at-a-time, so the shared (completion-ordered) ApplicationBridgeStage can never reorder
+/// or corrupt responses on a single H1.1 connection.
+/// </summary>
+public sealed class Http11ServerConnectionStagePipeliningSpec : StreamTestBase
+{
+    private sealed class FakeApplication(Func<IFeatureCollection, Task> handler)
+        : IHttpApplication<IFeatureCollection>
+    {
+        public IFeatureCollection CreateContext(IFeatureCollection contextFeatures) => contextFeatures;
+        public Task ProcessRequestAsync(IFeatureCollection context) => handler(context);
+        public void DisposeContext(IFeatureCollection context, Exception? exception) { }
+    }
+
+    private static TransportConnected Connected()
+        => new(new ConnectionInfo(
+            new IPEndPoint(IPAddress.Loopback, 80),
+            new IPEndPoint(IPAddress.Loopback, 50000),
+            TransportProtocol.Tcp));
+
+    private static TransportData PipelinedRequests(params string[] paths)
+    {
+        var sb = new StringBuilder();
+        foreach (var path in paths)
+        {
+            sb.Append("GET ").Append(path).Append(" HTTP/1.1\r\nHost: example.com\r\n\r\n");
+        }
+
+        var bytes = Encoding.ASCII.GetBytes(sb.ToString());
+        var buffer = TransportBuffer.Rent(bytes.Length);
+        bytes.CopyTo(buffer.FullMemory.Span);
+        buffer.Length = bytes.Length;
+        return TransportData.Rent(buffer);
+    }
+
+    [Fact(Timeout = 15000)]
+    [Trait("RFC", "RFC9112-9.3.2")]
+    public void Http11_pipelined_requests_are_dispatched_one_at_a_time()
+    {
+        var options = new TurboServerOptions();
+        var gates = new ConcurrentDictionary<string, TaskCompletionSource>();
+        var probe = CreateTestProbe();
+
+        // Each handler invocation blocks on its own gate. If pipelined requests were dispatched
+        // concurrently, the probe would observe all three paths before any gate is released.
+        var app = new FakeApplication(features =>
+        {
+            var path = features.Get<IHttpRequestFeature>()!.Path;
+            var tcs = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+            gates[path] = tcs;
+            features.Get<IHttpResponseFeature>()!.StatusCode = 204;
+            probe.Ref.Tell(path, ActorRefs.NoSender);
+            return tcs.Task;
+        });
+
+        // Production wires the bridge with unbounded parallelism and routes connections through the
+        // negotiating engine; the connection stage — not the bridge — must enforce H1.1 ordering, so
+        // the test reproduces that exact path (negotiating engine + int.MaxValue bridge).
+        var bridge = Flow.FromGraph(new ApplicationBridgeStage<IFeatureCollection>(
+            app, int.MaxValue, options.HandlerTimeout, options.HandlerGracePeriod));
+
+        var joined = new NegotiatingServerEngine(options).CreateFlow().Join(bridge);
+
+        var (netIn, netOut) = this.SourceProbe<ITransportInbound>()
+            .Via(joined)
+            .ToMaterialized(this.SinkProbe<ITransportOutbound>(), Keep.Both)
+            .Run(Materializer);
+
+        netOut.Request(100);
+        netIn.SendNext(Connected(), TestContext.Current.CancellationToken);
+        netIn.SendNext(PipelinedRequests("/p/1", "/p/2", "/p/3"), TestContext.Current.CancellationToken);
+
+        var first = probe.ExpectMsg<string>();
+        Assert.Equal("/p/1", first);
+        probe.ExpectNoMsg(TimeSpan.FromMilliseconds(500));
+        gates[first].SetResult();
+
+        var second = probe.ExpectMsg<string>();
+        Assert.Equal("/p/2", second);
+        probe.ExpectNoMsg(TimeSpan.FromMilliseconds(500));
+        gates[second].SetResult();
+
+        var third = probe.ExpectMsg<string>();
+        Assert.Equal("/p/3", third);
+        gates[third].SetResult();
+    }
+}

src/TurboHTTP/Protocol/Body/QueuedBodyReader.cs

@@ -12,6 +12,7 @@ internal sealed class QueuedBodyReader : IStreamingBodyReader, IValueTaskSource<
     // never executes on the producing stage thread.
     private readonly object _sync = new();
 
+    private readonly ArrayPool<byte> _pool;
     private OwnedChunk[] _slots;
     private readonly int _backpressureThreshold;
     private int _head;
@@ -25,8 +26,9 @@ internal sealed class QueuedBodyReader : IStreamingBodyReader, IValueTaskSource<
 
     private readonly int _initialSlotCount;
 
-    public QueuedBodyReader(int capacity)
+    public QueuedBodyReader(int capacity, ArrayPool<byte>? pool = null)
     {
+        _pool = pool ?? ArrayPool<byte>.Shared;
         _backpressureThreshold = capacity;
         _initialSlotCount = capacity * 2;
         _slots = new OwnedChunk[_initialSlotCount];
@@ -61,7 +63,7 @@ public bool IsFull
 
     public bool TryEnqueue(ReadOnlySpan<byte> data)
     {
-        var rental = ArrayPool<byte>.Shared.Rent(data.Length);
+        var rental = _pool.Rent(data.Length);
         data.CopyTo(rental);
         var chunk = new OwnedChunk(rental, data.Length);
 
@@ -208,7 +210,7 @@ public void AdvanceTo()
         {
             if (_current.Rental is not null)
             {
-                ArrayPool<byte>.Shared.Return(_current.Rental);
+                _pool.Return(_current.Rental);
             }
 
             _current = default;
@@ -248,18 +250,22 @@ public void Reset()
                 _head = (_head + 1) % _slots.Length;
                 _count--;
 
+                // Queued chunks were never handed to the consumer, so their rentals are
+                // safe to return here.
                 if (chunk.Rental is not null)
                 {
-                    ArrayPool<byte>.Shared.Return(chunk.Rental);
+                    _pool.Return(chunk.Rental);
                 }
             }
 
-            if (_current.Rental is not null)
-            {
-                ArrayPool<byte>.Shared.Return(_current.Rental);
-            }
-
-            _current = default;
+            // Do NOT return _current's rental here. Once _current is non-default its Memory
+            // has been published to the consumer (via the last ReadAsync result) and may
+            // still be read on another thread. On teardown/abort this lock can run while
+            // that read is in flight; returning the rental would recycle a buffer still in
+            // use and let a concurrent stream overwrite it (cross-stream, correct-length
+            // wrong-content corruption). Ownership stays with the consumer: its AdvanceTo
+            // returns the rental exactly once. If the consumer abandons it, the array is
+            // simply GC-reclaimed rather than returned to the pool — a rare, bounded miss.
             _head = 0;
             _tail = 0;
             _count = 0;

src/TurboHTTP/Protocol/IServerStateMachine.cs

@@ -10,6 +10,15 @@ internal interface IServerStateMachine
     bool ShouldPauseNetwork => false;
     int MaxQueuedRequests { get; }
 
+    /// <summary>
+    /// Maximum number of requests that may be dispatched to the application handler concurrently
+    /// on this connection. HTTP/1.x returns 1 so the connection stage serializes handler dispatch
+    /// and the shared (completion-ordered) <c>ApplicationBridgeStage</c> can never reorder responses
+    /// — RFC 9112 §9.3.2 requires pipelined responses in request order. Multiplexed protocols
+    /// (HTTP/2, HTTP/3) route responses by stream id and leave this unbounded.
+    /// </summary>
+    int MaxConcurrentRequests => int.MaxValue;
+
     void PreStart();
     void OnResponse(IFeatureCollection features);
     void DecodeClientData(ITransportInbound data);

src/TurboHTTP/Protocol/ProtocolNegotiatingStateMachine.cs

@@ -27,6 +27,11 @@ private enum Phase { WaitingForConnect, Sniffing, Running }
     public bool ShouldComplete => _phase == Phase.Running && _inner!.ShouldComplete;
     public int MaxQueuedRequests => _phase == Phase.Running ? _inner!.MaxQueuedRequests : 1;
 
+    // Forward the concurrency limit so a negotiated/sniffed HTTP/1.x connection still serializes
+    // handler dispatch (RFC 9112 §9.3.2). Until a protocol is chosen no request is dispatched, so
+    // the conservative default of 1 is safe.
+    public int MaxConcurrentRequests => _phase == Phase.Running ? _inner!.MaxConcurrentRequests : 1;
+
     public ProtocolNegotiatingStateMachine(TurboServerOptions options, IServerStageOperations ops)
     {
         _options = options;

src/TurboHTTP/Protocol/Syntax/Http10/Server/Http10ServerStateMachine.cs

@@ -44,6 +44,9 @@ internal sealed class Http10ServerStateMachine : IServerStateMachine
 
     public int MaxQueuedRequests => 1;
 
+    // HTTP/1.0 dispatches one request per connection; mirror H1.1 so handler dispatch stays serial.
+    public int MaxConcurrentRequests => 1;
+
     public Http10ServerStateMachine(Http1ConnectionOptions options, IServerStageOperations ops,
         TimeProvider? timeProvider = null)
     {

src/TurboHTTP/Protocol/Syntax/Http11/Server/Http11ServerStateMachine.cs

@@ -63,6 +63,11 @@
     public bool ShouldPauseNetwork => _activeStreamingReader?.IsFull ?? false;
     public int MaxQueuedRequests { get; }
 
+    // HTTP/1.1 responses are matched to requests by position on the wire, so a pipelined request
+    // must not be dispatched to the handler until the previous response has been emitted
+    // (RFC 9112 §9.3.2). One-at-a-time dispatch keeps the shared bridge from reordering responses.
+    public int MaxConcurrentRequests => 1;
+
     public Http11ServerStateMachine(Http1ConnectionOptions options, Http2ConnectionOptions h2UpgradeOptions,
         IServerStageOperations ops, TimeProvider? timeProvider = null)
     {
@@ -397,7 +402,7 @@
             while (remaining.Length > 0)
             {
                 var take = Math.Min(remaining.Length, _bodyEncoderOptions.ChunkSize);
                 var dest = writer.GetMemory(take);
                 remaining.Span[..take].CopyTo(dest.Span);
                 writer.Advance(take);
                 writer.FlushAsync();
@@ -405,7 +410,7 @@
             }
         }
 
         writer.CompleteAsync();
         // The response is fully handed to the transport: drop the rate entry, or the idle
         // keep-alive connection is flagged as a violation once the grace period elapses.
         _responseRate.Remove(0);