Don't throw Exceptions from PipeWriter APIs after RST_STREAM (#11675)

* Don't throw InvalidOperationExceptions from PipeWriter APIs after HTTP/2 aborts

src/Servers/Kestrel/Core/src/Internal/Http/Http1OutputProducer.cs

@@ -41,7 +41,6 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http
         private readonly object _contextLock = new object();
 
         private bool _pipeWriterCompleted;
-        private bool _completed;
         private bool _aborted;
         private long _unflushedBytes;
         private int _currentMemoryPrefixBytes;
@@ -56,6 +55,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http
         // and append the end terminator.
 
         private bool _autoChunk;
+        private bool _suffixSent;
         private int _advancedBytesForChunk;
         private Memory<byte> _currentChunkMemory;
         private bool _currentChunkMemoryUpdated;
@@ -111,7 +111,18 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http
 
         public ValueTask<FlushResult> WriteStreamSuffixAsync()
         {
-            return WriteAsync(EndChunkedResponseBytes);
+            lock (_contextLock)
+            {
+                if (_suffixSent || !_autoChunk)
+                {
+                    _suffixSent = true;
+                    return FlushAsync();
+                }
+
+                _suffixSent = true;
+                var writer = new BufferWriter<PipeWriter>(_pipeWriter);
+                return WriteAsyncInternal(ref writer, EndChunkedResponseBytes);
+            }
         }
 
         public ValueTask<FlushResult> FlushAsync(CancellationToken cancellationToken = default)
