diff --git a/src/Kestrel.Core/Internal/ConnectionDispatcher.cs b/src/Kestrel.Core/Internal/ConnectionDispatcher.cs
index 780ede5a35..f6b77155bf 100644
--- a/src/Kestrel.Core/Internal/ConnectionDispatcher.cs
+++ b/src/Kestrel.Core/Internal/ConnectionDispatcher.cs
@@ -143,7 +143,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal
             var bufferSize = serviceContext.ServerOptions.Limits.MaxResponseBufferSize;
             if (bufferSize == 0)
             {
-                // 0 = no buffering so we need to configure the pipe so the the writer waits on the reader directly
+                // 0 = no buffering so we need to configure the pipe so the writer waits on the reader directly
                 return 1;
             }
 
diff --git a/src/Kestrel.Core/Internal/Http/Http1Connection.cs b/src/Kestrel.Core/Internal/Http/Http1Connection.cs
index 68554aff49..26ec08808e 100644
--- a/src/Kestrel.Core/Internal/Http/Http1Connection.cs
+++ b/src/Kestrel.Core/Internal/Http/Http1Connection.cs
@@ -518,6 +518,8 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http
             }
         }
 
+        void IRequestProcessor.Tick(DateTimeOffset now) { }
+
         private Pipe CreateRequestBodyPipe()
             => new Pipe(new PipeOptions
             (
diff --git a/src/Kestrel.Core/Internal/Http2/Http2Connection.cs b/src/Kestrel.Core/Internal/Http2/Http2Connection.cs
index d744728714..fcbb3f193f 100644
--- a/src/Kestrel.Core/Internal/Http2/Http2Connection.cs
+++ b/src/Kestrel.Core/Internal/Http2/Http2Connection.cs
@@ -84,6 +84,8 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http2
         private readonly TaskCompletionSource<object> _streamsCompleted = new TaskCompletionSource<object>(TaskCreationOptions.RunContinuationsAsynchronously);
 
         private readonly ConcurrentDictionary<int, Http2Stream> _streams = new ConcurrentDictionary<int, Http2Stream>();
+        private readonly ConcurrentDictionary<int, Http2Stream> _drainingStreams = new ConcurrentDictionary<int, Http2Stream>();
+        private int _activeStreamCount = 0;
 
         public Http2Connection(HttpConnectionContext context)
         {
@@ -153,7 +155,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http2
             {
                 if (_state == Http2ConnectionState.Open)
                 {
-                    if (_streams.IsEmpty)
+                    if (_activeStreamCount == 0)
                     {
                         _frameWriter.WriteGoAwayAsync(_highestOpenedStreamId, Http2ErrorCode.NO_ERROR);
                         UpdateState(Http2ConnectionState.Closed);
@@ -246,7 +248,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http2
             catch (ConnectionResetException ex)
             {
                 // Don't log ECONNRESET errors when there are no active streams on the connection. Browsers like IE will reset connections regularly.
-                if (_streams.Count > 0)
+                if (_activeStreamCount > 0)
                 {
                     Log.RequestProcessingError(ConnectionId, ex);
                 }
@@ -291,7 +293,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http2
                             UpdateState(Http2ConnectionState.Closed);
                         }
 
-                        if (_streams.IsEmpty)
+                        if (_activeStreamCount == 0)
                         {
                             _streamsCompleted.TrySetResult(null);
                         }
@@ -458,11 +460,12 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http2
 
             if (_streams.TryGetValue(_incomingFrame.StreamId, out var stream))
             {
-                if (stream.DoNotDrainRequest)
+                if (stream.RstStreamReceived)
                 {
                     // Hard abort, do not allow any more frames on this stream.
                     throw new Http2ConnectionErrorException(CoreStrings.FormatHttp2ErrorStreamAborted(_incomingFrame.Type, stream.StreamId), Http2ErrorCode.STREAM_CLOSED);
                 }
+
                 if (stream.EndStreamReceived)
                 {
                     // http://httpwg.org/specs/rfc7540.html#rfc.section.5.1
@@ -475,6 +478,12 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http2
                     throw new Http2ConnectionErrorException(CoreStrings.FormatHttp2ErrorStreamHalfClosedRemote(_incomingFrame.Type, stream.StreamId), Http2ErrorCode.STREAM_CLOSED);
                 }
 
+                if (_incomingFrame.DataEndStream && stream.IsDraining)
+                {
+                    // No more frames expected.
+                    RemoveDrainingStream(_incomingFrame.StreamId);
+                }
+
                 return stream.OnDataAsync(_incomingFrame, payload);
             }
 
@@ -516,7 +525,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http2
 
             if (_streams.TryGetValue(_incomingFrame.StreamId, out var stream))
             {
-                if (stream.DoNotDrainRequest)
+                if (stream.RstStreamReceived)
                 {
                     // Hard abort, do not allow any more frames on this stream.
                     throw new Http2ConnectionErrorException(CoreStrings.FormatHttp2ErrorStreamAborted(_incomingFrame.Type, stream.StreamId), Http2ErrorCode.STREAM_CLOSED);
@@ -650,15 +659,22 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http2
             if (_streams.TryGetValue(_incomingFrame.StreamId, out var stream))
             {
                 // Second reset
-                if (stream.DoNotDrainRequest)
+                if (stream.RstStreamReceived)
                 {
                     // Hard abort, do not allow any more frames on this stream.
                     throw new Http2ConnectionErrorException(CoreStrings.FormatHttp2ErrorStreamAborted(_incomingFrame.Type, stream.StreamId), Http2ErrorCode.STREAM_CLOSED);
                 }
 
-                // No additional inbound header or data frames are allowed for this stream after receiving a reset.
-                stream.DisallowAdditionalRequestFrames();
-                stream.Abort(new IOException(CoreStrings.Http2StreamResetByClient));
+                if (stream.IsDraining)
+                {
+                    // This stream was aborted by the server earlier and now the client is aborting it as well. No more frames are expected.
+                    RemoveDrainingStream(_incomingFrame.StreamId);
+                }
+                else
+                {
+                    // No additional inbound header or data frames are allowed for this stream after receiving a reset.
+                    stream.AbortRstStreamReceived();
+                }
             }
 
             return Task.CompletedTask;
@@ -821,7 +837,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http2
             }
             else if (_streams.TryGetValue(_incomingFrame.StreamId, out var stream))
             {
-                if (stream.DoNotDrainRequest)
+                if (stream.RstStreamReceived)
                 {
                     // Hard abort, do not allow any more frames on this stream.
                     throw new Http2ConnectionErrorException(CoreStrings.FormatHttp2ErrorStreamAborted(_incomingFrame.Type, stream.StreamId), Http2ErrorCode.STREAM_CLOSED);
@@ -917,7 +933,16 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http2
 
             if (endHeaders)
             {
-                _currentHeadersStream.OnEndStreamReceived();
+                if (_currentHeadersStream.IsDraining)
+                {
+                    // This stream is aborted and abandon, no action required
+                    RemoveDrainingStream(_currentHeadersStream.StreamId);
+                }
+                else
+                {
+                    _currentHeadersStream.OnEndStreamReceived();
+                }
+
                 ResetRequestHeaderParsingState();
             }
 
@@ -934,7 +959,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http2
                 throw new Http2StreamErrorException(_currentHeadersStream.StreamId, CoreStrings.Http2ErrorMissingMandatoryPseudoHeaderFields, Http2ErrorCode.PROTOCOL_ERROR);
             }
 
-            if (_streams.Count >= _serverSettings.MaxConcurrentStreams)
+            if (_activeStreamCount >= _serverSettings.MaxConcurrentStreams)
             {
                 throw new Http2StreamErrorException(_currentHeadersStream.StreamId, CoreStrings.Http2ErrorMaxStreams, Http2ErrorCode.REFUSED_STREAM);
             }
@@ -948,6 +973,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http2
                 _currentHeadersStream.OnEndStreamReceived();
             }
 
+            _activeStreamCount++;
             _streams[_incomingFrame.StreamId] = _currentHeadersStream;
             // Must not allow app code to block the connection handling loop.
             ThreadPool.UnsafeQueueUserWorkItem(state =>
@@ -999,9 +1025,29 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http2
         {
             lock (_stateLock)
             {
-                _streams.TryRemove(streamId, out _);
+                // Get, Add, Remove so the steam is always registered in at least one collection at a time.
+                if (_streams.TryGetValue(streamId, out var stream))
+                {
+                    _activeStreamCount--;
 
-                if (_streams.IsEmpty)
+                    if (stream.IsDraining)
+                    {
+                        stream.DrainExpiration =
+                            _context.ServiceContext.SystemClock.UtcNow + Constants.RequestBodyDrainTimeout;
+
+                        _drainingStreams.TryAdd(streamId, stream);
+                    }
+                    else
+                    {
+                        _streams.TryRemove(streamId, out _);
+                    }
+                }
+                else
+                {
+                    Debug.Assert(false, "Missing stream");
+                }
+
+                if (_activeStreamCount == 0)
                 {
                     if (_state == Http2ConnectionState.Closing)
                     {
@@ -1030,6 +1076,17 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http2
             }
         }
 
+        void IRequestProcessor.Tick(DateTimeOffset now)
+        {
+            foreach (var stream in _drainingStreams)
+            {
+                if (now > stream.Value.DrainExpiration)
+                {
+                    RemoveDrainingStream(stream.Key);
+                }
+            }
+        }
+
         // We can't throw a Http2StreamErrorException here, it interrupts the header decompression state and may corrupt subsequent header frames on other streams.
         // For now these either need to be connection errors or BadRequests. If we want to downgrade any of them to stream errors later then we need to
         // rework the flow so that the remaining headers are drained and the decompression state is maintained.
@@ -1211,5 +1268,15 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http2
                 Log.Http2ConnectionClosed(_context.ConnectionId, _highestOpenedStreamId);
             }
         }
+
+        // Note this may be called concurrently based on incoming frames and Ticks.
+        private void RemoveDrainingStream(int key)
+        {
+            _streams.TryRemove(key, out _);
+            // It's possible to be marked as draining and have RemoveDrainingStream called
+            // before being added to the draining collection. In that case the next Tick would
+            // remove it anyways.
+            _drainingStreams.TryRemove(key, out _);
+        }
     }
 }
diff --git a/src/Kestrel.Core/Internal/Http2/Http2FrameWriter.cs b/src/Kestrel.Core/Internal/Http2/Http2FrameWriter.cs
index cd8bc7ae48..abaea59245 100644
--- a/src/Kestrel.Core/Internal/Http2/Http2FrameWriter.cs
+++ b/src/Kestrel.Core/Internal/Http2/Http2FrameWriter.cs
@@ -35,6 +35,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http2
         private readonly OutputFlowControl _connectionOutputFlowControl;
         private readonly string _connectionId;
         private readonly IKestrelTrace _log;
+        private readonly ITimeoutControl _timeoutControl;
         private readonly TimingPipeFlusher _flusher;
 
         private bool _completed;
@@ -54,6 +55,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http2
             _connectionOutputFlowControl = connectionOutputFlowControl;
             _connectionId = connectionId;
             _log = log;
+            _timeoutControl = timeoutControl;
             _flusher = new TimingPipeFlusher(_outputWriter, timeoutControl);
             _outgoingFrame = new Http2Frame();
             _headerEncodingBuffer = new byte[_maxFrameSize];
@@ -226,7 +228,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http2
             }
         }
 
-        public Task WriteDataAsync(int streamId, StreamOutputFlowControl flowControl, ReadOnlySequence<byte> data, bool endStream)
+        public Task WriteDataAsync(int streamId, StreamOutputFlowControl flowControl, MinDataRate minRate, ReadOnlySequence<byte> data, bool endStream)
         {
             // The Length property of a ReadOnlySequence can be expensive, so we cache the value.
             var dataLength = data.Length;
@@ -242,12 +244,12 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http2
                 // https://httpwg.org/specs/rfc7540.html#rfc.section.6.9.1
                 if (dataLength != 0 && dataLength > flowControl.Available)
                 {
-                    return WriteDataAsyncAwaited(streamId, flowControl, data, dataLength, endStream);
+                    return WriteDataAsyncAwaited(streamId, minRate, flowControl, data, dataLength, endStream);
                 }
 
                 // This cast is safe since if dataLength would overflow an int, it's guaranteed to be greater than the available flow control window.
                 flowControl.Advance((int)dataLength);
-                return WriteDataUnsynchronizedAsync(streamId, data, endStream);
+                return WriteDataUnsynchronizedAsync(streamId, minRate, data, dataLength, endStream);
             }
         }
 
@@ -260,7 +262,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http2
             |                           Padding (*)                       ...
             +---------------------------------------------------------------+
         */
-        private Task WriteDataUnsynchronizedAsync(int streamId, ReadOnlySequence<byte> data, bool endStream)
+        private Task WriteDataUnsynchronizedAsync(int streamId, MinDataRate minRate, ReadOnlySequence<byte> data, long dataLength, bool endStream)
         {
             // Note padding is not implemented
             _outgoingFrame.PrepareData(streamId);
@@ -300,15 +302,16 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http2
 
             // Plus padding
 
-            return _flusher.FlushAsync();
+            return _flusher.FlushAsync(minRate, dataLength);
         }
 
-        private async Task WriteDataAsyncAwaited(int streamId, StreamOutputFlowControl flowControl, ReadOnlySequence<byte> data, long dataLength, bool endStream)
+        private async Task WriteDataAsyncAwaited(int streamId, MinDataRate minRate, StreamOutputFlowControl flowControl, ReadOnlySequence<byte> data, long dataLength, bool endStream)
         {
             while (dataLength > 0)
             {
                 OutputFlowControlAwaitable availabilityAwaitable;
                 var writeTask = Task.CompletedTask;
+                int actual;
 
                 lock (_writeLock)
                 {
@@ -317,24 +320,37 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http2
                         break;
                     }
 
-                    var actual = flowControl.AdvanceUpToAndWait(dataLength, out availabilityAwaitable);
+                    actual = flowControl.AdvanceUpToAndWait(dataLength, out availabilityAwaitable);
 
                     if (actual > 0)
                     {
+                        // Don't pass minRate through to the inner WriteData calls.
+                        // We measure this ourselves below so we account for flow control in addition to socket backpressure.
                         if (actual < dataLength)
                         {
-                            writeTask = WriteDataUnsynchronizedAsync(streamId, data.Slice(0, actual), endStream: false);
+                            writeTask = WriteDataUnsynchronizedAsync(streamId, null, data.Slice(0, actual), actual, endStream: false);
                             data = data.Slice(actual);
                             dataLength -= actual;
                         }
                         else
                         {
-                            writeTask = WriteDataUnsynchronizedAsync(streamId, data, endStream);
+                            writeTask = WriteDataUnsynchronizedAsync(streamId, null, data, actual, endStream);
                             dataLength = 0;
                         }
                     }
                 }
 
+                // Avoid timing writes that are already complete. This is likely to happen during the last iteration.
+                if (availabilityAwaitable == null && writeTask.IsCompleted)
+                {
+                    continue;
+                }
+
+                if (minRate != null)
+                {
+                    _timeoutControl.StartTimingWrite(minRate, actual);
+                }
+
                 // This awaitable releases continuations in FIFO order when the window updates.
                 // It should be very rare for a continuation to run without any availability.
                 if (availabilityAwaitable != null)
@@ -343,6 +359,11 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http2
                 }
 
                 await writeTask;
+
+                if (minRate != null)
+                {
+                    _timeoutControl.StopTimingWrite();
+                }
             }
 
             // Ensure that the application continuation isn't executed inline by ProcessWindowUpdateFrameAsync.
diff --git a/src/Kestrel.Core/Internal/Http2/Http2OutputProducer.cs b/src/Kestrel.Core/Internal/Http2/Http2OutputProducer.cs
index 1485b65681..0bdf5186a8 100644
--- a/src/Kestrel.Core/Internal/Http2/Http2OutputProducer.cs
+++ b/src/Kestrel.Core/Internal/Http2/Http2OutputProducer.cs
@@ -74,7 +74,9 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http2
             }
         }
 
