HTTP/3: Request and response data rates

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

@@ -133,6 +133,11 @@ public void StopProcessingNextRequest(bool serverInitiated)
 
                 if (Interlocked.CompareExchange(ref _gracefulCloseInitiator, initiator, GracefulCloseInitiator.None) == GracefulCloseInitiator.None)
                 {
+                    // https://quicwg.org/base-drafts/draft-ietf-quic-http.html#section-5.2-11
+                    // An endpoint that completes a graceful shutdown SHOULD use the H3_NO_ERROR error code
+                    // when closing the connection.
+                    _errorCodeFeature.Error = (long)Http3ErrorCode.NoError;
+
                     // Abort accept async loop to initiate graceful shutdown
                     // TODO aborting connection isn't graceful due to runtime issue, will drop data on streams
                     // Either we need to swap to using a cts here or fix runtime to gracefully close connection.
@@ -249,10 +254,15 @@ public void OnTimeout(TimeoutReason reason)
                 case TimeoutReason.TimeoutFeature:
                     SendGoAway(GetHighestStreamId()).Preserve();
                     break;
-                case TimeoutReason.RequestHeaders: // Request header timeout is handled in starting stream queue
-                case TimeoutReason.KeepAlive:  // Keep-alive is handled by msquic
                 case TimeoutReason.ReadDataRate:
+                    Abort(new ConnectionAbortedException(CoreStrings.BadRequest_RequestBodyTimeout), Http3ErrorCode.InternalError);
+                    break;
                 case TimeoutReason.WriteDataRate:
+                    Log.ResponseMinimumDataRateNotSatisfied(_context.ConnectionId, traceIdentifier: null);
+                    Abort(new ConnectionAbortedException(CoreStrings.ConnectionTimedBecauseResponseMininumDataRateNotSatisfied), Http3ErrorCode.InternalError);
+                    break;
+                case TimeoutReason.RequestHeaders: // Request header timeout is handled in starting stream queue
+                case TimeoutReason.KeepAlive:  // Keep-alive is handled by msquic
                 case TimeoutReason.RequestBodyDrain:
                 default:
                     Debug.Assert(false, "Invalid TimeoutReason");
@@ -392,6 +402,9 @@ internal async Task InnerProcessStreamsAsync<TContext>(IHttpApplication<TContext
                     {
                         await _streamCompletionAwaitable;
                     }
+
+                    _timeoutControl.CancelTimeout();
+                    _timeoutControl.StartDrainTimeout(Limits.MinResponseDataRate, Limits.MaxResponseBufferSize);
                 }
                 catch
                 {

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

@@ -4,6 +4,7 @@
 using System;
 using System.Buffers;
 using System.Collections.Generic;
+using System.Diagnostics;
 using System.Globalization;
 using System.IO.Pipelines;
 using System.Net.Http;

src/Servers/Kestrel/Core/src/Internal/Infrastructure/TimeoutControl.cs

@@ -81,9 +81,19 @@ private void CheckForReadDataRateTimeout(long timestamp)
                 return;
             }
 
+            // HTTP/2
             // Don't enforce the rate timeout if there is back pressure due to HTTP/2 connection-level input
             // flow control. We don't consider stream-level flow control, because we wouldn't be timing a read
             // for any stream that didn't have a completely empty stream-level flow control window.
+            //
+            // HTTP/3
+            // This isn't (currently) checked. Reasons:
+            // - We're not sure how often people in the real-world run into this. If it
+            //   becomes a problem then we'll need to revisit.
+            // - There isn't a way to get this information easily and efficently from msquic.
+            // - With QUIC, bytes can be received out of order. The connection window could
+            //   be filled up out of order so that availablility is low but there is still
+            //   no data available to use. Would need a smarter way to handle this situation.
             if (_connectionInputFlowControl?.IsAvailabilityLow == true)
             {
                 return;

src/Servers/Kestrel/Kestrel.slnf

@@ -5,6 +5,8 @@
       "src\\Hosting\\Abstractions\\src\\Microsoft.AspNetCore.Hosting.Abstractions.csproj",
       "src\\Hosting\\Hosting\\src\\Microsoft.AspNetCore.Hosting.csproj",
       "src\\Hosting\\Server.Abstractions\\src\\Microsoft.AspNetCore.Hosting.Server.Abstractions.csproj",
+      "src\\Http\\Features\\src\\Microsoft.Extensions.Features.csproj",
+      "src\\Http\\Features\\test\\Microsoft.Extensions.Features.Tests.csproj",
       "src\\Http\\Headers\\src\\Microsoft.Net.Http.Headers.csproj",
       "src\\Http\\Http.Abstractions\\src\\Microsoft.AspNetCore.Http.Abstractions.csproj",
       "src\\Http\\Http.Extensions\\src\\Microsoft.AspNetCore.Http.Extensions.csproj",

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

@@ -61,12 +61,19 @@ await InitializeConnectionAsync(async context =>
         }
 
         [Fact]
-        public async Task GracefulServerShutdownSendsGoawayClosesConnection()
+        public async Task GracefulServerShutdownClosesConnection()
         {
             await InitializeConnectionAsync(_echoApplication);
+
+            var inboundControlStream = await GetInboundControlStream();
+            await inboundControlStream.ExpectSettingsAsync();
+
             // Trigger server shutdown.
-            MultiplexedConnectionContext.ConnectionClosingCts.Cancel();
+            CloseConnectionGracefully();
+
             Assert.Null(await MultiplexedConnectionContext.AcceptAsync().DefaultTimeout());
+
+            await WaitForConnectionStopAsync(0, false, expectedErrorCode: Http3ErrorCode.NoError);
         }
 
         [Theory]

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

@@ -6,12 +6,14 @@
 using System.Collections.Concurrent;
 using System.Collections.Generic;
 using System.Diagnostics;
+using System.Globalization;
 using System.IO;
 using System.IO.Pipelines;
 using System.Linq;
 using System.Net.Http;
 using System.Net.Http.QPack;
 using System.Reflection;
+using System.Text;
 using System.Threading;
 using System.Threading.Channels;
 using System.Threading.Tasks;
@@ -38,6 +40,8 @@ public abstract class Http3TestBase : TestApplicationErrorLoggerLoggedTest, IDis
     {
         protected static readonly int MaxRequestHeaderFieldSize = 16 * 1024;
         protected static readonly string _4kHeaderValue = new string('a', 4096);
+        protected static readonly byte[] _helloWorldBytes = Encoding.ASCII.GetBytes("hello, world");
+        protected static readonly byte[] _maxData = Encoding.ASCII.GetBytes(new string('a', 16 * 1024));
 
         internal TestServiceContext _serviceContext;
         internal readonly TimeoutControl _timeoutControl;
@@ -49,16 +53,37 @@ public abstract class Http3TestBase : TestApplicationErrorLoggerLoggedTest, IDis
         protected readonly TaskCompletionSource _closedStateReached = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
 
         internal readonly ConcurrentDictionary<long, Http3StreamBase> _runningStreams = new ConcurrentDictionary<long, Http3StreamBase>();
-
         protected readonly RequestDelegate _noopApplication;
         protected readonly RequestDelegate _echoApplication;
+        protected readonly RequestDelegate _readRateApplication;
         protected readonly RequestDelegate _echoMethod;
         protected readonly RequestDelegate _echoPath;
         protected readonly RequestDelegate _echoHost;
 
         private Http3ControlStream _inboundControlStream;
         private long _currentStreamId;
 
+        protected static readonly IEnumerable<KeyValuePair<string, string>> _browserRequestHeaders = new[]
+        {
+            new KeyValuePair<string, string>(HeaderNames.Method, "GET"),
+            new KeyValuePair<string, string>(HeaderNames.Path, "/"),
+            new KeyValuePair<string, string>(HeaderNames.Scheme, "http"),
+            new KeyValuePair<string, string>(HeaderNames.Authority, "localhost:80"),
+            new KeyValuePair<string, string>("user-agent", "Mozilla/5.0 (X11; Ubuntu; Linux x86_64; rv:54.0) Gecko/20100101 Firefox/54.0"),
+            new KeyValuePair<string, string>("accept", "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8"),
+            new KeyValuePair<string, string>("accept-language", "en-US,en;q=0.5"),
+            new KeyValuePair<string, string>("accept-encoding", "gzip, deflate, br"),
+            new KeyValuePair<string, string>("upgrade-insecure-requests", "1"),
+        };
+
+        protected static IEnumerable<KeyValuePair<string, string>> ReadRateRequestHeaders(int expectedBytes) => new[]
+        {
+            new KeyValuePair<string, string>(HeaderNames.Method, "POST"),
+            new KeyValuePair<string, string>(HeaderNames.Path, "/" + expectedBytes),
+            new KeyValuePair<string, string>(HeaderNames.Scheme, "http"),
+            new KeyValuePair<string, string>(HeaderNames.Authority, "localhost:80"),
+        };
+
         public Http3TestBase()
         {
             _timeoutControl = new TimeoutControl(_mockTimeoutHandler.Object);
@@ -83,6 +108,26 @@ public Http3TestBase()
                 }
             };
 
+            _readRateApplication = async context =>
+            {
+                var expectedBytes = int.Parse(context.Request.Path.Value.Substring(1), CultureInfo.InvariantCulture);
+
+                var buffer = new byte[16 * 1024];
+                var received = 0;
+
+                while (received < expectedBytes)
+                {
+                    received += await context.Request.Body.ReadAsync(buffer, 0, buffer.Length);
+                }
+
+                var stalledReadTask = context.Request.Body.ReadAsync(buffer, 0, buffer.Length);
+
+                // Write to the response so the test knows the app started the stalled read.
+                await context.Response.Body.WriteAsync(new byte[1], 0, 1);
+
+                await stalledReadTask;
+            };
+
             _echoMethod = context =>
             {
                 context.Response.Headers["Method"] = context.Request.Method;
@@ -151,7 +196,17 @@ internal async ValueTask<Http3ControlStream> GetInboundControlStream()
             return _inboundControlStream;
         }
 
-        internal async Task WaitForConnectionErrorAsync<TException>(bool ignoreNonGoAwayFrames, long expectedLastStreamId, Http3ErrorCode expectedErrorCode, params string[] expectedErrorMessage)
+        internal void CloseConnectionGracefully()
+        {
+            MultiplexedConnectionContext.ConnectionClosingCts.Cancel();
+        }
+
+        internal Task WaitForConnectionStopAsync(long expectedLastStreamId, bool ignoreNonGoAwayFrames, Http3ErrorCode? expectedErrorCode = null)
+        {
+            return WaitForConnectionErrorAsync<Exception>(ignoreNonGoAwayFrames, expectedLastStreamId, expectedErrorCode: expectedErrorCode ?? 0, expectedErrorMessage: null);
+        }
+
+        internal async Task WaitForConnectionErrorAsync<TException>(bool ignoreNonGoAwayFrames, long? expectedLastStreamId, Http3ErrorCode expectedErrorCode, params string[] expectedErrorMessage)
             where TException : Exception
         {
             var frame = await _inboundControlStream.ReceiveFrameAsync();
@@ -164,7 +219,10 @@ internal async Task WaitForConnectionErrorAsync<TException>(bool ignoreNonGoAway
                 }
             }
 
-            VerifyGoAway(frame, expectedLastStreamId);
+            if (expectedLastStreamId != null)
+            {
+                VerifyGoAway(frame, expectedLastStreamId.GetValueOrDefault());
+            }
 
             Assert.Equal((Http3ErrorCode)expectedErrorCode, (Http3ErrorCode)MultiplexedConnectionContext.Error);
 
@@ -216,8 +274,6 @@ protected async Task InitializeConnectionAsync(RequestDelegate application)
             _connectionTask = Connection.ProcessStreamsAsync(new DummyApplication(application));
 
             await GetInboundControlStream();
-
-            await Task.CompletedTask;
         }
 
         internal async ValueTask<Http3RequestStream> InitializeConnectionAndStreamsAsync(RequestDelegate application)
@@ -394,7 +450,7 @@ public class Http3StreamBase : IProtocolErrorCodeFeature
             internal DuplexPipe.DuplexPipePair _pair;
             internal Http3TestBase _testBase;
             internal Http3Connection _connection;
-            private long _bytesReceived;
+            public long BytesReceived { get; private set; }
             public long Error { get; set; }
 
             public Task OnStreamCreatedTask => _onStreamCreatedTcs.Task;
@@ -412,6 +468,12 @@ protected static async Task FlushAsync(PipeWriter writableBuffer)
                 await writableBuffer.FlushAsync().AsTask().DefaultTimeout();
             }
 
