HTTP/3: Fix incorrectly pooling aborted streams (#35441)

Author: JamesNK

src/Servers/Kestrel/Transport.Quic/src/Internal/QuicStreamContext.cs

@@ -449,11 +449,15 @@ private void ShutdownWrite(Exception? shutdownReason)
 
         public override async ValueTask DisposeAsync()
         {
+            // Be conservative about what can be pooled.
+            // Only pool bidirectional streams whose pipes have completed successfully and haven't been aborted.
             CanReuse = _stream.CanRead && _stream.CanWrite
                 && _transportPipeReader.IsCompletedSuccessfully
                 && _transportPipeWriter.IsCompletedSuccessfully
                 && !_clientAbort
-                && !_serverAborted;
+                && !_serverAborted
+                && _shutdownReadReason == null
+                && _shutdownWriteReason == null;
 
             _originalTransport.Input.Complete();
             _originalTransport.Output.Complete();

src/Servers/Kestrel/Transport.Quic/test/QuicConnectionListenerTests.cs

@@ -115,8 +115,7 @@ public async Task ClientCertificate_Required_NotSent_ConnectionAborted()
 
             // https://github.com/dotnet/runtime/issues/57246 The accept still completes even though the connection was rejected, but it's already failed.
             var serverContext = await connectionListener.AcceptAndAddFeatureAsync().DefaultTimeout();
-            qex = await Assert.ThrowsAsync<QuicException>(() => serverContext.ConnectAsync().DefaultTimeout());
-            Assert.Equal("Failed to open stream to peer. Error Code: INVALID_STATE", qex.Message);
+            await Assert.ThrowsAsync<QuicException>(() => serverContext.ConnectAsync().DefaultTimeout());
         }
     }
 }

src/Servers/Kestrel/Transport.Quic/test/QuicStreamContextTests.cs

@@ -49,6 +49,52 @@ public async Task BidirectionalStream_ServerReadsDataAndCompletes_GracefullyClos
             Assert.Contains(TestSink.Writes, m => m.Message.Contains(@"shutting down writes because: ""The QUIC transport's send loop completed gracefully.""."));
         }
 
+        [ConditionalFact]
+        [MsQuicSupported]
+        public async Task BidirectionalStream_ReadAborted_NotPooled()
+        {
+            // Arrange
+            await using var connectionListener = await QuicTestHelpers.CreateConnectionListenerFactory(LoggerFactory);
+
+            var options = QuicTestHelpers.CreateClientConnectionOptions(connectionListener.EndPoint);
+            using var clientConnection = new QuicConnection(QuicImplementationProviders.MsQuic, options);
+            await clientConnection.ConnectAsync().DefaultTimeout();
+
+            await using var serverConnection = await connectionListener.AcceptAndAddFeatureAsync().DefaultTimeout();
+
+            // Act
+            var clientStream = clientConnection.OpenBidirectionalStream();
+            await clientStream.WriteAsync(TestData).DefaultTimeout();
+            var serverStream = await serverConnection.AcceptAsync().DefaultTimeout();
+            var readResult = await serverStream.Transport.Input.ReadAtLeastAsync(TestData.Length).DefaultTimeout();
+            serverStream.Transport.Input.AdvanceTo(readResult.Buffer.End);
+
+            await clientStream.WriteAsync(TestData).DefaultTimeout();
+
+            // Complete writing.
+            await serverStream.Transport.Output.CompleteAsync();
+
+            // Abort read-side of the stream and then complete pipe.
+            // This simulates what Kestrel does when a request finishes without
+            // reading the request body to the end.
+            serverStream.Features.Get<IStreamAbortFeature>().AbortRead((long)Http3ErrorCode.NoError, new ConnectionAbortedException("Test message."));
+            await serverStream.Transport.Input.CompleteAsync();
+
+            var quicStreamContext = Assert.IsType<QuicStreamContext>(serverStream);
+
+            // Both send and receive loops have exited.
+            await quicStreamContext._processingTask.DefaultTimeout();
+            Assert.True(quicStreamContext.CanWrite);
+            Assert.True(quicStreamContext.CanRead);
+
+            await quicStreamContext.DisposeAsync();
+
+            var quicConnectionContext = Assert.IsType<QuicConnectionContext>(serverConnection);
+
+            // Assert
+            Assert.Equal(0, quicConnectionContext.StreamPool.Count);
+        }
+
         [ConditionalTheory]
         [MsQuicSupported]
         [InlineData(1024)]

src/Servers/Kestrel/test/Interop.FunctionalTests/Http3/Http3RequestTests.cs

@@ -672,6 +672,69 @@ public async Task GET_ClientCancellationAfterResponseHeaders_RequestAbortRaised(
             }
         }
 
+        [ConditionalFact]
+        [MsQuicSupported]
+        public async Task StreamResponseContent_DelayAndTrailers_ClientSuccess()
+        {
+            // Arrange
+            var builder = CreateHostBuilder(async context =>
+            {
+                var feature = context.Features.Get<IHttpResponseTrailersFeature>();
+
+                for (var i = 1; i < 200; i++)
+                {
+                    feature.Trailers.Append($"trailer-{i}", new string('!', i));
+                }
+
+                Logger.LogInformation($"Server trailer count: {feature.Trailers.Count}");
+
+                await context.Request.BodyReader.ReadAtLeastAsync(TestData.Length);
+
+                for (var i = 0; i < 3; i++)
+                {
+                    await context.Response.BodyWriter.WriteAsync(TestData);
+
+                    await Task.Delay(TimeSpan.FromMilliseconds(10));
+                }
+            });
+
+            using (var host = builder.Build())
+            using (var client = Http3Helpers.CreateClient())
+            {
+                await host.StartAsync();
+
+                // Act
+                var request = new HttpRequestMessage(HttpMethod.Post, $"https://127.0.0.1:{host.GetPort()}/");
+                request.Content = new ByteArrayContent(TestData);
+                request.Version = HttpVersion.Version30;
+                request.VersionPolicy = HttpVersionPolicy.RequestVersionExact;
+
+                var response = await client.SendAsync(request, CancellationToken.None);
+                response.EnsureSuccessStatusCode();
+
+                var responseStream = await response.Content.ReadAsStreamAsync();
+
+                await responseStream.ReadUntilEndAsync();
+
+                Logger.LogInformation($"Client trailer count: {response.TrailingHeaders.Count()}");
+
+                for (var i = 1; i < 200; i++)
+                {
+                    try
+                    {
+                        var value = response.TrailingHeaders.GetValues($"trailer-{i}").Single();
+                        Assert.Equal(new string('!', i), value);
+                    }
+                    catch (Exception ex)
+                    {
+                        throw new Exception($"Error checking trailer {i}", ex);
+                    }
+                }
+
+                await host.StopAsync();
+            }
+        }
+
         [ConditionalFact]
         [MsQuicSupported]
         public async Task GET_MultipleRequests_ConnectionAndTraceIdsUpdated()