-        public void Abort(ConnectionAbortedException abortReason)
+        // Review: This is called when a CancellationToken fires mid-write. In HTTP/1.x, this aborts the entire connection.
+        // Should we do that here?
+        void IHttpOutputAborter.Abort(ConnectionAbortedException abortReason)
         {
             Dispose();
         }
@@ -206,14 +208,14 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http2
                     {
                         if (readResult.Buffer.Length > 0)
                         {
-                            await _frameWriter.WriteDataAsync(_streamId, _flowControl, readResult.Buffer, endStream: false);
+                            await _frameWriter.WriteDataAsync(_streamId, _flowControl, _stream.MinResponseDataRate, readResult.Buffer, endStream: false);
                         }
 
                         await _frameWriter.WriteResponseTrailers(_streamId, _stream.Trailers);
                     }
                     else
                     {
-                        await _frameWriter.WriteDataAsync(_streamId, _flowControl, readResult.Buffer, endStream: readResult.IsCompleted);
+                        await _frameWriter.WriteDataAsync(_streamId, _flowControl, _stream.MinResponseDataRate, readResult.Buffer, endStream: readResult.IsCompleted);
                     }
 
                     _dataPipe.Reader.AdvanceTo(readResult.Buffer.End);
diff --git a/src/Kestrel.Core/Internal/Http2/Http2Stream.cs b/src/Kestrel.Core/Internal/Http2/Http2Stream.cs
index 79d93b2a4c..25623576ed 100644
--- a/src/Kestrel.Core/Internal/Http2/Http2Stream.cs
+++ b/src/Kestrel.Core/Internal/Http2/Http2Stream.cs
@@ -24,6 +24,8 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http2
         private readonly StreamInputFlowControl _inputFlowControl;
         private readonly StreamOutputFlowControl _outputFlowControl;
 
+        internal DateTimeOffset DrainExpiration { get; set; }
+
         private StreamCompletionFlags _completionState;
         private readonly object _completionLock = new object();
 
@@ -53,7 +55,8 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http2
         public bool RequestBodyStarted { get; private set; }
         public bool EndStreamReceived => (_completionState & StreamCompletionFlags.EndStreamReceived) == StreamCompletionFlags.EndStreamReceived;
         private bool IsAborted => (_completionState & StreamCompletionFlags.Aborted) == StreamCompletionFlags.Aborted;
-        internal bool DoNotDrainRequest => (_completionState & StreamCompletionFlags.DoNotDrainRequest) == StreamCompletionFlags.DoNotDrainRequest;
+        internal bool RstStreamReceived => (_completionState & StreamCompletionFlags.RstStreamReceived) == StreamCompletionFlags.RstStreamReceived;
+        internal bool IsDraining => (_completionState & StreamCompletionFlags.Draining) == StreamCompletionFlags.Draining;
 
         public override bool IsUpgradableRequest => false;
 
@@ -64,10 +67,25 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http2
 
         protected override void OnRequestProcessingEnded()
         {
-            var states = ApplyCompletionFlag(StreamCompletionFlags.RequestProcessingEnded);
-
             try
             {
+                // https://tools.ietf.org/html/rfc7540#section-8.1
+                // If the app finished without reading the request body tell the client not to finish sending it.
+                if (!EndStreamReceived && !RstStreamReceived)
+                {
+                    Log.RequestBodyNotEntirelyRead(ConnectionIdFeature, TraceIdentifier);
+
+                    ApplyCompletionFlag(StreamCompletionFlags.Draining);
+
+                    var states = ApplyCompletionFlag(StreamCompletionFlags.Aborted);
+                    if (states.OldState != states.NewState)
+                    {
+                        // Don't block on IO. This never faults.
+                        _ = _http2Output.WriteRstStreamAsync(Http2ErrorCode.NO_ERROR);
+                        RequestBodyPipe.Writer.Complete();
+                    }
+                }
+
                 _http2Output.Dispose();
 
                 RequestBodyPipe.Reader.Complete();
@@ -80,7 +98,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http2
             }
             finally
             {
-                TryFireOnStreamCompleted(states);
+                _context.StreamLifetimeHandler.OnStreamCompleted(StreamId);
             }
         }
 
@@ -309,34 +327,37 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http2
 
                 _inputFlowControl.Advance((int)dataPayload.Length);
 
-                if (IsAborted)
+                lock (_completionLock)
                 {
-                    // Ignore data frames for aborted streams, but only after counting them for purposes of connection level flow control.
-                    return Task.CompletedTask;
-                }
-
-                // This check happens after flow control so that when we throw and abort, the byte count is returned to the connection
-                // level accounting.
-                if (InputRemaining.HasValue)
-                {
-                    // https://tools.ietf.org/html/rfc7540#section-8.1.2.6
-                    if (dataPayload.Length > InputRemaining.Value)
+                    if (IsAborted)
                     {
-                        throw new Http2StreamErrorException(StreamId, CoreStrings.Http2StreamErrorMoreDataThanLength, Http2ErrorCode.PROTOCOL_ERROR);
+                        // Ignore data frames for aborted streams, but only after counting them for purposes of connection level flow control.
+                        return Task.CompletedTask;
                     }
 
-                    InputRemaining -= dataPayload.Length;
-                }
+                    // This check happens after flow control so that when we throw and abort, the byte count is returned to the connection
+                    // level accounting.
+                    if (InputRemaining.HasValue)
+                    {
+                        // https://tools.ietf.org/html/rfc7540#section-8.1.2.6
+                        if (dataPayload.Length > InputRemaining.Value)
+                        {
+                            throw new Http2StreamErrorException(StreamId, CoreStrings.Http2StreamErrorMoreDataThanLength, Http2ErrorCode.PROTOCOL_ERROR);
+                        }
 
-                foreach (var segment in dataPayload)
-                {
-                    RequestBodyPipe.Writer.Write(segment.Span);
-                }
-                var flushTask = RequestBodyPipe.Writer.FlushAsync();
+                        InputRemaining -= dataPayload.Length;
+                    }
 
-                // It shouldn't be possible for the RequestBodyPipe to fill up an return an incomplete task if
-                // _inputFlowControl.Advance() didn't throw.
-                Debug.Assert(flushTask.IsCompleted);
+                    foreach (var segment in dataPayload)
+                    {
+                        RequestBodyPipe.Writer.Write(segment.Span);
+                    }
+                    var flushTask = RequestBodyPipe.Writer.FlushAsync();
+
+                    // It shouldn't be possible for the RequestBodyPipe to fill up an return an incomplete task if
+                    // _inputFlowControl.Advance() didn't throw.
+                    Debug.Assert(flushTask.IsCompleted);
+                }
             }
 
             if (endStream)
@@ -349,6 +370,8 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http2
 
         public void OnEndStreamReceived()
         {
+            ApplyCompletionFlag(StreamCompletionFlags.EndStreamReceived);
+
             if (InputRemaining.HasValue)
             {
                 // https://tools.ietf.org/html/rfc7540#section-8.1.2.6
@@ -358,18 +381,9 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http2
                 }
             }
 
-            var states = ApplyCompletionFlag(StreamCompletionFlags.EndStreamReceived);
+            RequestBodyPipe.Writer.Complete();
 
-            try
-            {
-                RequestBodyPipe.Writer.Complete();
-
-                _inputFlowControl.StopWindowUpdates();
-            }
-            finally
-            {
-                TryFireOnStreamCompleted(states);
-            }
+            _inputFlowControl.StopWindowUpdates();
         }
 
         public void OnDataRead(int bytesRead)
@@ -382,28 +396,22 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http2
             return _context.FrameWriter.TryUpdateStreamWindow(_outputFlowControl, bytes);
         }
 
