Merged PR 54041: Fix Http3 Encoder/Decoder stream exit condition

Author: BrennanConroy

src/Servers/Kestrel/Core/src/Internal/Http3/Http3ControlStream.cs

@@ -65,6 +65,12 @@ public Http3ControlStream(Http3StreamContext context, long? headerType)
             context.ClientPeerSettings,
             this);
         _frameWriter.Reset(context.Transport.Output, context.ConnectionId);
+
+        _streamClosedFeature.OnClosed(static state =>
+        {
+            var stream = (Http3ControlStream)state!;
+            stream.OnStreamClosed();
+        }, this);
     }
 
     private void OnStreamClosed()
@@ -135,12 +141,6 @@ private bool TryClose()
 
     internal async ValueTask ProcessOutboundSendsAsync(long id)
     {
-        _streamClosedFeature.OnClosed(static state =>
-        {
-            var stream = (Http3ControlStream)state!;
-            stream.OnStreamClosed();
-        }, this);
-
         await _frameWriter.WriteStreamIdAsync(id);
         await _frameWriter.WriteSettingsAsync(_serverPeerSettings.GetNonProtocolDefaults());
     }
@@ -311,18 +311,13 @@ private async Task HandleControlStream()
         }
     }
 
-    private async ValueTask HandleEncodingDecodingTask()
+    private Task HandleEncodingDecodingTask()
     {
         // Noop encoding and decoding task. Settings make it so we don't need to read content of encoder and decoder.
         // An endpoint MUST allow its peer to create an encoder stream and a
         // decoder stream even if the connection's settings prevent their use.
 
-        while (_isClosed == 0)
-        {
-            var result = await Input.ReadAsync();
-            var readableBuffer = result.Buffer;
-            Input.AdvanceTo(readableBuffer.End);
-        }
+        return Input.CopyToAsync(Stream.Null);
     }
 
     private ValueTask ProcessHttp3ControlStream(Http3RawFrame incomingFrame, bool isContinuedFrame, in ReadOnlySequence<byte> payload, out SequencePosition consumed)
@@ -372,11 +367,6 @@ private ValueTask ProcessSettingsFrameAsync(bool isContinuedFrame, ReadOnlySeque
             }
 
             _haveReceivedSettingsFrame = true;
-            _streamClosedFeature.OnClosed(static state =>
-            {
-                var stream = (Http3ControlStream)state!;
-                stream.OnStreamClosed();
-            }, this);
         }
 
         while (true)

src/Servers/Kestrel/shared/test/Http3/Http3InMemory.cs

@@ -301,12 +301,22 @@ public void OnInboundControlStreamSetting(Http3SettingType type, long value)
 
         public bool OnInboundDecoderStream(Server.Kestrel.Core.Internal.Http3.Http3ControlStream stream)
         {
-            return _inner.OnInboundDecoderStream(stream);
+            var res = _inner.OnInboundDecoderStream(stream);
+            if (_http3TestBase._runningStreams.TryGetValue(stream.StreamId, out var testStream))
+            {
+                testStream.OnDecoderStreamCreatedTcs.TrySetResult();
+            }
+            return res;
         }
 
         public bool OnInboundEncoderStream(Server.Kestrel.Core.Internal.Http3.Http3ControlStream stream)
         {
-            return _inner.OnInboundEncoderStream(stream);
+            var res = _inner.OnInboundEncoderStream(stream);
+            if (_http3TestBase._runningStreams.TryGetValue(stream.StreamId, out var testStream))
+            {
+                testStream.OnEncoderStreamCreatedTcs.TrySetResult();
+            }
+            return res;
         }
 
         public void OnStreamCompleted(IHttp3Stream stream)
@@ -480,6 +490,8 @@ internal class Http3StreamBase
     internal TaskCompletionSource OnStreamCreatedTcs { get; } = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
     internal TaskCompletionSource OnStreamCompletedTcs { get; } = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
     internal TaskCompletionSource OnHeaderReceivedTcs { get; } = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+    internal TaskCompletionSource OnDecoderStreamCreatedTcs { get; } = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+    internal TaskCompletionSource OnEncoderStreamCreatedTcs { get; } = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
 
     internal TestStreamContext StreamContext { get; }
     internal DuplexPipe.DuplexPipePair Pair { get; }