@@ -146,7 +157,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http
 
             ValueTask<FlushResult> FlushAsyncChunked(Http1OutputProducer producer, CancellationToken token)
             {
-                // Local function so in the common-path the stack space for BufferWriter isn't reservered and cleared when it isn't used.
+                // Local function so in the common-path the stack space for BufferWriter isn't reserved and cleared when it isn't used.
 
                 Debug.Assert(!producer._pipeWriterCompleted);
                 Debug.Assert(producer._autoChunk && producer._advancedBytesForChunk > 0);
@@ -169,7 +180,9 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http
         {
             lock (_contextLock)
             {
-                if (_completed)
+                ThrowIfSuffixSent();
+
+                if (_pipeWriterCompleted)
                 {
                     return GetFakeMemory(sizeHint);
                 }
@@ -192,7 +205,9 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http
         {
             lock (_contextLock)
             {
-                if (_completed)
+                ThrowIfSuffixSent();
+
+                if (_pipeWriterCompleted)
                 {
                     return GetFakeMemory(sizeHint).Span;
                 }
@@ -215,7 +230,9 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http
         {
             lock (_contextLock)
             {
-                if (_completed)
+                ThrowIfSuffixSent();
+
+                if (_pipeWriterCompleted)
                 {
                     return;
                 }
@@ -257,6 +274,8 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http
         {
             lock (_contextLock)
             {
+                ThrowIfSuffixSent();
+
                 if (_pipeWriterCompleted)
                 {
                     return default;
@@ -297,6 +316,8 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http
         {
             lock (_contextLock)
             {
+                ThrowIfSuffixSent();
+
                 if (_pipeWriterCompleted)
                 {
                     return;
@@ -404,7 +425,6 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http
             {
                 _log.ConnectionDisconnect(_connectionId);
                 _pipeWriterCompleted = true;
-                _completed = true;
             }
         }
 
@@ -426,11 +446,11 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http
             }
         }
 
-        public void Complete()
+        public void Stop()
         {
             lock (_contextLock)
             {
-                _completed = true;
+                CompletePipe();
             }
         }
 
@@ -443,6 +463,8 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http
         {
             lock (_contextLock)
             {
+                ThrowIfSuffixSent();
+
                 if (_pipeWriterCompleted)
                 {
                     return default;
@@ -461,6 +483,8 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http
         {
             lock (_contextLock)
             {
+                ThrowIfSuffixSent();
+
                 if (_pipeWriterCompleted)
                 {
                     return default;
@@ -486,6 +510,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http
             // Cleared in sequential address ascending order 
             _currentMemoryPrefixBytes = 0;
             _autoChunk = false;
+            _suffixSent = false;
             _currentChunkMemoryUpdated = false;
             _startCalled = false;
         }
@@ -496,6 +521,8 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http
         {
             lock (_contextLock)
             {
+                ThrowIfSuffixSent();
+
                 if (_pipeWriterCompleted)
                 {
                     return default;
@@ -671,6 +698,14 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http
             _position = 0;
         }
 
+        [StackTraceHidden]
+        private void ThrowIfSuffixSent()
+        {
+            if (_suffixSent)
+            {
+                throw new InvalidOperationException("Writing is not allowed after writer was completed.");
+            }
+        }
 
         /// <summary>
         /// Holds a byte[] from the pool and a size value. Basically a Memory but guaranteed to be backed by an ArrayPool byte[], so that we know we can return it.

src/Servers/Kestrel/Core/src/Internal/Http/HttpProtocol.FeatureCollection.cs

@@ -319,7 +319,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http
 
             await FlushAsync();
 
-            return bodyControl.Upgrade();
+            return _bodyControl.Upgrade();
         }
 
         void IHttpRequestLifetimeFeature.Abort()

src/Servers/Kestrel/Core/src/Internal/Http/HttpProtocol.cs

@@ -32,7 +32,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http
         private static readonly byte[] _bytesTransferEncodingChunked = Encoding.ASCII.GetBytes("\r\nTransfer-Encoding: chunked");
         private static readonly byte[] _bytesServer = Encoding.ASCII.GetBytes("\r\nServer: " + Constants.ServerName);
 
-        protected BodyControl bodyControl;
+        protected BodyControl _bodyControl;
         private Stack<KeyValuePair<Func<object, Task>, object>> _onStarting;
         private Stack<KeyValuePair<Func<object, Task>, object>> _onCompleted;
 
@@ -315,18 +315,16 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http
 
         public void InitializeBodyControl(MessageBody messageBody)
         {
-            if (bodyControl == null)
+            if (_bodyControl == null)
             {
-                bodyControl = new BodyControl(bodyControl: this, this);
+                _bodyControl = new BodyControl(bodyControl: this, this);
             }
 
-            (RequestBody, ResponseBody, RequestBodyPipeReader, ResponseBodyPipeWriter) = bodyControl.Start(messageBody);
+            (RequestBody, ResponseBody, RequestBodyPipeReader, ResponseBodyPipeWriter) = _bodyControl.Start(messageBody);
             _requestStreamInternal = RequestBody;
             _responseStreamInternal = ResponseBody;
         }
 
-        public void StopBodies() => bodyControl.Stop();
-
         // For testing
         internal void ResetState()
         {
@@ -497,7 +495,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http
 
         protected void PoisonRequestBodyStream(Exception abortReason)
         {
-            bodyControl?.Abort(abortReason);
+            _bodyControl?.Abort(abortReason);
         }
 
         // Prevents the RequestAborted token from firing for the duration of the request.
@@ -666,7 +664,17 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http
 
                 // At this point all user code that needs use to the request or response streams has completed.
                 // Using these streams in the OnCompleted callback is not allowed.
-                StopBodies();
+                try
+                {
+                    await _bodyControl.StopAsync();
+                }
+                catch (Exception ex)
+                {
+                    // BodyControl.StopAsync() can throw if the PipeWriter was completed prior to the application writing
+                    // enough bytes to satisfy the specified Content-Length. This risks double-logging the exception,
+                    // but this scenario generally indicates an app bug, so I don't want to risk not logging it.
+                    ReportApplicationError(ex);
+                }
 
                 // 4XX responses are written by TryProduceInvalidRequestResponse during connection tear down.
                 if (_requestRejectedException == null)
@@ -1019,6 +1027,11 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http
 
         protected Task ProduceEnd()
         {
+            if (HasResponseCompleted)
+            {
+                return Task.CompletedTask;
+            }
+
             if (_requestRejectedException != null || _applicationException != null)
             {
                 if (HasResponseStarted)
@@ -1052,18 +1065,21 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http
 
         private Task WriteSuffix()
         {
-            if (HasResponseCompleted)
-            {
-                return Task.CompletedTask;
-            }
-
-            // _autoChunk should be checked after we are sure ProduceStart() has been called
-            // since ProduceStart() may set _autoChunk to true.
             if (_autoChunk || _httpVersion == Http.HttpVersion.Http2)
             {
-                return WriteSuffixAwaited();
+                // For the same reason we call CheckLastWrite() in Content-Length responses.
+                PreventRequestAbortedCancellation();
             }
 
+            var writeTask = Output.WriteStreamSuffixAsync();
+
+            if (!writeTask.IsCompletedSuccessfully)
+            {
+                return WriteSuffixAwaited(writeTask);
+            }
+
+            _requestProcessingStatus = RequestProcessingStatus.ResponseCompleted;
+
             if (_keepAlive)
             {
                 Log.ConnectionKeepAlive(ConnectionId);
@@ -1074,23 +1090,14 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http
                 Log.ConnectionHeadResponseBodyWrite(ConnectionId, _responseBytesWritten);
             }
 
-            if (!HasFlushedHeaders)
-            {
-                _requestProcessingStatus = RequestProcessingStatus.ResponseCompleted;
-                return FlushAsyncInternal();
-            }
-
             return Task.CompletedTask;
         }
 
-        private async Task WriteSuffixAwaited()
+        private async Task WriteSuffixAwaited(ValueTask<FlushResult> writeTask)
         {
-            // For the same reason we call CheckLastWrite() in Content-Length responses.
-            PreventRequestAbortedCancellation();
-
             _requestProcessingStatus = RequestProcessingStatus.HeadersFlushed;
 
-            await Output.WriteStreamSuffixAsync();
+            await writeTask;
 
             _requestProcessingStatus = RequestProcessingStatus.ResponseCompleted;
 
@@ -1405,11 +1412,11 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http
             Output.CancelPendingFlush();
         }
 
-        public void Complete(Exception ex)
+        public Task CompleteAsync(Exception exception = null)
         {
-            if (ex != null)
+            if (exception != null)
             {
-                var wrappedException = new ConnectionAbortedException("The BodyPipe was completed with an exception.", ex);
+                var wrappedException = new ConnectionAbortedException("The BodyPipe was completed with an exception.", exception);
                 ReportApplicationError(wrappedException);
 
                 if (HasResponseStarted)
@@ -1418,7 +1425,47 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http
                 }
             }
 
-            Output.Complete();
+            // Finalize headers
+            if (!HasResponseStarted)
+            {
+                var onStartingTask = FireOnStarting();
+                if (!onStartingTask.IsCompletedSuccessfully)
+                {
+                    return CompleteAsyncAwaited(onStartingTask);
+                }
+            }
+
+            // Flush headers, body, trailers...
+            if (!HasResponseCompleted)
+            {
+                if (!VerifyResponseContentLength(out var lengthException))
+                {
+                    // Try to throw this exception from CompleteAsync() instead of CompleteAsyncAwaited() if possible,
+                    // so it can be observed by BodyWriter.Complete(). If this isn't possible because an
+                    // async OnStarting callback hadn't yet run, it's OK, since the Exception will be observed with
+                    // the call to _bodyControl.StopAsync() in ProcessRequests().
+                    throw lengthException;
+                }
+
+                return ProduceEnd();
+            }
+
+            return Task.CompletedTask;
+        }
+
+        private async Task CompleteAsyncAwaited(Task onStartingTask)
+        {
+            await onStartingTask;
+
+            if (!HasResponseCompleted)
+            {
+                if (!VerifyResponseContentLength(out var lengthException))
+                {
+                    throw lengthException;
+                }
+
+                await ProduceEnd();
+            }
         }
 
         public ValueTask<FlushResult> WritePipeAsync(ReadOnlyMemory<byte> data, CancellationToken cancellationToken)

src/Servers/Kestrel/Core/src/Internal/Http/HttpResponsePipeWriter.cs

@@ -11,9 +11,11 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http
 {
     internal sealed class HttpResponsePipeWriter : PipeWriter
     {
-        private HttpStreamState _state;
         private readonly IHttpResponseControl _pipeControl;
 
+        private HttpStreamState _state;
+        private Task _completeTask = Task.CompletedTask;
+
         public HttpResponsePipeWriter(IHttpResponseControl pipeControl)
         {
             _pipeControl = pipeControl;
@@ -35,7 +37,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http
         public override void Complete(Exception exception = null)
         {
             ValidateState();
-            _pipeControl.Complete(exception);
+            _completeTask = _pipeControl.CompleteAsync(exception);
         }
 
         public override ValueTask<FlushResult> FlushAsync(CancellationToken cancellationToken = default)
@@ -77,11 +79,12 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http
             }
         }
 
-        public void StopAcceptingWrites()
+        public Task StopAcceptingWritesAsync()
         {
             // Can't use dispose (or close) as can be disposed too early by user code
             // As exampled in EngineTests.ZeroContentLengthNotSetAutomaticallyForCertainStatusCodes
             _state = HttpStreamState.Closed;
+            return _completeTask;
         }
 
         public void Abort()

src/Servers/Kestrel/Core/src/Internal/Http/IHttpOutputProducer.cs

@@ -22,7 +22,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http
         Span<byte> GetSpan(int sizeHint = 0);
         Memory<byte> GetMemory(int sizeHint = 0);
         void CancelPendingFlush();
-        void Complete();
+        void Stop();
         ValueTask<FlushResult> FirstWriteAsync(int statusCode, string reasonPhrase, HttpResponseHeaders responseHeaders, bool autoChunk, ReadOnlySpan<byte> data, CancellationToken cancellationToken);
         ValueTask<FlushResult> FirstWriteChunkedAsync(int statusCode, string reasonPhrase, HttpResponseHeaders responseHeaders, bool autoChunk, ReadOnlySpan<byte> data, CancellationToken cancellationToken);
         void Reset();

src/Servers/Kestrel/Core/src/Internal/Http/IHttpResponseControl.cs

@@ -17,6 +17,6 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http
         ValueTask<FlushResult> FlushPipeAsync(CancellationToken cancellationToken);
         ValueTask<FlushResult> WritePipeAsync(ReadOnlyMemory<byte> source, CancellationToken cancellationToken);
         void CancelPendingFlush();
-        void Complete(Exception exception = null);
+        Task CompleteAsync(Exception exception = null);
     }
 }

src/Servers/Kestrel/Core/src/Internal/Http2/Http2OutputProducer.cs

@@ -3,6 +3,7 @@
 
 using System;
 using System.Buffers;
+using System.Diagnostics;
 using System.IO.Pipelines;
 using System.Threading;
 using System.Threading.Tasks;
@@ -25,15 +26,19 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http2
         // This should only be accessed via the FrameWriter. The connection-level output flow control is protected by the
         // FrameWriter's connection-level write lock.
         private readonly StreamOutputFlowControl _flowControl;
+        private readonly MemoryPool<byte> _memoryPool;
         private readonly Http2Stream _stream;
         private readonly object _dataWriterLock = new object();
         private readonly Pipe _dataPipe;
         private readonly ValueTask<FlushResult> _dataWriteProcessingTask;
         private bool _startedWritingDataFrames;
         private bool _completed;
+        private bool _suffixSent;
         private bool _streamEnded;
         private bool _disposed;
 
+        private IMemoryOwner<byte> _fakeMemoryOwner;
+
         public Http2OutputProducer(
             int streamId,
             Http2FrameWriter frameWriter,
@@ -46,6 +51,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http2
             _streamId = streamId;
             _frameWriter = frameWriter;
             _flowControl = flowControl;
+            _memoryPool = pool;
             _stream = stream;
             _log = log;
 
@@ -65,17 +71,13 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http2
 
                 _disposed = true;
 
-                if (!_completed)
+                Stop();
+
+                if (_fakeMemoryOwner != null)
                 {
-                    _completed = true;
+                    _fakeMemoryOwner.Dispose();
-
+                    _fakeMemoryOwner = null;
-                    // Complete with an exception to prevent an end of stream data frame from being sent without an
-                    // explicit call to WriteStreamSuffixAsync. ConnectionAbortedExceptions are swallowed, so the
-                    // message doesn't matter
-                    _dataPipe.Writer.Complete(new OperationCanceledException());
                 }
-
-                _frameWriter.AbortPendingStreamDataWrites(_flowControl);
             }
         }
 
@@ -84,7 +86,6 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http2
         void IHttpOutputAborter.Abort(ConnectionAbortedException abortReason)
         {
             _stream.ResetAndAbort(abortReason, Http2ErrorCode.INTERNAL_ERROR);
-            Dispose();
         }
 
         public Task WriteChunkAsync(ReadOnlySpan<byte> span, CancellationToken cancellationToken)
@@ -101,6 +102,8 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http2
 
             lock (_dataWriterLock)
             {
+                ThrowIfSuffixSent();
+
                 if (_completed)
                 {
                     return default;
@@ -125,6 +128,8 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http2
         {
             lock (_dataWriterLock)
             {
+                ThrowIfSuffixSent();
+
                 if (_completed)
                 {
                     return default;
@@ -177,6 +182,8 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http2
 
             lock (_dataWriterLock)
             {
+                ThrowIfSuffixSent();
+
                 // This length check is important because we don't want to set _startedWritingDataFrames unless a data
                 // frame will actually be written causing the headers to be flushed.
                 if (_completed || data.Length == 0)
@@ -197,10 +204,11 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http2
             {
                 if (_completed)
                 {
-                    return default;
+                    return _dataWriteProcessingTask;
                 }
 
                 _completed = true;
+                _suffixSent = true;
 
                 _dataPipe.Writer.Complete();
                 return _dataWriteProcessingTask;
@@ -212,8 +220,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http2
             lock (_dataWriterLock)
             {
                 // Always send the reset even if the response body is _completed. The request body may not have completed yet.
-
+                Stop();
-                Dispose();
 
                 return _frameWriter.WriteRstStreamAsync(_streamId, error);
             }
@@ -223,6 +230,13 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http2
         {
             lock (_dataWriterLock)
             {
+                ThrowIfSuffixSent();
+
+                if (_completed)
+                {
+                    return;
+                }
+
                 _startedWritingDataFrames = true;
 
                 _dataPipe.Writer.Advance(bytes);
@@ -233,6 +247,13 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http2
         {
             lock (_dataWriterLock)
             {
+                ThrowIfSuffixSent();
+
+                if (_completed)
+                {
+                    return GetFakeMemory(sizeHint).Span;
+                }
+
                 return _dataPipe.Writer.GetSpan(sizeHint);
             }
         }
@@ -241,6 +262,13 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http2
         {
             lock (_dataWriterLock)
             {
+                ThrowIfSuffixSent();
+
+                if (_completed)
+                {
+                    return GetFakeMemory(sizeHint);
+                }
+
                 return _dataPipe.Writer.GetMemory(sizeHint);
             }
         }
@@ -249,6 +277,11 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http2
         {
             lock (_dataWriterLock)
             {
+                if (_completed)
+                {
+                    return;
+                }
+
                 _dataPipe.Writer.CancelPendingFlush();
             }
         }
@@ -262,6 +295,8 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http2
 
             lock (_dataWriterLock)
             {
+                ThrowIfSuffixSent();
+
                 // This length check is important because we don't want to set _startedWritingDataFrames unless a data
                 // frame will actually be written causing the headers to be flushed.
                 if (_completed || data.Length == 0)
@@ -296,9 +331,24 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http2
             throw new NotImplementedException();
         }
 
-        public void Complete()
+        public void Stop()
         {
-            // This will noop for now. See: https://github.com/aspnet/AspNetCore/issues/7370
+            lock (_dataWriterLock)
+            {
+                if (_completed)
+                {
+                    return;
+                }
+
+                _completed = true;
+
+                // Complete with an exception to prevent an end of stream data frame from being sent without an
+                // explicit call to WriteStreamSuffixAsync. ConnectionAbortedExceptions are swallowed, so the
+                // message doesn't matter
+                _dataPipe.Writer.Complete(new OperationCanceledException());
+
+                _frameWriter.AbortPendingStreamDataWrites(_flowControl);
+            }
         }
 
         public void Reset()
@@ -360,6 +410,25 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http2
             return flushResult;
         }
 
+        private Memory<byte> GetFakeMemory(int sizeHint)
+        {
+            if (_fakeMemoryOwner == null)
+            {
+                _fakeMemoryOwner = _memoryPool.Rent(sizeHint);
+            }
+
+            return _fakeMemoryOwner.Memory;
+        }
+
+        [StackTraceHidden]
+        private void ThrowIfSuffixSent()
+        {
+            if (_suffixSent)
+            {
+                throw new InvalidOperationException("Writing is not allowed after writer was completed.");
+            }
+        }
+
         private static Pipe CreateDataPipe(MemoryPool<byte> pool)
             => new Pipe(new PipeOptions
             (

src/Servers/Kestrel/Core/src/Internal/Http2/Http2Stream.FeatureCollection.cs

@@ -56,24 +56,9 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http2
             }
         }
 
-        async Task IHttpResponseCompletionFeature.CompleteAsync()
+        Task IHttpResponseCompletionFeature.CompleteAsync()
         {
-            // Finalize headers
+            return CompleteAsync();
-            if (!HasResponseStarted)
-            {
-                await FireOnStarting();
-            }
-
-            // Flush headers, body, trailers...
-            if (!HasResponseCompleted)
-            {
-                if (!VerifyResponseContentLength(out var lengthException))
-                {
-                    throw lengthException;
-                }
-
-                await ProduceEnd();
-            }
         }
 
         void IHttpResetFeature.Reset(int errorCode)

src/Servers/Kestrel/Core/src/Internal/Http2/Http2Stream.cs

@@ -469,9 +469,9 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http2
 
         private void AbortCore(Exception abortReason)
         {
-            // Call _http2Output.Dispose() prior to poisoning the request body stream or pipe to
+            // Call _http2Output.Stop() prior to poisoning the request body stream or pipe to
             // ensure that an app that completes early due to the abort doesn't result in header frames being sent.
-            _http2Output.Dispose();
+            _http2Output.Stop();
 
             AbortRequest();
 

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

@@ -4,6 +4,7 @@
 using System;
 using System.IO;
 using System.IO.Pipelines;
+using System.Threading.Tasks;
 using Microsoft.AspNetCore.Http.Features;
 using Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http;
 
@@ -62,11 +63,11 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Infrastructure
             }
         }
 
-        public void Stop()
+        public Task StopAsync()
         {
             _requestReader.StopAcceptingReads();
             _emptyRequestReader.StopAcceptingReads();
-            _responseWriter.StopAcceptingWrites();
+            return _responseWriter.StopAcceptingWritesAsync();
         }
 
         public void Abort(Exception error)

src/Servers/Kestrel/Core/test/BodyControlTests.cs

@@ -112,7 +112,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Tests
 
             var (_, response, requestPipe, responsePipe) = bodyControl.Start(new MockMessageBody());
 
-            bodyControl.Stop();
+            await bodyControl.StopAsync();
 
             Assert.Throws<ObjectDisposedException>(() => requestPipe.AdvanceTo(new SequencePosition()));
             Assert.Throws<ObjectDisposedException>(() => requestPipe.AdvanceTo(new SequencePosition(), new SequencePosition()));
@@ -130,7 +130,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Tests
 
             var (_, response, requestPipe, responsePipe) = bodyControl.Start(new MockMessageBody());
 
-            bodyControl.Stop();
+            await bodyControl.StopAsync();
 
             Assert.Throws<ObjectDisposedException>(() => responsePipe.Advance(1));
             Assert.Throws<ObjectDisposedException>(() => responsePipe.CancelPendingFlush());

src/Servers/Kestrel/Core/test/HttpResponsePipeWriterTests.cs

@@ -23,7 +23,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Tests
         {
             var pipeWriter = CreateHttpResponsePipeWriter();
             pipeWriter.StartAcceptingWrites();
-            pipeWriter.StopAcceptingWrites();
+            pipeWriter.StopAcceptingWritesAsync();
             var ex = Assert.Throws<ObjectDisposedException>(() => { pipeWriter.Advance(1); });
             Assert.Contains(CoreStrings.WritingToResponseBodyAfterResponseCompleted, ex.Message);
         }
@@ -33,7 +33,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Tests
         {
             var pipeWriter = CreateHttpResponsePipeWriter();
             pipeWriter.StartAcceptingWrites();
-            pipeWriter.StopAcceptingWrites();
+            pipeWriter.StopAcceptingWritesAsync();
             var ex = Assert.Throws<ObjectDisposedException>(() => { pipeWriter.GetMemory(); });
             Assert.Contains(CoreStrings.WritingToResponseBodyAfterResponseCompleted, ex.Message);
         }
@@ -43,7 +43,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Tests
         {
             var pipeWriter = CreateHttpResponsePipeWriter();
             pipeWriter.StartAcceptingWrites();
-            pipeWriter.StopAcceptingWrites();
+            pipeWriter.StopAcceptingWritesAsync();
             var ex = Assert.Throws<ObjectDisposedException>(() => { pipeWriter.GetSpan(); });
             Assert.Contains(CoreStrings.WritingToResponseBodyAfterResponseCompleted, ex.Message);
         }
@@ -53,7 +53,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Tests
         {
             var pipeWriter = CreateHttpResponsePipeWriter();
             pipeWriter.StartAcceptingWrites();
-            pipeWriter.StopAcceptingWrites();
+            pipeWriter.StopAcceptingWritesAsync();
             var ex = Assert.Throws<ObjectDisposedException>(() => { pipeWriter.Complete(); });
             Assert.Contains(CoreStrings.WritingToResponseBodyAfterResponseCompleted, ex.Message);
         }
@@ -63,7 +63,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Tests
         {
             var pipeWriter = CreateHttpResponsePipeWriter();
             pipeWriter.StartAcceptingWrites();
-            pipeWriter.StopAcceptingWrites();
+            pipeWriter.StopAcceptingWritesAsync();
             var ex = Assert.Throws<ObjectDisposedException>(() => { pipeWriter.FlushAsync(); });
             Assert.Contains(CoreStrings.WritingToResponseBodyAfterResponseCompleted, ex.Message);
         }
@@ -73,7 +73,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Tests
         {
             var pipeWriter = CreateHttpResponsePipeWriter();
             pipeWriter.StartAcceptingWrites();
-            pipeWriter.StopAcceptingWrites();
+            pipeWriter.StopAcceptingWritesAsync();
             var ex = Assert.Throws<ObjectDisposedException>(() => { pipeWriter.WriteAsync(new Memory<byte>()); });
             Assert.Contains(CoreStrings.WritingToResponseBodyAfterResponseCompleted, ex.Message);
         }

src/Servers/Kestrel/Core/test/HttpResponseStreamTests.cs

@@ -99,7 +99,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Tests
             var pipeWriter = new HttpResponsePipeWriter(Mock.Of<IHttpResponseControl>());
             var stream = new HttpResponseStream(Mock.Of<IHttpBodyControlFeature>(), pipeWriter);
             pipeWriter.StartAcceptingWrites();
-            pipeWriter.StopAcceptingWrites();
+            pipeWriter.StopAcceptingWritesAsync();
             var ex = Assert.Throws<ObjectDisposedException>(() => { stream.WriteAsync(new byte[1], 0, 1); });
             Assert.Contains(CoreStrings.WritingToResponseBodyAfterResponseCompleted, ex.Message);
         }

src/Servers/Kestrel/test/InMemory.FunctionalTests/Http2/Http2StreamTests.cs

@@ -2128,9 +2128,29 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Tests
         }
 
         [Fact]
-        public async Task RST_STREAM_Received_AbortsStream_FlushedHeadersNotSent()
+        public async Task RST_STREAM_Received_AbortsStream_StreamFlushedDataNotSent()
         {
-            await InitializeConnectionAsync(_waitForAbortFlushingApplication);
+            await InitializeConnectionAsync(async context =>
+            {
+                var streamIdFeature = context.Features.Get<IHttp2StreamIdFeature>();
+                var sem = new SemaphoreSlim(0);
+
+                context.RequestAborted.Register(() =>
+                {
+                    lock (_abortedStreamIdsLock)
+                    {
+                        _abortedStreamIds.Add(streamIdFeature.StreamId);
+                    }
+
+                    sem.Release();
+                });
+
+                await sem.WaitAsync().DefaultTimeout();
+
+                await context.Response.Body.WriteAsync(new byte[10], 0, 10);
+
+                _runningStreams[streamIdFeature.StreamId].TrySetResult(null);
+            });
 
             await StartStreamAsync(1, _browserRequestHeaders, endStream: true);
             await SendRstStreamAsync(1);
@@ -2141,9 +2161,31 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Tests
         }
 
         [Fact]
-        public async Task RST_STREAM_Received_AbortsStream_FlushedDataNotSent()
+        public async Task RST_STREAM_Received_AbortsStream_PipeWriterFlushedDataNotSent()
         {
-            await InitializeConnectionAsync(_waitForAbortWithDataApplication);
+            await InitializeConnectionAsync(async context =>
+            {
+                var streamIdFeature = context.Features.Get<IHttp2StreamIdFeature>();
+                var sem = new SemaphoreSlim(0);
+
+                context.RequestAborted.Register(() =>
+                {
+                    lock (_abortedStreamIdsLock)
+                    {
+                        _abortedStreamIds.Add(streamIdFeature.StreamId);
+                    }
+
+                    sem.Release();
+                });
+
+                await sem.WaitAsync().DefaultTimeout();
+
+                context.Response.BodyWriter.GetMemory();
+                context.Response.BodyWriter.Advance(10);
+                await context.Response.BodyWriter.FlushAsync();
+
+                _runningStreams[streamIdFeature.StreamId].TrySetResult(null);
+            });
 
             await StartStreamAsync(1, _browserRequestHeaders, endStream: true);
             await SendRstStreamAsync(1);
@@ -3263,8 +3305,9 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Tests
         }
 
         [Fact]
-        public async Task ResponseBodyPipeCompleteWithoutExceptionWritesDoNotThrow()
+        public async Task ResponseBodyPipeCompleteWithoutExceptionWritesDoesThrow()
         {
+            InvalidOperationException writeEx = null;
             var headers = new[]
             {
                 new KeyValuePair<string, string>(HeaderNames.Method, "GET"),
@@ -3274,29 +3317,25 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Tests
             await InitializeConnectionAsync(async context =>
             {
                 context.Response.BodyWriter.Complete();
-                await context.Response.WriteAsync("");
+                writeEx = await Assert.ThrowsAsync<InvalidOperationException>(() => context.Response.WriteAsync(""));
             });
 
             await StartStreamAsync(1, headers, endStream: true);
 
             // Don't receive content length because we called WriteAsync which caused an invalid response
             var headersFrame = await ExpectAsync(Http2FrameType.HEADERS,
-                withLength: 37,
+                withLength: 55,
-                withFlags: (byte)Http2HeadersFrameFlags.END_HEADERS,
+                withFlags: (byte)Http2HeadersFrameFlags.END_HEADERS | (byte)Http2HeadersFrameFlags.END_STREAM,
-                withStreamId: 1);
-
-            await ExpectAsync(Http2FrameType.DATA,
-                withLength: 0,
-                withFlags: (byte)Http2DataFrameFlags.END_STREAM,
                 withStreamId: 1);
 
             await StopConnectionAsync(expectedLastStreamId: 1, ignoreNonGoAwayFrames: false);
 
             _hpackDecoder.Decode(headersFrame.PayloadSequence, endHeaders: false, handler: this);
 
-            Assert.Equal(2, _decodedHeaders.Count);
+            Assert.Equal(3, _decodedHeaders.Count);
             Assert.Contains("date", _decodedHeaders.Keys, StringComparer.OrdinalIgnoreCase);
             Assert.Equal("200", _decodedHeaders[HeaderNames.Status]);
+            Assert.NotNull(writeEx);
         }
 
         [Fact]
@@ -3941,6 +3980,77 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Tests
             Assert.Equal("Hello World", Encoding.UTF8.GetString(bodyFrame.Payload.Span));
         }
 
+        [Fact]
+        public async Task PipeWriterComplete_AfterBodyStarted_WithTrailers_TruncatedContentLength_ThrowsAndReset()
+        {
+            var startingTcs = new TaskCompletionSource<int>(TaskCreationOptions.RunContinuationsAsynchronously);
+            var appTcs = new TaskCompletionSource<int>(TaskCreationOptions.RunContinuationsAsynchronously);
+            var clientTcs = new TaskCompletionSource<int>(TaskCreationOptions.RunContinuationsAsynchronously);
+            var headers = new[]
+            {
+                new KeyValuePair<string, string>(HeaderNames.Method, "GET"),
+                new KeyValuePair<string, string>(HeaderNames.Path, "/"),
+                new KeyValuePair<string, string>(HeaderNames.Scheme, "http"),
+            };
+
+            await InitializeConnectionAsync(async context =>
+            {
+                try
+                {
+                    context.Response.OnStarting(() => { startingTcs.SetResult(0); return Task.CompletedTask; });
+
+                    context.Response.ContentLength = 25;
+                    await context.Response.WriteAsync("Hello World");
+                    Assert.True(startingTcs.Task.IsCompletedSuccessfully); // OnStarting got called.
+                    Assert.True(context.Response.Headers.IsReadOnly);
+
+                    context.Response.AppendTrailer("CustomName", "Custom Value");
+
+                    var ex = Assert.Throws<InvalidOperationException>(() => context.Response.BodyWriter.Complete());
+                    Assert.Equal(CoreStrings.FormatTooFewBytesWritten(11, 25), ex.Message);
+
+                    Assert.False(context.Features.Get<IHttpResponseTrailersFeature>().Trailers.IsReadOnly);
+
+                    // Make sure the client gets our results from CompleteAsync instead of from the request delegate exiting.
+                    await clientTcs.Task.DefaultTimeout();
+                    appTcs.SetResult(0);
+                }
+                catch (Exception ex)
+                {
+                    appTcs.SetException(ex);
+                }
+            });
+
+            await StartStreamAsync(1, headers, endStream: true);
+
+            var headersFrame = await ExpectAsync(Http2FrameType.HEADERS,
+                withLength: 56,
+                withFlags: (byte)(Http2HeadersFrameFlags.END_HEADERS),
+                withStreamId: 1);
+            var bodyFrame = await ExpectAsync(Http2FrameType.DATA,
+                withLength: 11,
+                withFlags: (byte)(Http2HeadersFrameFlags.NONE),
+                withStreamId: 1);
+
+            clientTcs.SetResult(0);
+
+            await WaitForStreamErrorAsync(1, Http2ErrorCode.INTERNAL_ERROR,
+                expectedErrorMessage: CoreStrings.FormatTooFewBytesWritten(11, 25));
+
+            await appTcs.Task;
+
+            await StopConnectionAsync(expectedLastStreamId: 1, ignoreNonGoAwayFrames: false);
+
+            _hpackDecoder.Decode(headersFrame.PayloadSequence, endHeaders: false, handler: this);
+
+            Assert.Equal(3, _decodedHeaders.Count);
+            Assert.Contains("date", _decodedHeaders.Keys, StringComparer.OrdinalIgnoreCase);
+            Assert.Equal("200", _decodedHeaders[HeaderNames.Status]);
+            Assert.Equal("25", _decodedHeaders[HeaderNames.ContentLength]);
+
+            Assert.Equal("Hello World", Encoding.UTF8.GetString(bodyFrame.Payload.Span));
+        }
+
         [Fact]
         public async Task AbortAfterCompleteAsync_GETWithResponseBodyAndTrailers_ResetsAfterResponse()
         {

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

@@ -145,7 +145,6 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Tests
         protected readonly RequestDelegate _largeHeadersApplication;
         protected readonly RequestDelegate _waitForAbortApplication;
         protected readonly RequestDelegate _waitForAbortFlushingApplication;
-        protected readonly RequestDelegate _waitForAbortWithDataApplication;
         protected readonly RequestDelegate _readRateApplication;
         protected readonly RequestDelegate _echoMethod;
         protected readonly RequestDelegate _echoHost;
@@ -322,28 +321,6 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Tests
                 _runningStreams[streamIdFeature.StreamId].TrySetResult(null);
             };
 
-            _waitForAbortWithDataApplication = async context =>
-            {
-                var streamIdFeature = context.Features.Get<IHttp2StreamIdFeature>();
-                var sem = new SemaphoreSlim(0);
-
-                context.RequestAborted.Register(() =>
-                {
-                    lock (_abortedStreamIdsLock)
-                    {
-                        _abortedStreamIds.Add(streamIdFeature.StreamId);
-                    }
-
-                    sem.Release();
-                });
-
-                await sem.WaitAsync().DefaultTimeout();
-
-                await context.Response.Body.WriteAsync(new byte[10], 0, 10);
-
-                _runningStreams[streamIdFeature.StreamId].TrySetResult(null);
-            };
-
             _readRateApplication = async context =>
             {
                 var expectedBytes = int.Parse(context.Request.Path.Value.Substring(1));

src/Servers/Kestrel/test/InMemory.FunctionalTests/ResponseTests.cs

@@ -874,6 +874,66 @@ namespace Microsoft.AspNetCore.Server.Kestrel.InMemory.FunctionalTests
                         ex.Message.Equals($"Response Content-Length mismatch: too few bytes written (12 of 13).", StringComparison.Ordinal))));
         }
 
+        [Fact]
+        public async Task WhenAppWritesLessThanContentLengthCompleteThrowsAndErrorLogged()
+        {
+            InvalidOperationException completeEx = null;
+
+            var logTcs = new TaskCompletionSource<object>(TaskCreationOptions.RunContinuationsAsynchronously);
+            var mockTrace = new Mock<IKestrelTrace>();
+            mockTrace
+                .Setup(trace => trace.ApplicationError(It.IsAny<string>(), It.IsAny<string>(), It.IsAny<InvalidOperationException>()))
+                .Callback<string, string, Exception>((connectionId, requestId, ex) =>
+                {
+                    logTcs.SetResult(null);
+                });
+
+            await using (var server = new TestServer(async httpContext =>
+            {
+                httpContext.Response.ContentLength = 13;
+                await httpContext.Response.WriteAsync("hello, world");
+
+                completeEx = Assert.Throws<InvalidOperationException>(() => httpContext.Response.BodyWriter.Complete());
+
+            }, new TestServiceContext(LoggerFactory, mockTrace.Object)))
+            {
+                using (var connection = server.CreateConnection())
+                {
+                    await connection.Send(
+                        "GET / HTTP/1.1",
+                        "Host:",
+                        "",
+                        "");
+
+                    // Don't use ReceiveEnd here, otherwise the FIN might
+                    // abort the request before the server checks the
+                    // response content length, in which case the check
+                    // will be skipped.
+                    await connection.Receive(
+                        $"HTTP/1.1 200 OK",
+                        $"Date: {server.Context.DateHeaderValue}",
+                        "Content-Length: 13",
+                        "",
+                        "hello, world");
+
+                    // Wait for error message to be logged.
+                    await logTcs.Task.DefaultTimeout();
+
+                    // The server should close the connection in this situation.
+                    await connection.WaitForConnectionClose();
+                }
+            }
+
+            mockTrace.Verify(trace =>
+                trace.ApplicationError(
+                    It.IsAny<string>(),
+                    It.IsAny<string>(),
+                    It.Is<InvalidOperationException>(ex =>
+                        ex.Message.Equals($"Response Content-Length mismatch: too few bytes written (12 of 13).", StringComparison.Ordinal))));
+
+            Assert.NotNull(completeEx);
+        }
+
         [Fact]
         public async Task WhenAppWritesLessThanContentLengthButRequestIsAbortedErrorNotLogged()
         {
@@ -3426,12 +3486,14 @@ namespace Microsoft.AspNetCore.Server.Kestrel.InMemory.FunctionalTests
         }
 
         [Fact]
-        public async Task ResponseBodyWriterCompleteWithoutExceptionWritesDoNotThrow()
+        public async Task ResponseBodyWriterCompleteWithoutExceptionWritesDoesThrow()
         {
+            InvalidOperationException writeEx = null;
+
             await using (var server = new TestServer(async httpContext =>
             {
                 httpContext.Response.BodyWriter.Complete();
-                await httpContext.Response.WriteAsync("test");
+                writeEx = await Assert.ThrowsAsync<InvalidOperationException>(() => httpContext.Response.WriteAsync("test"));
             }, new TestServiceContext(LoggerFactory)))
             {
                 using (var connection = server.CreateConnection())
@@ -3441,16 +3503,17 @@ namespace Microsoft.AspNetCore.Server.Kestrel.InMemory.FunctionalTests
                         "Host:",
                         "",
                         "");
+
                     await connection.Receive(
                         "HTTP/1.1 200 OK",
                         $"Date: {server.Context.DateHeaderValue}",
-                        "Transfer-Encoding: chunked",
+                        "Content-Length: 0",
-                        "",
-                        "0",
                         "",
                         "");
                 }
             }
+
+            Assert.NotNull(writeEx);
         }
 
         [Fact]
@@ -3819,19 +3882,17 @@ namespace Microsoft.AspNetCore.Server.Kestrel.InMemory.FunctionalTests
         }
 
         [Fact]
-        public async Task ResponseCompleteGetMemoryAdvanceInLoopDoesNotThrow()
+        public async Task ResponseCompleteGetMemoryDoesThrow()
         {
-            await using (var server = new TestServer(async httpContext =>
+            InvalidOperationException writeEx = null;
+
+            await using (var server = new TestServer(httpContext =>
             {
-
                 httpContext.Response.BodyWriter.Complete();
-                for (var i = 0; i < 5; i++)
-                {
-                    var memory = httpContext.Response.BodyWriter.GetMemory(); // Shouldn't throw
-                    httpContext.Response.BodyWriter.Advance(memory.Length);
-                }
 
-                await Task.CompletedTask;
+                writeEx = Assert.Throws<InvalidOperationException>(() => httpContext.Response.BodyWriter.GetMemory());
+  
+                return Task.CompletedTask;
             }, new TestServiceContext(LoggerFactory)))
             {
                 using (var connection = server.CreateConnection())
@@ -3844,13 +3905,13 @@ namespace Microsoft.AspNetCore.Server.Kestrel.InMemory.FunctionalTests
                     await connection.Receive(
                         "HTTP/1.1 200 OK",
                         $"Date: {server.Context.DateHeaderValue}",
-                        "Transfer-Encoding: chunked",
+                        "Content-Length: 0",
-                        "",
-                        "0",
                         "",
                         "");
                 }
             }
+
+            Assert.NotNull(writeEx);
         }
 
         [Fact]