-        public void DisallowAdditionalRequestFrames()
+        public void AbortRstStreamReceived()
         {
-            ApplyCompletionFlag(StreamCompletionFlags.DoNotDrainRequest);
+            ApplyCompletionFlag(StreamCompletionFlags.RstStreamReceived);
+            Abort(new IOException(CoreStrings.Http2StreamResetByClient));
         }
 
         public void Abort(IOException abortReason)
         {
             var states = ApplyCompletionFlag(StreamCompletionFlags.Aborted);
 
-            try
+            if (states.OldState == states.NewState)
             {
-                if (states.OldState == states.NewState)
-                {
-                    return;
-                }
+                return;
+            }
 
-                AbortCore(abortReason);
-            }
-            finally
-            {
-                TryFireOnStreamCompleted(states);
-            }
+            AbortCore(abortReason);
         }
 
         protected override void OnErrorAfterResponseStarted()
@@ -429,19 +437,12 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http2
                 return;
             }
 
-            try
-            {
-                Log.Http2StreamResetAbort(TraceIdentifier, error, abortReason);
+            Log.Http2StreamResetAbort(TraceIdentifier, error, abortReason);
 
-                // Don't block on IO. This never faults.
-                _ = _http2Output.WriteRstStreamAsync(error);
+            // Don't block on IO. This never faults.
+            _ = _http2Output.WriteRstStreamAsync(error);
 
-                AbortCore(abortReason);
-            }
-            finally
-            {
-                TryFireOnStreamCompleted(states);
-            }
+            AbortCore(abortReason);
         }
 
         private void AbortCore(Exception abortReason)
@@ -484,37 +485,14 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http2
             }
         }
 
-        private void TryFireOnStreamCompleted((StreamCompletionFlags OldState, StreamCompletionFlags NewState) states)
-        {
-            if (!ShouldStopTrackingStream(states.OldState) && ShouldStopTrackingStream(states.NewState))
-            {
-                _context.StreamLifetimeHandler.OnStreamCompleted(StreamId);
-            }
-        }
-
-        private static bool ShouldStopTrackingStream(StreamCompletionFlags completionState)
-        {
-            // This could be a single condition, but I think it reads better as two if's.
-            if ((completionState & StreamCompletionFlags.RequestProcessingEnded) == StreamCompletionFlags.RequestProcessingEnded)
-            {
-                if ((completionState & StreamCompletionFlags.EndStreamReceived) == StreamCompletionFlags.EndStreamReceived ||
-                    (completionState & StreamCompletionFlags.Aborted) == StreamCompletionFlags.Aborted)
-                {
-                    return true;
-                }
-            }
-
-            return false;
-        }
-
         [Flags]
         private enum StreamCompletionFlags
         {
             None = 0,
-            RequestProcessingEnded = 1,
+            RstStreamReceived = 1,
             EndStreamReceived = 2,
             Aborted = 4,
-            DoNotDrainRequest = 8,
+            Draining = 8,
         }
     }
 }
diff --git a/src/Kestrel.Core/Internal/HttpConnection.cs b/src/Kestrel.Core/Internal/HttpConnection.cs
index 8c866f57a5..fb347de2c7 100644
--- a/src/Kestrel.Core/Internal/HttpConnection.cs
+++ b/src/Kestrel.Core/Internal/HttpConnection.cs
@@ -356,7 +356,9 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal
                 return;
             }
 
-            _timeoutControl.Tick(_systemClock.UtcNow);
+            var now = _systemClock.UtcNow;
+            _timeoutControl.Tick(now);
+            _requestProcessor?.Tick(now);
         }
 
         private void CloseUninitializedConnection(ConnectionAbortedException abortReason)
diff --git a/src/Kestrel.Core/Internal/IRequestProcessor.cs b/src/Kestrel.Core/Internal/IRequestProcessor.cs
index 23dbea150e..cf28d71ab6 100644
--- a/src/Kestrel.Core/Internal/IRequestProcessor.cs
+++ b/src/Kestrel.Core/Internal/IRequestProcessor.cs
@@ -1,6 +1,7 @@
 // Copyright (c) .NET Foundation. All rights reserved.
 // Licensed under the Apache License, Version 2.0. See License.txt in the project root for license information.
 
+using System;
 using System.Threading.Tasks;
 using Microsoft.AspNetCore.Connections;
 using Microsoft.AspNetCore.Hosting.Server;
@@ -13,6 +14,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal
         void StopProcessingNextRequest();
         void HandleRequestHeadersTimeout();
         void OnInputOrOutputCompleted();
+        void Tick(DateTimeOffset now);
         void Abort(ConnectionAbortedException ex);
     }
 }
\ No newline at end of file
diff --git a/src/Kestrel.Core/Internal/Infrastructure/TimingPipeFlusher.cs b/src/Kestrel.Core/Internal/Infrastructure/TimingPipeFlusher.cs
index eb9554571a..794b2a3430 100644
--- a/src/Kestrel.Core/Internal/Infrastructure/TimingPipeFlusher.cs
+++ b/src/Kestrel.Core/Internal/Infrastructure/TimingPipeFlusher.cs
@@ -40,6 +40,11 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Infrastructure
             return FlushAsync(minRate: null, count: 0, outputAborter: outputAborter, cancellationToken: cancellationToken);
         }
 