@@ -496,6 +508,8 @@ public long Error
     public Task OnStreamCreatedTask => OnStreamCreatedTcs.Task;
     public Task OnStreamCompletedTask => OnStreamCompletedTcs.Task;
     public Task OnHeaderReceivedTask => OnHeaderReceivedTcs.Task;
+    public Task OnDecoderStreamCreatedTask => OnDecoderStreamCreatedTcs.Task;
+    public Task OnEncoderStreamCreatedTask => OnEncoderStreamCreatedTcs.Task;
 
     public ConnectionAbortedException AbortReadException => StreamContext.AbortReadException;
     public ConnectionAbortedException AbortWriteException => StreamContext.AbortWriteException;

src/Servers/Kestrel/test/InMemory.FunctionalTests/Http3/Http3ConnectionTests.cs

@@ -18,7 +18,6 @@
 using Microsoft.Extensions.Primitives;
 using Microsoft.Net.Http.Headers;
 using Http3SettingType = Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http3.Http3SettingType;
-using Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Infrastructure;
 
 namespace Microsoft.AspNetCore.Server.Kestrel.Core.Tests;
 
@@ -768,6 +767,59 @@ public async Task ErrorCodeIsValidOnConnectionTimeout()
         Assert.InRange(errorCodeFeature.Error, 0, (1L << 62) - 1); // Valid range for HTTP/3 error codes
     }
 
+    [Theory]
+    [InlineData(2)] // encoder
+    [InlineData(3)] // decoder
+    public async Task IgnoredControlStreams_CloseConnectionOnEndStream(int streamType)
+    {
+        await Http3Api.InitializeConnectionAsync(_noopApplication);
+
+        var stream = await Http3Api.CreateControlStream(streamType);
+
+        // PipeWriter will be completed when end of stream is received. Should exit read loop and close stream
+        // which will cause the connection to close with an error.
+        await stream.SendFrameAsync(Http3FrameType.Data, Memory<byte>.Empty, endStream: true);
+
+        await stream.OnStreamCompletedTask.DefaultTimeout();
+
+        Http3Api.TriggerTick();
+        Http3Api.TriggerTick(TimeSpan.FromSeconds(1));
+
+        await Http3Api.WaitForConnectionErrorAsync<Http3ConnectionErrorException>(
+            ignoreNonGoAwayFrames: true,
+            expectedLastStreamId: 0,
+            expectedErrorCode: Http3ErrorCode.ClosedCriticalStream,
+            matchExpectedErrorMessage: AssertExpectedErrorMessages,
+            expectedErrorMessage: CoreStrings.Http3ErrorControlStreamClosed);
+        MetricsAssert.Equal(ConnectionEndReason.ClosedCriticalStream, Http3Api.ConnectionTags);
+    }
+
+    [Theory]
+    [InlineData(2)] // encoder
+    [InlineData(3)] // decoder
+    public async Task IgnoredControlStreams_CloseConnectionOnStreamClose(int streamType)
+    {
+        await Http3Api.InitializeConnectionAsync(_noopApplication);
+
+        var stream = await Http3Api.CreateControlStream(streamType);
+
+        await (streamType == 2 ? stream.OnEncoderStreamCreatedTask : stream.OnDecoderStreamCreatedTask).DefaultTimeout();
+
+        // Simulate quic layer closing the stream
+        stream.StreamContext.Close();
+
+        Http3Api.TriggerTick();
+        Http3Api.TriggerTick(TimeSpan.FromSeconds(1));
+
+        await Http3Api.WaitForConnectionErrorAsync<Http3ConnectionErrorException>(
+            ignoreNonGoAwayFrames: true,
+            expectedLastStreamId: 0,
+            expectedErrorCode: Http3ErrorCode.ClosedCriticalStream,
+            matchExpectedErrorMessage: AssertExpectedErrorMessages,
+            expectedErrorMessage: CoreStrings.Http3ErrorControlStreamClosed);
+        MetricsAssert.Equal(ConnectionEndReason.ClosedCriticalStream, Http3Api.ConnectionTags);
+    }
+
     private sealed class ThrowingMultiplexedConnectionContext : TestMultiplexedConnectionContext
     {
         private int _skipCount;