+            internal async Task ReceiveEndAsync()
+            {
+                var result = await _pair.Application.Input.ReadAsync().AsTask().DefaultTimeout();
+                Assert.True(result.IsCompleted);
+            }
+
             internal async Task<Http3FrameWithPayload> ReceiveFrameAsync()
             {
                 var frame = new Http3FrameWithPayload();
@@ -446,7 +508,7 @@ internal async Task<Http3FrameWithPayload> ReceiveFrameAsync()
                     }
                     finally
                     {
-                        _bytesReceived += copyBuffer.Slice(copyBuffer.Start, consumed).Length;
+                        BytesReceived += copyBuffer.Slice(copyBuffer.Start, consumed).Length;
                         _pair.Application.Input.AdvanceTo(consumed, examined);
                     }
                 }
@@ -655,6 +717,17 @@ void WriteSpan(PipeWriter pw)
                 await FlushAsync(writableBuffer);
             }
 
+            internal async Task SendGoAwayAsync(long streamId, bool endStream = false)
+            {
+                var frame = new Http3RawFrame();
+                frame.PrepareGoAway();
+
+                var data = new byte[VariableLengthIntegerHelper.GetByteCount(streamId)];
+                VariableLengthIntegerHelper.WriteInteger(data, streamId);
+
+                await SendFrameAsync(frame, data, endStream);
+            }
+
             internal async Task SendSettingsAsync(List<Http3PeerSetting> settings, bool endStream = false)
             {
                 var frame = new Http3RawFrame();
@@ -738,6 +811,7 @@ public class TestMultiplexedConnectionContext : MultiplexedConnectionContext, IC
             });
 
             private readonly Http3TestBase _testBase;
+            private long _error;
 
             public TestMultiplexedConnectionContext(Http3TestBase testBase)
             {
@@ -759,7 +833,11 @@ public TestMultiplexedConnectionContext(Http3TestBase testBase)
 
             public CancellationTokenSource ConnectionClosingCts { get; set; } = new CancellationTokenSource();
 
-            public long Error { get; set; }
+            public long Error
+            {
+                get => _error;
+                set => _error = value;
+            }
 
             public override void Abort()
             {