+        public Task FlushAsync(MinDataRate minRate, long count)
+        {
+            return FlushAsync(minRate, count, outputAborter: null, cancellationToken: default);
+        }
+
         public Task FlushAsync(MinDataRate minRate, long count, IHttpOutputAborter outputAborter, CancellationToken cancellationToken)
         {
             var flushValueTask = _writer.FlushAsync(cancellationToken);
diff --git a/test/Kestrel.InMemory.FunctionalTests/Http2/Http2ConnectionTests.cs b/test/Kestrel.InMemory.FunctionalTests/Http2/Http2ConnectionTests.cs
index 1654641733..28e4fb3a56 100644
--- a/test/Kestrel.InMemory.FunctionalTests/Http2/Http2ConnectionTests.cs
+++ b/test/Kestrel.InMemory.FunctionalTests/Http2/Http2ConnectionTests.cs
@@ -16,68 +16,13 @@ using Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http2.HPack;
 using Microsoft.AspNetCore.Testing;
 using Microsoft.Extensions.Logging;
 using Microsoft.Net.Http.Headers;
+using Moq;
 using Xunit;
 
 namespace Microsoft.AspNetCore.Server.Kestrel.Core.Tests
 {
     public class Http2ConnectionTests : Http2TestBase
     {
-        private static readonly IEnumerable<KeyValuePair<string, string>> _postRequestHeaders = new[]
-        {
-            new KeyValuePair<string, string>(HeaderNames.Method, "POST"),
-            new KeyValuePair<string, string>(HeaderNames.Path, "/"),
-            new KeyValuePair<string, string>(HeaderNames.Scheme, "http"),
-            new KeyValuePair<string, string>(HeaderNames.Authority, "localhost:80"),
-        };
-
-        private static readonly IEnumerable<KeyValuePair<string, string>> _expectContinueRequestHeaders = new[]
-        {
-            new KeyValuePair<string, string>(HeaderNames.Method, "POST"),
-            new KeyValuePair<string, string>(HeaderNames.Path, "/"),
-            new KeyValuePair<string, string>(HeaderNames.Authority, "127.0.0.1"),
-            new KeyValuePair<string, string>(HeaderNames.Scheme, "http"),
-            new KeyValuePair<string, string>("expect", "100-continue"),
-        };
-
-        private static readonly IEnumerable<KeyValuePair<string, string>> _requestTrailers = new[]
-        {
-            new KeyValuePair<string, string>("trailer-one", "1"),
-            new KeyValuePair<string, string>("trailer-two", "2"),
-        };
-
-        private static readonly IEnumerable<KeyValuePair<string, string>> _oneContinuationRequestHeaders = 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>("a", _4kHeaderValue),
-            new KeyValuePair<string, string>("b", _4kHeaderValue),
-            new KeyValuePair<string, string>("c", _4kHeaderValue),
-            new KeyValuePair<string, string>("d", _4kHeaderValue)
-        };
-
-        private static readonly IEnumerable<KeyValuePair<string, string>> _twoContinuationsRequestHeaders = 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>("a", _4kHeaderValue),
-            new KeyValuePair<string, string>("b", _4kHeaderValue),
-            new KeyValuePair<string, string>("c", _4kHeaderValue),
-            new KeyValuePair<string, string>("d", _4kHeaderValue),
-            new KeyValuePair<string, string>("e", _4kHeaderValue),
-            new KeyValuePair<string, string>("f", _4kHeaderValue),
-            new KeyValuePair<string, string>("g", _4kHeaderValue),
-        };
-
-        private static readonly byte[] _helloBytes = Encoding.ASCII.GetBytes("hello");
-        private static readonly byte[] _worldBytes = Encoding.ASCII.GetBytes("world");
-        private static readonly byte[] _helloWorldBytes = Encoding.ASCII.GetBytes("hello, world");
-        private static readonly byte[] _noData = new byte[0];
-        private static readonly byte[] _maxData = Encoding.ASCII.GetBytes(new string('a', Http2PeerSettings.MinAllowedMaxFrameSize));
-
         [Fact]
         public async Task Frame_Received_OverMaxSize_FrameError()
         {
@@ -99,8 +44,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Tests
         public async Task ServerSettings_ChangesRequestMaxFrameSize()
         {
             var length = Http2PeerSettings.MinAllowedMaxFrameSize + 10;
-            _connectionContext.ServiceContext.ServerOptions.Limits.Http2.MaxFrameSize = length;
-            _connection = new Http2Connection(_connectionContext);
+            _serviceContext.ServerOptions.Limits.Http2.MaxFrameSize = length;
 
             await InitializeConnectionAsync(_echoApplication, expectedSettingsCount: 4);
 
@@ -184,9 +128,9 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Tests
         [Fact]
         public async Task DATA_Received_GreaterThanInitialWindowSize_ReadByStream()
         {
-            var initialStreamWindowSize = _connectionContext.ServiceContext.ServerOptions.Limits.Http2.InitialStreamWindowSize;
+            var initialStreamWindowSize = _serviceContext.ServerOptions.Limits.Http2.InitialStreamWindowSize;
             var framesInStreamWindow = initialStreamWindowSize / Http2PeerSettings.DefaultMaxFrameSize;
-            var initialConnectionWindowSize = _connectionContext.ServiceContext.ServerOptions.Limits.Http2.InitialConnectionWindowSize;
+            var initialConnectionWindowSize = _serviceContext.ServerOptions.Limits.Http2.InitialConnectionWindowSize;
             var framesInConnectionWindow = initialConnectionWindowSize / Http2PeerSettings.DefaultMaxFrameSize;
 
             // Grow the client stream windows so no stream WINDOW_UPDATEs need to be sent.
@@ -285,9 +229,9 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Tests
         [Fact]
         public async Task DATA_Received_RightAtWindowLimit_DoesNotPausePipe()
         {
-            var initialStreamWindowSize = _connectionContext.ServiceContext.ServerOptions.Limits.Http2.InitialStreamWindowSize;
+            var initialStreamWindowSize = _serviceContext.ServerOptions.Limits.Http2.InitialStreamWindowSize;
             var framesInStreamWindow = initialStreamWindowSize / Http2PeerSettings.DefaultMaxFrameSize;
-            var initialConnectionWindowSize = _connectionContext.ServiceContext.ServerOptions.Limits.Http2.InitialConnectionWindowSize;
+            var initialConnectionWindowSize = _serviceContext.ServerOptions.Limits.Http2.InitialConnectionWindowSize;
             var framesInConnectionWindow = initialConnectionWindowSize / Http2PeerSettings.DefaultMaxFrameSize;
 
             await InitializeConnectionAsync(_waitForAbortApplication);
@@ -413,8 +357,8 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Tests
         [Fact]
         public async Task DATA_Received_Multiplexed_GreaterThanInitialWindowSize_ReadByStream()
         {
-            var initialStreamWindowSize = _connectionContext.ServiceContext.ServerOptions.Limits.Http2.InitialStreamWindowSize;
-            var initialConnectionWindowSize = _connectionContext.ServiceContext.ServerOptions.Limits.Http2.InitialConnectionWindowSize;
+            var initialStreamWindowSize = _serviceContext.ServerOptions.Limits.Http2.InitialStreamWindowSize;
+            var initialConnectionWindowSize = _serviceContext.ServerOptions.Limits.Http2.InitialConnectionWindowSize;
             var framesInStreamWindow = initialStreamWindowSize / Http2PeerSettings.DefaultMaxFrameSize;
             var framesInConnectionWindow = initialConnectionWindowSize / Http2PeerSettings.DefaultMaxFrameSize;
 
@@ -553,10 +497,10 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Tests
             await StartStreamAsync(1, _browserRequestHeaders, endStream: false);
             await StartStreamAsync(3, _browserRequestHeaders, endStream: false);
 
-            await SendDataAsync(1, _helloBytes, endStream: false);
+            await SendDataAsync(1, _helloBytes, endStream: true);
             Assert.True(stream1Read.WaitOne(TimeSpan.FromSeconds(10)));
 
-            await SendDataAsync(3, _helloBytes, endStream: false);
+            await SendDataAsync(3, _helloBytes, endStream: true);
             Assert.True(stream3Read.WaitOne(TimeSpan.FromSeconds(10)));
 
             stream3ReadFinished.Set();
@@ -627,7 +571,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Tests
         [InlineData(255)]
         public async Task DATA_Received_WithPadding_CountsTowardsInputFlowControl(byte padLength)
         {
-            var initialWindowSize = _connectionContext.ServiceContext.ServerOptions.Limits.Http2.InitialStreamWindowSize;
+            var initialWindowSize = _serviceContext.ServerOptions.Limits.Http2.InitialStreamWindowSize;
             var framesInWindow = initialWindowSize / Http2PeerSettings.DefaultMaxFrameSize;
             var maxDataMinusPadding = _maxData.AsMemory(0, _maxData.Length - padLength - 1);
 
@@ -696,7 +640,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Tests
         [Fact]
         public async Task DATA_Received_ButNotConsumedByApp_CountsTowardsInputFlowControl()
         {
-            var initialConnectionWindowSize = _connectionContext.ServiceContext.ServerOptions.Limits.Http2.InitialConnectionWindowSize;
+            var initialConnectionWindowSize = _serviceContext.ServerOptions.Limits.Http2.InitialConnectionWindowSize;
             var framesConnectionInWindow = initialConnectionWindowSize / Http2PeerSettings.DefaultMaxFrameSize;
 
             await InitializeConnectionAsync(_noopApplication);
@@ -716,6 +660,10 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Tests
                 withFlags: (byte)Http2DataFrameFlags.END_STREAM,
                 withStreamId: 1);
 
+            await WaitForStreamErrorAsync(expectedStreamId: 1, Http2ErrorCode.NO_ERROR, null);
+            // Logged without an exception.
+            Assert.Contains(TestApplicationErrorLogger.Messages, m => m.Message.Contains("the application completed without reading the entire request body."));
+
             // Writing over half the initial window size induces a connection-level window update.
             // But no stream window update since this is the last frame.
             await SendDataAsync(1, _maxData, endStream: true);
@@ -918,7 +866,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Tests
         [Fact]
         public async Task DATA_Received_NoStreamWindowSpace_ConnectionError()
         {
-            var initialWindowSize = _connectionContext.ServiceContext.ServerOptions.Limits.Http2.InitialStreamWindowSize;
+            var initialWindowSize = _serviceContext.ServerOptions.Limits.Http2.InitialStreamWindowSize;
             var framesInWindow = (initialWindowSize / Http2PeerSettings.DefaultMaxFrameSize) + 1; // Round up to overflow the window
 
             await InitializeConnectionAsync(_waitForAbortApplication);
@@ -940,7 +888,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Tests
         [Fact]
         public async Task DATA_Received_NoConnectionWindowSpace_ConnectionError()
         {
-            var initialWindowSize = _connectionContext.ServiceContext.ServerOptions.Limits.Http2.InitialConnectionWindowSize;
+            var initialWindowSize = _serviceContext.ServerOptions.Limits.Http2.InitialConnectionWindowSize;
             var framesInWindow = initialWindowSize / Http2PeerSettings.DefaultMaxFrameSize;
 
             await InitializeConnectionAsync(_waitForAbortApplication);
@@ -1307,6 +1255,8 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Tests
         [Fact]
         public async Task HEADERS_OverMaxStreamLimit_Refused()
         {
+            CreateConnection();
+
             _connection.ServerSettings.MaxConcurrentStreams = 1;
 
             var requestBlocker = new TaskCompletionSource<object>();
@@ -2086,7 +2036,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Tests
         [Fact]
         public async Task RST_STREAM_Received_ReturnsSpaceToConnectionInputFlowControlWindow()
         {
-            var initialConnectionWindowSize = _connectionContext.ServiceContext.ServerOptions.Limits.Http2.InitialConnectionWindowSize;
+            var initialConnectionWindowSize = _serviceContext.ServerOptions.Limits.Http2.InitialConnectionWindowSize;
             var framesInConnectionWindow = initialConnectionWindowSize / Http2PeerSettings.DefaultMaxFrameSize;
 
             await InitializeConnectionAsync(_waitForAbortApplication);
@@ -2291,6 +2241,8 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Tests
         [Fact]
         public async Task SETTINGS_KestrelDefaults_Sent()
         {
+            CreateConnection();
+
             _connectionTask = _connection.ProcessRequestsAsync(new DummyApplication(_noopApplication));
 
             await SendPreambleAsync().ConfigureAwait(false);
@@ -2335,6 +2287,8 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Tests
         [Fact]
         public async Task SETTINGS_Custom_Sent()
         {
+            CreateConnection();
+
             _connection.ServerSettings.MaxConcurrentStreams = 1;
             _connection.ServerSettings.MaxHeaderListSize = 4 * 1024;
             _connection.ServerSettings.InitialWindowSize = 1024 * 1024 * 10;
@@ -2515,6 +2469,8 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Tests
         [Fact]
         public async Task SETTINGS_Received_ChangesAllowedResponseMaxFrameSize()
         {
+            CreateConnection();
+
             _connection.ServerSettings.MaxFrameSize = Http2PeerSettings.MaxAllowedMaxFrameSize;
             // This includes the default response headers such as :status, etc
             var defaultResponseHeaderLength = 37;
@@ -2571,6 +2527,9 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Tests
         public async Task SETTINGS_Received_ClientMaxFrameSizeCannotExceedServerMaxFrameSize()
         {
             var serverMaxFrame = Http2PeerSettings.MinAllowedMaxFrameSize + 1024;
+
+            CreateConnection();
+
             _connection.ServerSettings.MaxFrameSize = Http2PeerSettings.MinAllowedMaxFrameSize + 1024;
             var clientMaxFrame = serverMaxFrame + 1024 * 5;
             _clientSettings.MaxFrameSize = (uint)clientMaxFrame;
@@ -3742,6 +3701,9 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Tests
         [Fact]
         public async Task IgnoreNewStreamsDuringClosedConnection()
         {
+            // Remove callback that completes _pair.Application.Output on abort.
+            _mockConnectionContext.Reset();
+
             await InitializeConnectionAsync(_echoApplication);
 
             await StartStreamAsync(1, _browserRequestHeaders, endStream: false);
@@ -3759,6 +3721,8 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Tests
         [Fact]
         public void IOExceptionDuringFrameProcessingLoggedAsInfo()
         {
+            CreateConnection();
+
             var ioException = new IOException();
             _pair.Application.Output.Complete(ioException);
 
@@ -3774,6 +3738,8 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Tests
         [Fact]
         public void UnexpectedExceptionDuringFrameProcessingLoggedAWarning()
         {
+            CreateConnection();
+
             var exception = new Exception();
             _pair.Application.Output.Complete(exception);
 
@@ -3786,6 +3752,213 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Tests
             Assert.Same(exception, logMessage.Exception);
         }
 
+        [Theory]
+        [InlineData(Http2FrameType.DATA)]
+        [InlineData(Http2FrameType.WINDOW_UPDATE)]
+        [InlineData(Http2FrameType.HEADERS)]
+        [InlineData(Http2FrameType.CONTINUATION)]
+        public async Task AppDoesNotReadRequestBody_ResetsAndDrainsRequest(Http2FrameType finalFrameType)
+        {
+            var headers = new[]
+            {
+                new KeyValuePair<string, string>(HeaderNames.Method, "POST"),
+                new KeyValuePair<string, string>(HeaderNames.Path, "/"),
+                new KeyValuePair<string, string>(HeaderNames.Scheme, "http"),
+            };
+            await InitializeConnectionAsync(_noopApplication);
+
+            await StartStreamAsync(1, headers, endStream: false);
+
+            await ExpectAsync(Http2FrameType.HEADERS,
+                withLength: 55,
+                withFlags: (byte)Http2HeadersFrameFlags.END_HEADERS,
+                withStreamId: 1);
+            await ExpectAsync(Http2FrameType.DATA,
+                withLength: 0,
+                withFlags: (byte)Http2DataFrameFlags.END_STREAM,
+                withStreamId: 1);
+
+            await WaitForStreamErrorAsync(1, Http2ErrorCode.NO_ERROR, null);
+            // Logged without an exception.
+            Assert.Contains(TestApplicationErrorLogger.Messages, m => m.Message.Contains("the application completed without reading the entire request body."));
+
+            // There's a race when the appfunc is exiting about how soon it unregisters the stream.
+            for (var i = 0; i < 10; i++)
+            {
+                await SendDataAsync(1, new byte[100], endStream: false);
+            }
+
+            // These would be refused if the cool-down period had expired
+            switch (finalFrameType)
+            {
+                case Http2FrameType.DATA:
+                    await SendDataAsync(1, new byte[100], endStream: true);
+                    break;
+                case Http2FrameType.WINDOW_UPDATE:
+                    await SendWindowUpdateAsync(1, 1024);
+                    break;
+                case Http2FrameType.HEADERS:
+                    await SendHeadersAsync(1, Http2HeadersFrameFlags.END_STREAM | Http2HeadersFrameFlags.END_HEADERS, _requestTrailers);
+                    break;
+                case Http2FrameType.CONTINUATION:
+                    await SendHeadersAsync(1, Http2HeadersFrameFlags.END_STREAM, _requestTrailers);
+                    await SendContinuationAsync(1, Http2ContinuationFrameFlags.END_HEADERS, _requestTrailers);
+                    break;
+                default:
+                    throw new NotImplementedException(finalFrameType.ToString());
+            }
+
+            await StopConnectionAsync(expectedLastStreamId: 1, ignoreNonGoAwayFrames: false);
+        }
+
+        [Theory]
+        [InlineData(Http2FrameType.DATA)]
+        [InlineData(Http2FrameType.WINDOW_UPDATE)]
+        [InlineData(Http2FrameType.HEADERS)]
+        [InlineData(Http2FrameType.CONTINUATION)]
+        public async Task AbortedStream_ResetsAndDrainsRequest(Http2FrameType finalFrameType)
+        {
+            var headers = new[]
+            {
+                new KeyValuePair<string, string>(HeaderNames.Method, "POST"),
+                new KeyValuePair<string, string>(HeaderNames.Path, "/"),
+                new KeyValuePair<string, string>(HeaderNames.Scheme, "http"),
+            };
+            await InitializeConnectionAsync(_appAbort);
+
+            await StartStreamAsync(1, headers, endStream: false);
+
+            await WaitForStreamErrorAsync(1, Http2ErrorCode.INTERNAL_ERROR, "The connection was aborted by the application.");
+
+            // There's a race when the appfunc is exiting about how soon it unregisters the stream.
+            for (var i = 0; i < 10; i++)
+            {
+                await SendDataAsync(1, new byte[100], endStream: false);
+            }
+
+            // These would be refused if the cool-down period had expired
+            switch (finalFrameType)
+            {
+                case Http2FrameType.DATA:
+                    await SendDataAsync(1, new byte[100], endStream: true);
+                    break;
+                case Http2FrameType.WINDOW_UPDATE:
+                    await SendWindowUpdateAsync(1, 1024);
+                    break;
+                case Http2FrameType.HEADERS:
+                    await SendHeadersAsync(1, Http2HeadersFrameFlags.END_STREAM | Http2HeadersFrameFlags.END_HEADERS, _requestTrailers);
+                    break;
+                case Http2FrameType.CONTINUATION:
+                    await SendHeadersAsync(1, Http2HeadersFrameFlags.END_STREAM, _requestTrailers);
+                    await SendContinuationAsync(1, Http2ContinuationFrameFlags.END_HEADERS, _requestTrailers);
+                    break;
+                default:
+                    throw new NotImplementedException(finalFrameType.ToString());
+            }
+
+            await StopConnectionAsync(expectedLastStreamId: 1, ignoreNonGoAwayFrames: false);
+        }
+
+        [Theory]
+        [InlineData(Http2FrameType.DATA)]
+        [InlineData(Http2FrameType.HEADERS)]
+        [InlineData(Http2FrameType.CONTINUATION)]
+        public async Task AbortedStream_ResetsAndDrainsRequest_RefusesFramesAfterEndOfStream(Http2FrameType finalFrameType)
+        {
+            var headers = new[]
+            {
+                new KeyValuePair<string, string>(HeaderNames.Method, "POST"),
+                new KeyValuePair<string, string>(HeaderNames.Path, "/"),
+                new KeyValuePair<string, string>(HeaderNames.Scheme, "http"),
+            };
+            await InitializeConnectionAsync(_appAbort);
+
+            await StartStreamAsync(1, headers, endStream: false);
+
+            await WaitForStreamErrorAsync(1, Http2ErrorCode.INTERNAL_ERROR, "The connection was aborted by the application.");
+
+            // There's a race when the appfunc is exiting about how soon it unregisters the stream.
+            for (var i = 0; i < 10; i++)
+            {
+                await SendDataAsync(1, new byte[100], endStream: false);
+            }
+
+            switch (finalFrameType)
+            {
+                case Http2FrameType.DATA:
+                    await SendDataAsync(1, new byte[100], endStream: true);
+                    // An extra one to break it
+                    await SendDataAsync(1, new byte[100], endStream: true);
+
+                    await WaitForConnectionErrorAsync<Http2ConnectionErrorException>(ignoreNonGoAwayFrames: false, expectedLastStreamId: 1, Http2ErrorCode.STREAM_CLOSED,
+                        CoreStrings.FormatHttp2ErrorStreamClosed(Http2FrameType.DATA, 1));
+                    break;
+
+                case Http2FrameType.HEADERS:
+                    await SendHeadersAsync(1, Http2HeadersFrameFlags.END_STREAM | Http2HeadersFrameFlags.END_HEADERS, _requestTrailers);
+                    // An extra one to break it
+                    await SendHeadersAsync(1, Http2HeadersFrameFlags.END_STREAM | Http2HeadersFrameFlags.END_HEADERS, _requestTrailers);
+
+                    await WaitForConnectionErrorAsync<Http2ConnectionErrorException>(ignoreNonGoAwayFrames: false, expectedLastStreamId: 1, Http2ErrorCode.STREAM_CLOSED,
+                        CoreStrings.FormatHttp2ErrorStreamClosed(Http2FrameType.HEADERS, 1));
+                    break;
+
+                case Http2FrameType.CONTINUATION:
+                    await SendHeadersAsync(1, Http2HeadersFrameFlags.END_STREAM, _requestTrailers);
+                    await SendContinuationAsync(1, Http2ContinuationFrameFlags.END_HEADERS, _requestTrailers);
+                    // An extra one to break it. It's not a Continuation because that would fail with an error that no headers were in progress.
+                    await SendHeadersAsync(1, Http2HeadersFrameFlags.END_STREAM, _requestTrailers);
+                    await WaitForConnectionErrorAsync<Http2ConnectionErrorException>(ignoreNonGoAwayFrames: false, expectedLastStreamId: 1, Http2ErrorCode.STREAM_CLOSED,
+                        CoreStrings.FormatHttp2ErrorStreamClosed(Http2FrameType.HEADERS, 1));
+                    break;
+                default:
+                    throw new NotImplementedException(finalFrameType.ToString());
+            }
+        }
+
+        [Theory]
+        [InlineData(Http2FrameType.DATA)]
+        [InlineData(Http2FrameType.HEADERS)]
+        public async Task AbortedStream_ResetsAndDrainsRequest_RefusesFramesAfterClientReset(Http2FrameType finalFrameType)
+        {
+            var headers = new[]
+            {
+                new KeyValuePair<string, string>(HeaderNames.Method, "POST"),
+                new KeyValuePair<string, string>(HeaderNames.Path, "/"),
+                new KeyValuePair<string, string>(HeaderNames.Scheme, "http"),
+            };
+            await InitializeConnectionAsync(_appAbort);
+
+            await StartStreamAsync(1, headers, endStream: false);
+
+            await WaitForStreamErrorAsync(1, Http2ErrorCode.INTERNAL_ERROR, "The connection was aborted by the application.");
+
+            // There's a race when the appfunc is exiting about how soon it unregisters the stream.
+            for (var i = 0; i < 10; i++)
+            {
+                await SendDataAsync(1, new byte[100], endStream: false);
+            }
+            await SendRstStreamAsync(1);
+
+            // Send an extra frame to make it fail
+            switch (finalFrameType)
+            {
+                case Http2FrameType.DATA:
+                    await SendDataAsync(1, new byte[100], endStream: true);
+                    break;
+
+                case Http2FrameType.HEADERS:
+                    await SendHeadersAsync(1, Http2HeadersFrameFlags.END_STREAM | Http2HeadersFrameFlags.END_HEADERS, _requestTrailers);
+                    break;
+
+                default:
+                    throw new NotImplementedException(finalFrameType.ToString());
+            }
+
+            await WaitForConnectionErrorAsync<Http2ConnectionErrorException>(ignoreNonGoAwayFrames: false, expectedLastStreamId: 1, Http2ErrorCode.STREAM_CLOSED,
+                CoreStrings.FormatHttp2ErrorStreamClosed(finalFrameType, 1));
+        }
+
         public static TheoryData<byte[]> UpperCaseHeaderNameData
         {
             get
diff --git a/test/Kestrel.InMemory.FunctionalTests/Http2/Http2StreamTests.cs b/test/Kestrel.InMemory.FunctionalTests/Http2/Http2StreamTests.cs
index 74912e65b6..3d71d253d3 100644
--- a/test/Kestrel.InMemory.FunctionalTests/Http2/Http2StreamTests.cs
+++ b/test/Kestrel.InMemory.FunctionalTests/Http2/Http2StreamTests.cs
@@ -606,6 +606,8 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Tests
                 var buffer = new byte[100];
                 var read = await context.Request.Body.ReadAsync(buffer, 0, buffer.Length);
                 Assert.Equal(12, read);
+                read = await context.Request.Body.ReadAsync(buffer, 0, buffer.Length);
+                Assert.Equal(0, read);
             });
 
             await StartStreamAsync(1, headers, endStream: false);
@@ -638,6 +640,8 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Tests
                 var buffer = new byte[100];
                 var read = await context.Request.Body.ReadAsync(buffer, 0, buffer.Length);
                 Assert.Equal(12, read);
+                read = await context.Request.Body.ReadAsync(buffer, 0, buffer.Length);
+                Assert.Equal(0, read);
             });
 
             var headers = new[]
@@ -833,7 +837,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Tests
             Assert.IsType<Http2StreamErrorException>(thrownEx.InnerException);
         }
 
-        [Fact(Skip = "Flaky test #2799, #2832")]
+        [Fact]
         public async Task ContentLength_Received_MultipleDataFramesOverSize_Reset()
         {
             IOException thrownEx = null;
@@ -858,8 +862,6 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Tests
             await SendDataAsync(1, new byte[1], endStream: false);
             await SendDataAsync(1, new byte[2], endStream: false);
             await SendDataAsync(1, new byte[10], endStream: false);
-            await SendDataAsync(1, new byte[2], endStream: true);
-
             await WaitForStreamErrorAsync(1, Http2ErrorCode.PROTOCOL_ERROR, CoreStrings.Http2StreamErrorMoreDataThanLength);
 
             await StopConnectionAsync(expectedLastStreamId: 1, ignoreNonGoAwayFrames: false);
@@ -1063,7 +1065,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Tests
         [Fact]
         public async Task MaxRequestBodySize_ContentLengthUnder_200()
         {
-            _connectionContext.ServiceContext.ServerOptions.Limits.MaxRequestBodySize = 15;
+            _serviceContext.ServerOptions.Limits.MaxRequestBodySize = 15;
             var headers = new[]
             {
                 new KeyValuePair<string, string>(HeaderNames.Method, "POST"),
@@ -1076,6 +1078,8 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Tests
                 var buffer = new byte[100];
                 var read = await context.Request.Body.ReadAsync(buffer, 0, buffer.Length);
                 Assert.Equal(12, read);
+                read = await context.Request.Body.ReadAsync(buffer, 0, buffer.Length);
+                Assert.Equal(0, read);
             });
 
             await StartStreamAsync(1, headers, endStream: false);
@@ -1104,7 +1108,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Tests
         public async Task MaxRequestBodySize_ContentLengthOver_413()
         {
             BadHttpRequestException exception = null;
-            _connectionContext.ServiceContext.ServerOptions.Limits.MaxRequestBodySize = 10;
+            _serviceContext.ServerOptions.Limits.MaxRequestBodySize = 10;
             var headers = new[]
             {
                 new KeyValuePair<string, string>(HeaderNames.Method, "POST"),
@@ -1133,6 +1137,10 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Tests
                 withFlags: (byte)Http2DataFrameFlags.END_STREAM,
                 withStreamId: 1);
 
+            await WaitForStreamErrorAsync(expectedStreamId: 1, Http2ErrorCode.NO_ERROR, null);
+            // Logged without an exception.
+            Assert.Contains(TestApplicationErrorLogger.Messages, m => m.Message.Contains("the application completed without reading the entire request body."));
+
             await StopConnectionAsync(expectedLastStreamId: 1, ignoreNonGoAwayFrames: false);
 
             _hpackDecoder.Decode(headersFrame.PayloadSequence, endHeaders: false, handler: this);
@@ -1148,7 +1156,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Tests
         [Fact]
         public async Task MaxRequestBodySize_NoContentLength_Under_200()
         {
-            _connectionContext.ServiceContext.ServerOptions.Limits.MaxRequestBodySize = 15;
+            _serviceContext.ServerOptions.Limits.MaxRequestBodySize = 15;
             var headers = new[]
             {
                 new KeyValuePair<string, string>(HeaderNames.Method, "POST"),
@@ -1160,6 +1168,8 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Tests
                 var buffer = new byte[100];
                 var read = await context.Request.Body.ReadAsync(buffer, 0, buffer.Length);
                 Assert.Equal(12, read);
+                read = await context.Request.Body.ReadAsync(buffer, 0, buffer.Length);
+                Assert.Equal(0, read);
             });
 
             await StartStreamAsync(1, headers, endStream: false);
@@ -1188,7 +1198,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Tests
         public async Task MaxRequestBodySize_NoContentLength_Over_413()
         {
             BadHttpRequestException exception = null;
-            _connectionContext.ServiceContext.ServerOptions.Limits.MaxRequestBodySize = 10;
+            _serviceContext.ServerOptions.Limits.MaxRequestBodySize = 10;
             var headers = new[]
             {
                 new KeyValuePair<string, string>(HeaderNames.Method, "POST"),
@@ -1237,7 +1247,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Tests
         public async Task MaxRequestBodySize_AppCanLowerLimit(bool includeContentLength)
         {
             BadHttpRequestException exception = null;
-            _connectionContext.ServiceContext.ServerOptions.Limits.MaxRequestBodySize = 20;
+            _serviceContext.ServerOptions.Limits.MaxRequestBodySize = 20;
             var headers = new[]
             {
                 new KeyValuePair<string, string>(HeaderNames.Method, "POST"),
@@ -1295,7 +1305,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Tests
         [InlineData(false)]
         public async Task MaxRequestBodySize_AppCanRaiseLimit(bool includeContentLength)
         {
-            _connectionContext.ServiceContext.ServerOptions.Limits.MaxRequestBodySize = 10;
+            _serviceContext.ServerOptions.Limits.MaxRequestBodySize = 10;
             var headers = new[]
             {
                 new KeyValuePair<string, string>(HeaderNames.Method, "POST"),
@@ -1317,6 +1327,8 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Tests
                 var read = await context.Request.Body.ReadAsync(buffer, 0, buffer.Length);
                 Assert.Equal(12, read);
                 Assert.True(context.Features.Get<IHttpMaxRequestBodySizeFeature>().IsReadOnly);
+                read = await context.Request.Body.ReadAsync(buffer, 0, buffer.Length);
+                Assert.Equal(0, read);
             });
 
             await StartStreamAsync(1, headers, endStream: false);
diff --git a/test/Kestrel.InMemory.FunctionalTests/Http2/Http2TestBase.cs b/test/Kestrel.InMemory.FunctionalTests/Http2/Http2TestBase.cs
index c1d1339c26..e92f81d157 100644
--- a/test/Kestrel.InMemory.FunctionalTests/Http2/Http2TestBase.cs
+++ b/test/Kestrel.InMemory.FunctionalTests/Http2/Http2TestBase.cs
@@ -10,6 +10,7 @@ using System.IO;
 using System.IO.Pipelines;
 using System.Linq;
 using System.Reflection;
+using System.Text;
 using System.Threading;
 using System.Threading.Tasks;
 using Microsoft.AspNetCore.Connections;
@@ -48,8 +49,63 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Tests
             new KeyValuePair<string, string>("upgrade-insecure-requests", "1"),
         };
 
+        protected static readonly IEnumerable<KeyValuePair<string, string>> _postRequestHeaders = new[]
+        {
+            new KeyValuePair<string, string>(HeaderNames.Method, "POST"),
+            new KeyValuePair<string, string>(HeaderNames.Path, "/"),
+            new KeyValuePair<string, string>(HeaderNames.Scheme, "http"),
+            new KeyValuePair<string, string>(HeaderNames.Authority, "localhost:80"),
+        };
+
+        protected static readonly IEnumerable<KeyValuePair<string, string>> _expectContinueRequestHeaders = new[]
+        {
+            new KeyValuePair<string, string>(HeaderNames.Method, "POST"),
+            new KeyValuePair<string, string>(HeaderNames.Path, "/"),
+            new KeyValuePair<string, string>(HeaderNames.Authority, "127.0.0.1"),
+            new KeyValuePair<string, string>(HeaderNames.Scheme, "http"),
+            new KeyValuePair<string, string>("expect", "100-continue"),
+        };
+
+        protected static readonly IEnumerable<KeyValuePair<string, string>> _requestTrailers = new[]
+        {
+            new KeyValuePair<string, string>("trailer-one", "1"),
+            new KeyValuePair<string, string>("trailer-two", "2"),
+        };
+
+        protected static readonly IEnumerable<KeyValuePair<string, string>> _oneContinuationRequestHeaders = 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>("a", _4kHeaderValue),
+            new KeyValuePair<string, string>("b", _4kHeaderValue),
+            new KeyValuePair<string, string>("c", _4kHeaderValue),
+            new KeyValuePair<string, string>("d", _4kHeaderValue)
+        };
+
+        protected static readonly IEnumerable<KeyValuePair<string, string>> _twoContinuationsRequestHeaders = 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>("a", _4kHeaderValue),
+            new KeyValuePair<string, string>("b", _4kHeaderValue),
+            new KeyValuePair<string, string>("c", _4kHeaderValue),
+            new KeyValuePair<string, string>("d", _4kHeaderValue),
+            new KeyValuePair<string, string>("e", _4kHeaderValue),
+            new KeyValuePair<string, string>("f", _4kHeaderValue),
+            new KeyValuePair<string, string>("g", _4kHeaderValue),
+        };
+
+        protected static readonly byte[] _helloBytes = Encoding.ASCII.GetBytes("hello");
+        protected static readonly byte[] _worldBytes = Encoding.ASCII.GetBytes("world");
+        protected static readonly byte[] _helloWorldBytes = Encoding.ASCII.GetBytes("hello, world");
+        protected static readonly byte[] _noData = new byte[0];
+        protected static readonly byte[] _maxData = Encoding.ASCII.GetBytes(new string('a', Http2PeerSettings.MinAllowedMaxFrameSize));
+
         private readonly MemoryPool<byte> _memoryPool = KestrelMemoryPool.Create();
-        internal readonly DuplexPipe.DuplexPipePair _pair;
 
         protected readonly Http2PeerSettings _clientSettings = new Http2PeerSettings();
         protected readonly HPackEncoder _hpackEncoder = new HPackEncoder();
@@ -57,6 +113,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Tests
         private readonly byte[] _headerEncodingBuffer = new byte[Http2PeerSettings.MinAllowedMaxFrameSize];
 
         protected readonly TimeoutControl _timeoutControl;
+        protected readonly Mock<IKestrelTrace> _mockKestrelTrace = new Mock<IKestrelTrace>();
         protected readonly Mock<ConnectionContext> _mockConnectionContext = new Mock<ConnectionContext>();
         protected readonly Mock<ITimeoutHandler> _mockTimeoutHandler = new Mock<ITimeoutHandler>();
         protected readonly Mock<MockTimeoutControlBase> _mockTimeoutControl;
@@ -82,34 +139,34 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Tests
         protected readonly RequestDelegate _echoMethod;
         protected readonly RequestDelegate _echoHost;
         protected readonly RequestDelegate _echoPath;
+        protected readonly RequestDelegate _appAbort;
 
-        protected HttpConnectionContext _connectionContext;
+        protected TestServiceContext _serviceContext;
+
+        internal DuplexPipe.DuplexPipePair _pair;
         protected Http2Connection _connection;
         protected Task _connectionTask;
 
         public Http2TestBase()
         {
-            // Always dispatch test code back to the ThreadPool. This prevents deadlocks caused by continuing
-            // Http2Connection.ProcessRequestsAsync() loop with writer locks acquired. Run product code inline to make
-            // it easier to verify request frames are processed correctly immediately after sending the them.
-            var inputPipeOptions = new PipeOptions(
-                pool: _memoryPool,
-                readerScheduler: PipeScheduler.Inline,
-                writerScheduler: PipeScheduler.ThreadPool,
-                useSynchronizationContext: false
-            );
-            var outputPipeOptions = new PipeOptions(
-                pool: _memoryPool,
-                readerScheduler: PipeScheduler.ThreadPool,
-                writerScheduler: PipeScheduler.Inline,
-                useSynchronizationContext: false
-            );
-
-            _pair = DuplexPipe.CreateConnectionPair(inputPipeOptions, outputPipeOptions);
             _hpackDecoder = new HPackDecoder((int)_clientSettings.HeaderTableSize, MaxRequestHeaderFieldSize);
 
             _timeoutControl = new TimeoutControl(_mockTimeoutHandler.Object);
             _mockTimeoutControl = new Mock<MockTimeoutControlBase>(_timeoutControl) { CallBase = true };
+            _timeoutControl.Debugger = Mock.Of<IDebugger>();
+
+            _mockKestrelTrace
+                .Setup(m => m.Http2ConnectionClosing(It.IsAny<string>()))
+                .Callback(() => _closingStateReached.SetResult(null));
+            _mockKestrelTrace
+                .Setup(m => m.Http2ConnectionClosed(It.IsAny<string>(), It.IsAny<int>()))
+                .Callback(() => _closedStateReached.SetResult(null));
+
+            _mockConnectionContext.Setup(c => c.Abort(It.IsAny<ConnectionAbortedException>())).Callback<ConnectionAbortedException>(ex =>
+            {
+                // Emulate transport abort so the _connectionTask completes.
+                _pair.Application.Output.Complete(ex);
+            });
 
             _noopApplication = context => Task.CompletedTask;
 
@@ -277,44 +334,30 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Tests
 
                 return Task.CompletedTask;
             };
+
+            _appAbort = context =>
+            {
+                context.Abort();
+                return Task.CompletedTask;
+            };
         }
 
         public override void Initialize(MethodInfo methodInfo, object[] testMethodArguments, ITestOutputHelper testOutputHelper)
         {
             base.Initialize(methodInfo, testMethodArguments, testOutputHelper);
 
-            var mockKestrelTrace = new Mock<IKestrelTrace>();
-            mockKestrelTrace
-                .Setup(m => m.Http2ConnectionClosing(It.IsAny<string>()))
-                .Callback(() => _closingStateReached.SetResult(null));
-            mockKestrelTrace
-                .Setup(m => m.Http2ConnectionClosed(It.IsAny<string>(), It.IsAny<int>()))
-                .Callback(() => _closedStateReached.SetResult(null));
-
-            _connectionContext = new HttpConnectionContext
+            _serviceContext = new TestServiceContext(LoggerFactory, _mockKestrelTrace.Object)
             {
-                ConnectionContext = _mockConnectionContext.Object,
-                ConnectionFeatures = new FeatureCollection(),
-                ServiceContext = new TestServiceContext(LoggerFactory, mockKestrelTrace.Object),
-                MemoryPool = _memoryPool,
-                Transport = _pair.Transport,
-                TimeoutControl = _mockTimeoutControl.Object
+                Scheduler = PipeScheduler.Inline
             };
-
-            _connection = new Http2Connection(_connectionContext);
-
-            var httpConnection = new HttpConnection(_connectionContext);
-            httpConnection.Initialize(_connection);
-            _mockTimeoutHandler.Setup(h => h.OnTimeout(It.IsAny<TimeoutReason>()))
-                               .Callback<TimeoutReason>(r => httpConnection.OnTimeout(r));
         }
 
         public override void Dispose()
         {
-            _pair.Application.Input.Complete();
-            _pair.Application.Output.Complete();
-            _pair.Transport.Input.Complete();
-            _pair.Transport.Output.Complete();
+            _pair.Application?.Input.Complete();
+            _pair.Application?.Output.Complete();
+            _pair.Transport?.Input.Complete();
+            _pair.Transport?.Output.Complete();
             _memoryPool.Dispose();
 
             base.Dispose();
@@ -325,8 +368,43 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Tests
             _decodedHeaders[name.GetAsciiStringNonNullCharacters()] = value.GetAsciiOrUTF8StringNonNullCharacters();
         }
 
+        protected void CreateConnection()
+        {
+            var limits = _serviceContext.ServerOptions.Limits;
+
+            // Always dispatch test code back to the ThreadPool. This prevents deadlocks caused by continuing
+            // Http2Connection.ProcessRequestsAsync() loop with writer locks acquired. Run product code inline to make
+            // it easier to verify request frames are processed correctly immediately after sending the them.
+            var inputPipeOptions = ConnectionDispatcher.GetInputPipeOptions(_serviceContext, _memoryPool, PipeScheduler.ThreadPool);
+            var outputPipeOptions = ConnectionDispatcher.GetOutputPipeOptions(_serviceContext, _memoryPool, PipeScheduler.ThreadPool);
+
+            _pair = DuplexPipe.CreateConnectionPair(inputPipeOptions, outputPipeOptions);
+
+            var httpConnectionContext = new HttpConnectionContext
+            {
+                ConnectionContext = _mockConnectionContext.Object,
+                ConnectionFeatures = new FeatureCollection(),
+                ServiceContext = _serviceContext,
+                MemoryPool = _memoryPool,
+                Transport = _pair.Transport,
+                TimeoutControl = _mockTimeoutControl.Object
+            };
+
+            _connection = new Http2Connection(httpConnectionContext);
+
+            var httpConnection = new HttpConnection(httpConnectionContext);
+            httpConnection.Initialize(_connection);
+            _mockTimeoutHandler.Setup(h => h.OnTimeout(It.IsAny<TimeoutReason>()))
+                               .Callback<TimeoutReason>(r => httpConnection.OnTimeout(r));
+        }
+
         protected async Task InitializeConnectionAsync(RequestDelegate application, int expectedSettingsCount = 3)
         {
+            if (_connection == null)
+            {
+                CreateConnection();
+            }
+
             _connectionTask = _connection.ProcessRequestsAsync(new DummyApplication(application));
 
             await SendPreambleAsync().ConfigureAwait(false);
@@ -716,6 +794,22 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Tests
             await SendAsync(payload);
         }
 
+        protected async Task<bool> SendContinuationAsync(int streamId, Http2ContinuationFrameFlags flags, IEnumerable<KeyValuePair<string, string>> headers)
+        {
+            var outputWriter = _pair.Application.Output;
+            var frame = new Http2Frame();
+
+            frame.PrepareContinuation(flags, streamId);
+            var buffer = _headerEncodingBuffer.AsMemory();
+            var done = _hpackEncoder.BeginEncode(headers, buffer.Span, out var length);
+            frame.PayloadLength = length;
+
+            Http2FrameWriter.WriteHeader(frame, outputWriter);
+            await SendAsync(buffer.Span.Slice(0, length));
+
+            return done;
+        }
+
         protected Task SendEmptyContinuationFrameAsync(int streamId, Http2ContinuationFrameFlags flags)
         {
             var outputWriter = _pair.Application.Output;
diff --git a/test/Kestrel.InMemory.FunctionalTests/Http2/Http2TimeoutTests.cs b/test/Kestrel.InMemory.FunctionalTests/Http2/Http2TimeoutTests.cs
index 1fc51cfb12..147c153226 100644
--- a/test/Kestrel.InMemory.FunctionalTests/Http2/Http2TimeoutTests.cs
+++ b/test/Kestrel.InMemory.FunctionalTests/Http2/Http2TimeoutTests.cs
@@ -2,11 +2,13 @@
 // Licensed under the Apache License, Version 2.0. See License.txt in the project root for license information.
 
 using System;
+using System.Collections.Generic;
 using System.Threading.Tasks;
 using Microsoft.AspNetCore.Connections;
+using Microsoft.AspNetCore.Server.Kestrel.Core.Internal;
 using Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http2;
 using Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Infrastructure;
-using Microsoft.AspNetCore.Testing;
+using Microsoft.Net.Http.Headers;
 using Moq;
 using Xunit;
 
@@ -17,8 +19,8 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Tests
         [Fact]
         public async Task HEADERS_NotReceivedInitially_WithinKeepAliveTimeout_ClosesConnection()
         {
-            var mockSystemClock = new MockSystemClock();
-            var limits = _connectionContext.ServiceContext.ServerOptions.Limits;
+            var mockSystemClock = _serviceContext.MockSystemClock;
+            var limits = _serviceContext.ServerOptions.Limits;
 
             _timeoutControl.Initialize(mockSystemClock.UtcNow);
 
@@ -40,8 +42,8 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Tests
         [Fact]
         public async Task HEADERS_NotReceivedAfterFirstRequest_WithinKeepAliveTimeout_ClosesConnection()
         {
-            var mockSystemClock = new MockSystemClock();
-            var limits = _connectionContext.ServiceContext.ServerOptions.Limits;
+            var mockSystemClock = _serviceContext.MockSystemClock;
+            var limits = _serviceContext.ServerOptions.Limits;
 
             _timeoutControl.Initialize(mockSystemClock.UtcNow);
 
@@ -97,14 +99,8 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Tests
         [Fact]
         public async Task HEADERS_ReceivedWithoutAllCONTINUATIONs_WithinRequestHeadersTimeout_AbortsConnection()
         {
-            var mockSystemClock = new MockSystemClock();
-            var limits = _connectionContext.ServiceContext.ServerOptions.Limits;
-
-            _mockConnectionContext.Setup(c => c.Abort(It.IsAny<ConnectionAbortedException>())).Callback<ConnectionAbortedException>(ex =>
-            {
-                // Emulate transport abort so the _connectionTask completes.
-                _pair.Application.Output.Complete(ex);
-            });
+            var mockSystemClock = _serviceContext.MockSystemClock;
+            var limits = _serviceContext.ServerOptions.Limits;;
 
             _timeoutControl.Initialize(mockSystemClock.UtcNow);
 
@@ -139,8 +135,8 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Tests
         [Fact]
         public async Task ResponseDrain_SlowerThanMinimumDataRate_AbortsConnection()
         {
-            var mockSystemClock = new MockSystemClock();
-            var limits = _connectionContext.ServiceContext.ServerOptions.Limits;
+            var mockSystemClock = _serviceContext.MockSystemClock;
+            var limits = _serviceContext.ServerOptions.Limits;
 
             _timeoutControl.Initialize(mockSystemClock.UtcNow);
 
@@ -166,5 +162,314 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Tests
             _mockConnectionContext.Verify(c =>c.Abort(It.Is<ConnectionAbortedException>(e => 
                 e.Message == CoreStrings.ConnectionTimedBecauseResponseMininumDataRateNotSatisfied)), Times.Once);
         }
+
+        [Theory]
+        [InlineData(Http2FrameType.DATA)]
+        [InlineData(Http2FrameType.HEADERS)]
+        public async Task AbortedStream_ResetsAndDrainsRequest_RefusesFramesAfterCooldownExpires(Http2FrameType finalFrameType)
+        {
+            var mockSystemClock = _serviceContext.MockSystemClock;
+
+            var headers = new[]
+            {
+                new KeyValuePair<string, string>(HeaderNames.Method, "POST"),
+                new KeyValuePair<string, string>(HeaderNames.Path, "/"),
+                new KeyValuePair<string, string>(HeaderNames.Scheme, "http"),
+            };
+            await InitializeConnectionAsync(_appAbort);
+
+            await StartStreamAsync(1, headers, endStream: false);
+
+            await WaitForStreamErrorAsync(1, Http2ErrorCode.INTERNAL_ERROR, "The connection was aborted by the application.");
+
+            // There's a race when the appfunc is exiting about how soon it unregisters the stream.
+            for (var i = 0; i < 10; i++)
+            {
+                await SendDataAsync(1, new byte[100], endStream: false);
+            }
+
+            // Just short of the timeout
+            mockSystemClock.UtcNow += Constants.RequestBodyDrainTimeout;
+            (_connection as IRequestProcessor).Tick(mockSystemClock.UtcNow);
+
+            // Still fine
+            await SendDataAsync(1, new byte[100], endStream: false);
+
+            // Just past the timeout
+            mockSystemClock.UtcNow += TimeSpan.FromTicks(1);
+            (_connection as IRequestProcessor).Tick(mockSystemClock.UtcNow);
+
+            // Send an extra frame to make it fail
+            switch (finalFrameType)
+            {
+                case Http2FrameType.DATA:
+                    await SendDataAsync(1, new byte[100], endStream: true);
+                    break;
+
+                case Http2FrameType.HEADERS:
+                    await SendHeadersAsync(1, Http2HeadersFrameFlags.END_STREAM | Http2HeadersFrameFlags.END_HEADERS, _requestTrailers);
+                    break;
+
+                default:
+                    throw new NotImplementedException(finalFrameType.ToString());
+            }
+
+            await WaitForConnectionErrorAsync<Http2ConnectionErrorException>(ignoreNonGoAwayFrames: false, expectedLastStreamId: 1, Http2ErrorCode.STREAM_CLOSED,
+                CoreStrings.FormatHttp2ErrorStreamClosed(finalFrameType, 1));
+        }
+
+        [Fact]
+        public async Task DATA_Sent_TooSlowlyDueToSocketBackPressureOnSmallWrite_AbortsConnectionAfterGracePeriod()
+        {
+            var mockSystemClock = _serviceContext.MockSystemClock;
+            var limits = _serviceContext.ServerOptions.Limits;
+
+            // Disable response buffering so "socket" backpressure is observed immediately.
+            limits.MaxResponseBufferSize = 0;
+
+            _timeoutControl.Initialize(mockSystemClock.UtcNow);
+
+            await InitializeConnectionAsync(_echoApplication);
+            
+            await StartStreamAsync(1, _browserRequestHeaders, endStream: false);
+            await SendDataAsync(1, _helloWorldBytes, endStream: true);
+
+            await ExpectAsync(Http2FrameType.HEADERS,
+                withLength: 37,
+                withFlags: (byte)Http2HeadersFrameFlags.END_HEADERS,
+                withStreamId: 1);
+
+            // Don't read data frame to induce "socket" backpressure.
+            mockSystemClock.UtcNow += limits.MinResponseDataRate.GracePeriod + Heartbeat.Interval;
+            _timeoutControl.Tick(mockSystemClock.UtcNow);
+
+            _mockTimeoutHandler.Verify(h => h.OnTimeout(It.IsAny<TimeoutReason>()), Times.Never);
+
+            mockSystemClock.UtcNow += TimeSpan.FromTicks(1);
+            _timeoutControl.Tick(mockSystemClock.UtcNow);
+
+            _mockTimeoutHandler.Verify(h => h.OnTimeout(TimeoutReason.WriteDataRate), Times.Once);
+
+            // The "hello, world" bytes are buffered from before the timeout, but not an END_STREAM data frame.
+            await ExpectAsync(Http2FrameType.DATA,
+                withLength: _helloWorldBytes.Length,
+                withFlags: (byte)Http2DataFrameFlags.NONE,
+                withStreamId: 1);
+
+            await WaitForConnectionErrorAsync<ConnectionAbortedException>(
+                ignoreNonGoAwayFrames: false,
+                expectedLastStreamId: 1,
+                Http2ErrorCode.INTERNAL_ERROR,
+                null);
+
+            _mockConnectionContext.Verify(c =>c.Abort(It.Is<ConnectionAbortedException>(e => 
+                e.Message == CoreStrings.ConnectionTimedBecauseResponseMininumDataRateNotSatisfied)), Times.Once);
+        }
+
+        [Fact]
+        public async Task DATA_Sent_TooSlowlyDueToSocketBackPressureOnLargeWrite_AbortsConnectionAfterRateTimeout()
+        {
+            var mockSystemClock = _serviceContext.MockSystemClock;
+            var limits = _serviceContext.ServerOptions.Limits;
+
+            // Disable response buffering so "socket" backpressure is observed immediately.
+            limits.MaxResponseBufferSize = 0;
+
+            _timeoutControl.Initialize(mockSystemClock.UtcNow);
+
+            await InitializeConnectionAsync(_echoApplication);
+
+            await StartStreamAsync(1, _browserRequestHeaders, endStream: false);
+            await SendDataAsync(1, _maxData, endStream: true);
+
+            await ExpectAsync(Http2FrameType.HEADERS,
+                withLength: 37,
+                withFlags: (byte)Http2HeadersFrameFlags.END_HEADERS,
+                withStreamId: 1);
+
+            var timeToWriteMaxData = TimeSpan.FromSeconds(_maxData.Length / limits.MinResponseDataRate.BytesPerSecond);
+
+            // Don't read data frame to induce "socket" backpressure.
+            mockSystemClock.UtcNow += timeToWriteMaxData + Heartbeat.Interval;
+            _timeoutControl.Tick(mockSystemClock.UtcNow);
+
+            _mockTimeoutHandler.Verify(h => h.OnTimeout(It.IsAny<TimeoutReason>()), Times.Never);
+
+            mockSystemClock.UtcNow += TimeSpan.FromTicks(1);
+            _timeoutControl.Tick(mockSystemClock.UtcNow);
+
+            _mockTimeoutHandler.Verify(h => h.OnTimeout(TimeoutReason.WriteDataRate), Times.Once);
+
+            // The "hello, world" bytes are buffered from before the timeout, but not an END_STREAM data frame.
+            await ExpectAsync(Http2FrameType.DATA,
+                withLength: _maxData.Length,
+                withFlags: (byte)Http2DataFrameFlags.NONE,
+                withStreamId: 1);
+
+            await WaitForConnectionErrorAsync<ConnectionAbortedException>(
+                ignoreNonGoAwayFrames: false,
+                expectedLastStreamId: 1,
+                Http2ErrorCode.INTERNAL_ERROR,
+                null);
+
+            _mockConnectionContext.Verify(c => c.Abort(It.Is<ConnectionAbortedException>(e =>
+                 e.Message == CoreStrings.ConnectionTimedBecauseResponseMininumDataRateNotSatisfied)), Times.Once);
+        }
+
+        [Fact]
+        public async Task DATA_Sent_TooSlowlyDueToFlowControlOnSmallWrite_AbortsConnectionAfterGracePeriod()
+        {
+            var mockSystemClock = _serviceContext.MockSystemClock;
+            var limits = _serviceContext.ServerOptions.Limits;
+
+            // This only affects the stream windows. The connection-level window is always initialized at 64KiB.
+            _clientSettings.InitialWindowSize = 6;
+
+            _timeoutControl.Initialize(mockSystemClock.UtcNow);
+
+            await InitializeConnectionAsync(_echoApplication);
+
+            await StartStreamAsync(1, _browserRequestHeaders, endStream: false);
+            await SendDataAsync(1, _helloWorldBytes, endStream: true);
+
+            await ExpectAsync(Http2FrameType.HEADERS,
+                withLength: 37,
+                withFlags: (byte)Http2HeadersFrameFlags.END_HEADERS,
+                withStreamId: 1);
+            await ExpectAsync(Http2FrameType.DATA,
+                withLength: (int)_clientSettings.InitialWindowSize,
+                withFlags: (byte)Http2DataFrameFlags.NONE,
+                withStreamId: 1);
+
+            // Don't send WINDOW_UPDATE to induce flow-control backpressure
+            mockSystemClock.UtcNow += limits.MinResponseDataRate.GracePeriod + Heartbeat.Interval;
+            _timeoutControl.Tick(mockSystemClock.UtcNow);
+
+            _mockTimeoutHandler.Verify(h => h.OnTimeout(It.IsAny<TimeoutReason>()), Times.Never);
+
+            mockSystemClock.UtcNow += TimeSpan.FromTicks(1);
+            _timeoutControl.Tick(mockSystemClock.UtcNow);
+
+            _mockTimeoutHandler.Verify(h => h.OnTimeout(TimeoutReason.WriteDataRate), Times.Once);
+
+            await WaitForConnectionErrorAsync<ConnectionAbortedException>(
+                ignoreNonGoAwayFrames: false,
+                expectedLastStreamId: 1,
+                Http2ErrorCode.INTERNAL_ERROR,
+                null);
+
+            _mockConnectionContext.Verify(c => c.Abort(It.Is<ConnectionAbortedException>(e =>
+                 e.Message == CoreStrings.ConnectionTimedBecauseResponseMininumDataRateNotSatisfied)), Times.Once);
+        }
+
+        [Fact]
+        public async Task DATA_Sent_TooSlowlyDueToOutputFlowControlOnLargeWrite_AbortsConnectionAfterRateTimeout()
+        {
+            var mockSystemClock = _serviceContext.MockSystemClock;
+            var limits = _serviceContext.ServerOptions.Limits;
+
+            // This only affects the stream windows. The connection-level window is always initialized at 64KiB.
+            _clientSettings.InitialWindowSize = (uint)_maxData.Length - 1;
+
+            _timeoutControl.Initialize(mockSystemClock.UtcNow);
+
+            await InitializeConnectionAsync(_echoApplication);
+
+            await StartStreamAsync(1, _browserRequestHeaders, endStream: false);
+            await SendDataAsync(1, _maxData, endStream: true);
+
+            await ExpectAsync(Http2FrameType.HEADERS,
+                withLength: 37,
+                withFlags: (byte)Http2HeadersFrameFlags.END_HEADERS,
+                withStreamId: 1);
+            await ExpectAsync(Http2FrameType.DATA,
+                withLength: (int)_clientSettings.InitialWindowSize,
+                withFlags: (byte)Http2DataFrameFlags.NONE,
+                withStreamId: 1);
+
+            var timeToWriteMaxData = TimeSpan.FromSeconds(_clientSettings.InitialWindowSize / limits.MinResponseDataRate.BytesPerSecond);
+
+            // Don't send WINDOW_UPDATE to induce flow-control backpressure
+            mockSystemClock.UtcNow += timeToWriteMaxData + Heartbeat.Interval;
+            _timeoutControl.Tick(mockSystemClock.UtcNow);
+
+            _mockTimeoutHandler.Verify(h => h.OnTimeout(It.IsAny<TimeoutReason>()), Times.Never);
+
+            mockSystemClock.UtcNow += TimeSpan.FromTicks(1);
+            _timeoutControl.Tick(mockSystemClock.UtcNow);
+
+            _mockTimeoutHandler.Verify(h => h.OnTimeout(TimeoutReason.WriteDataRate), Times.Once);
+
+            await WaitForConnectionErrorAsync<ConnectionAbortedException>(
+                ignoreNonGoAwayFrames: false,
+                expectedLastStreamId: 1,
+                Http2ErrorCode.INTERNAL_ERROR,
+                null);
+
+            _mockConnectionContext.Verify(c => c.Abort(It.Is<ConnectionAbortedException>(e =>
+                 e.Message == CoreStrings.ConnectionTimedBecauseResponseMininumDataRateNotSatisfied)), Times.Once);
+        }
+
+        [Fact]
+        public async Task DATA_Sent_TooSlowlyDueToOutputFlowControlOnMultipleStreams_AbortsConnectionAfterAdditiveRateTimeout()
+        {
+            var mockSystemClock = _serviceContext.MockSystemClock;
+            var limits = _serviceContext.ServerOptions.Limits;
+
+            // This only affects the stream windows. The connection-level window is always initialized at 64KiB.
+            _clientSettings.InitialWindowSize = (uint)_maxData.Length - 1;
+
+            _timeoutControl.Initialize(mockSystemClock.UtcNow);
+
+            await InitializeConnectionAsync(_echoApplication);
+
+            await StartStreamAsync(1, _browserRequestHeaders, endStream: false);
+            await SendDataAsync(1, _maxData, endStream: true);
+
+            await ExpectAsync(Http2FrameType.HEADERS,
+                withLength: 37,
+                withFlags: (byte)Http2HeadersFrameFlags.END_HEADERS,
+                withStreamId: 1);
+            await ExpectAsync(Http2FrameType.DATA,
+                withLength: (int)_clientSettings.InitialWindowSize,
+                withFlags: (byte)Http2DataFrameFlags.NONE,
+                withStreamId: 1);
+
+            await StartStreamAsync(3, _browserRequestHeaders, endStream: false);
+            await SendDataAsync(3, _maxData, endStream: true);
+
+            await ExpectAsync(Http2FrameType.HEADERS,
+                withLength: 37,
+                withFlags: (byte)Http2HeadersFrameFlags.END_HEADERS,
+                withStreamId: 3);
+            await ExpectAsync(Http2FrameType.DATA,
+                withLength: (int)_clientSettings.InitialWindowSize,
+                withFlags: (byte)Http2DataFrameFlags.NONE,
+                withStreamId: 3);
+
+            var timeToWriteMaxData = TimeSpan.FromSeconds(_clientSettings.InitialWindowSize / limits.MinResponseDataRate.BytesPerSecond);
+            // Double the timeout for the second stream.
+            timeToWriteMaxData += timeToWriteMaxData;
+
+            // Don't send WINDOW_UPDATE to induce flow-control backpressure
+            mockSystemClock.UtcNow += timeToWriteMaxData + Heartbeat.Interval;
+            _timeoutControl.Tick(mockSystemClock.UtcNow);
+
+            _mockTimeoutHandler.Verify(h => h.OnTimeout(It.IsAny<TimeoutReason>()), Times.Never);
+
+            mockSystemClock.UtcNow += TimeSpan.FromTicks(1);
+            _timeoutControl.Tick(mockSystemClock.UtcNow);
+
+            _mockTimeoutHandler.Verify(h => h.OnTimeout(TimeoutReason.WriteDataRate), Times.Once);
+
+            await WaitForConnectionErrorAsync<ConnectionAbortedException>(
+                ignoreNonGoAwayFrames: false,
+                expectedLastStreamId: 3,
+                Http2ErrorCode.INTERNAL_ERROR,
+                null);
+
+            _mockConnectionContext.Verify(c => c.Abort(It.Is<ConnectionAbortedException>(e =>
+                 e.Message == CoreStrings.ConnectionTimedBecauseResponseMininumDataRateNotSatisfied)), Times.Once);
+        }
     }
 }
diff --git a/test/Kestrel.InMemory.FunctionalTests/HttpProtocolSelectionTests.cs b/test/Kestrel.InMemory.FunctionalTests/HttpProtocolSelectionTests.cs
index 22994a3130..a5765579f5 100644
--- a/test/Kestrel.InMemory.FunctionalTests/HttpProtocolSelectionTests.cs
+++ b/test/Kestrel.InMemory.FunctionalTests/HttpProtocolSelectionTests.cs
@@ -35,35 +35,24 @@ namespace Microsoft.AspNetCore.Server.Kestrel.InMemory.FunctionalTests
         }
 
         [Fact]
-        public async Task Server_Http2Only_Cleartext_Success()
+        public Task Server_Http2Only_Cleartext_Success()
         {
-            // Expect a SETTINGS frame (type 0x4) with default settings
+            // Expect a SETTINGS frame with default settings then a connection-level WINDOW_UPDATE frame.
             var expected = new byte[]
             {
                 0x00, 0x00, 0x12, // Payload Length (6 * settings count)
                 0x04, 0x00, 0x00, 0x00, 0x00, 0x00, // SETTINGS frame (type 0x04)
-                0x00, 0x03, 0x00, 0x00, 0x00, 0x64, // Connection limit
-                0x00, 0x04, 0x00, 0x01, 0x80, 0x00, // Initial window size
-                0x00, 0x06, 0x00, 0x00, 0x80, 0x00 // Header size limit
-            };
-            var testContext = new TestServiceContext(LoggerFactory);
-            var listenOptions = new ListenOptions(new IPEndPoint(IPAddress.Loopback, 0))
-            {
-                Protocols = HttpProtocols.Http2
+                0x00, 0x03, 0x00, 0x00, 0x00, 0x64, // Connection limit (100)
+                0x00, 0x04, 0x00, 0x01, 0x80, 0x00, // Initial stream window size (96 KiB)
+                0x00, 0x06, 0x00, 0x00, 0x80, 0x00, // Header size limit (32 KiB)
+                0x00, 0x00, 0x04, // Payload Length (4)
+                0x08, 0x00, 0x00, 0x00, 0x00, 0x00, // WINDOW_UPDATE frame (type 0x08)
+                0x00, 0x01, 0x00, 0x01, // Diff between configured and protocol default (128 KiB - 0XFFFF)
             };
 
-            using (var server = new TestServer(context => Task.CompletedTask, testContext, listenOptions))
-            {
-                using (var connection = server.CreateConnection())
-                {
-                    await connection.Send(Encoding.ASCII.GetString(Http2Connection.ClientPreface));
-                    // Can't use Receive when expecting binary data
-                    var actual = new byte[expected.Length];
-                    var read = await connection.Stream.ReadAsync(actual, 0, actual.Length);
-                    Assert.Equal(expected.Length, read);
-                    Assert.Equal(expected, actual);
-                }
-            }
+            return TestSuccess(HttpProtocols.Http2,
+                Encoding.ASCII.GetString(Http2Connection.ClientPreface),
+                Encoding.ASCII.GetString(expected));
         }
 
         private async Task TestSuccess(HttpProtocols serverProtocols, string request, string expectedResponse)