Track aborted streams for a given grace period #2832

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
             (

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.
+                if (stream.IsDraining)
-                stream.DisallowAdditionalRequestFrames();
+                {
-                stream.Abort(new IOException(CoreStrings.Http2StreamResetByClient));
+                    // 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 _);
+        }
     }
 }

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.
+                    if (IsAborted)
-                    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)
                     {
-                        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)
+                        InputRemaining -= dataPayload.Length;
-                {
+                    }
-                    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
+                    foreach (var segment in dataPayload)
-                // _inputFlowControl.Advance() didn't throw.
+                    {
-                Debug.Assert(flushTask.IsCompleted);
+                        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
+            _inputFlowControl.StopWindowUpdates();
-            {
-                RequestBodyPipe.Writer.Complete();
-
-                _inputFlowControl.StopWindowUpdates();
-            }
-            finally
-            {
-                TryFireOnStreamCompleted(states);
-            }
         }
 
         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);
+            AbortCore(abortReason);
-            }
-            finally
-            {
-                TryFireOnStreamCompleted(states);
-            }
         }
 
         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.
+            // Don't block on IO. This never faults.
-                _ = _http2Output.WriteRstStreamAsync(error);
+            _ = _http2Output.WriteRstStreamAsync(error);
 
-                AbortCore(abortReason);
+            AbortCore(abortReason);
-            }
-            finally
-            {
-                TryFireOnStreamCompleted(states);
-            }
         }
 
         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,
         }
     }
 }

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)

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);
     }
 }

test/Kestrel.InMemory.FunctionalTests/Http2/Http2ConnectionTests.cs

@@ -10,9 +10,12 @@ using System.Text;
 using System.Threading;
 using System.Threading.Tasks;
 using Microsoft.AspNetCore.Connections;
+using Microsoft.AspNetCore.Http;
 using Microsoft.AspNetCore.Server.Kestrel.Core.Features;
+using Microsoft.AspNetCore.Server.Kestrel.Core.Internal;
 using Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http2;
 using Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http2.HPack;
+using Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Infrastructure;
 using Microsoft.AspNetCore.Testing;
 using Microsoft.Extensions.Logging;
 using Microsoft.Net.Http.Headers;
@@ -553,10 +556,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();
@@ -716,6 +719,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);
@@ -3786,6 +3793,269 @@ 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));
+        }
+
+        [Theory]
+        [InlineData(Http2FrameType.DATA)]
+        [InlineData(Http2FrameType.HEADERS)]
+        public async Task AbortedStream_ResetsAndDrainsRequest_RefusesFramesAfterCooldownExpires(Http2FrameType finalFrameType)
+        {
+            var mockSystemClock = new MockSystemClock();
+            _connectionContext.ServiceContext.SystemClock = 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 - TimeSpan.FromTicks(1);
+            (_connection as IRequestProcessor).Tick(mockSystemClock.UtcNow);
+
+            // Still fine
+            await SendDataAsync(1, new byte[100], endStream: false);
+
+            // Just past the timeout
+            mockSystemClock.UtcNow += TimeSpan.FromTicks(2);
+            (_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));
+        }
+
         public static TheoryData<byte[]> UpperCaseHeaderNameData
         {
             get

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);
@@ -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);
@@ -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);
@@ -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);
@@ -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);

test/Kestrel.InMemory.FunctionalTests/Http2/Http2TestBase.cs

@@ -82,6 +82,7 @@ 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 Http2Connection _connection;
@@ -277,6 +278,12 @@ 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)
@@ -716,6 +723,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;