[2.1.3] Consistently handle connection aborts (#2619)

* Decouple connection objects from the server (#2535)

- Making progress towards being able to use the connection objects on the client side.

* Wait for input writer to complete before calling OnConnectionClosed (#2566)

* Wait for the ConnectionClosed token to stop tracking connections (#2574)

- The prior strategy of waiting for the pipe completed callbacks doesn't work
  because blocks are returned to the memory pool after the callbacks are fired.

* Consistently handle connection resets (#2547)

* Provide better connection abort exceptions and logs

* void IConnectionDispatcher.OnConnection

benchmarks/Kestrel.Performance/InMemoryTransportBenchmark.cs

@@ -3,6 +3,7 @@
 
 using BenchmarkDotNet.Attributes;
 using Microsoft.AspNetCore.Builder;
+using Microsoft.AspNetCore.Connections;
 using Microsoft.AspNetCore.Hosting;
 using Microsoft.AspNetCore.Http;
 using Microsoft.AspNetCore.Server.Kestrel.Transport.Abstractions.Internal;

benchmarks/Kestrel.Performance/Mocks/MockTrace.cs

@@ -44,6 +44,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Performance
         public void RequestBodyDrainTimedOut(string connectionId, string traceIdentifier) { }
         public void RequestBodyMininumDataRateNotSatisfied(string connectionId, string traceIdentifier, double rate) { }
         public void ResponseMininumDataRateNotSatisfied(string connectionId, string traceIdentifier) { }
+        public void ApplicationAbortedConnection(string connectionId, string traceIdentifier) { }
         public void Http2ConnectionError(string connectionId, Http2ConnectionErrorException ex) { }
         public void Http2StreamError(string connectionId, Http2StreamErrorException ex) { }
         public void HPackDecodingError(string connectionId, int streamId, HPackDecodingException ex) { }

src/Connections.Abstractions/ConnectionContext.cs

@@ -1,5 +1,9 @@
-using System.Collections.Generic;
+// 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.Collections.Generic;
 using System.IO.Pipelines;
+using Microsoft.AspNetCore.Connections.Features;
 using Microsoft.AspNetCore.Http.Features;
 
 namespace Microsoft.AspNetCore.Connections
@@ -13,5 +17,15 @@ namespace Microsoft.AspNetCore.Connections
         public abstract IDictionary<object, object> Items { get; set; }
 
         public abstract IDuplexPipe Transport { get; set; }
+
+        public virtual void Abort(ConnectionAbortedException abortReason)
+        {
+            // We expect this to be overridden, but this helps maintain back compat
+            // with implementations of ConnectionContext that predate the addition of
+            // ConnectioContext.Abort()
+            Features.Get<IConnectionLifetimeFeature>()?.Abort();
+        }
+
+        public virtual void Abort() => Abort(new ConnectionAbortedException("The connection was aborted by the application."));
     }
 }

src/Connections.Abstractions/DefaultConnectionContext.cs

@@ -65,7 +65,7 @@ namespace Microsoft.AspNetCore.Connections
 
         public CancellationToken ConnectionClosed { get; set; }
 
-        public virtual void Abort()
+        public override void Abort(ConnectionAbortedException abortReason)
         {
             ThreadPool.QueueUserWorkItem(cts => ((CancellationTokenSource)cts).Cancel(), _connectionClosedTokenSource);
         }

src/Kestrel.Core/Adapter/Internal/AdaptedPipeline.cs

@@ -6,7 +6,9 @@ using System.IO;
 using System.IO.Pipelines;
 using System.Threading.Tasks;
 using Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http;
+using Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Infrastructure;
 using Microsoft.AspNetCore.Server.Kestrel.Transport.Abstractions.Internal;
+using Microsoft.Extensions.Logging;
 
 namespace Microsoft.AspNetCore.Server.Kestrel.Core.Adapter.Internal
 {
@@ -15,23 +17,24 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Adapter.Internal
         private static readonly int MinAllocBufferSize = KestrelMemoryPool.MinimumSegmentSize / 2;
 
         private readonly IDuplexPipe _transport;
-        private readonly IDuplexPipe _application;
 
         public AdaptedPipeline(IDuplexPipe transport,
-                               IDuplexPipe application,
                                Pipe inputPipe,
-                               Pipe outputPipe)
+                               Pipe outputPipe,
+                               IKestrelTrace log)
         {
             _transport = transport;
-            _application = application;
             Input = inputPipe;
             Output = outputPipe;
+            Log = log;
         }
 
         public Pipe Input { get; }
 
         public Pipe Output { get; }
 
+        public IKestrelTrace Log { get; }
+
         PipeReader IDuplexPipe.Input => Input.Reader;
 
         PipeWriter IDuplexPipe.Output => Output.Writer;
@@ -47,8 +50,6 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Adapter.Internal
 
         private async Task WriteOutputAsync(Stream stream)
         {
-            Exception error = null;
-
             try
             {
                 if (stream == null)
@@ -63,13 +64,6 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Adapter.Internal
 
                     try
                     {
-                        if (result.IsCanceled)
-                        {
-                            // Forward the cancellation to the transport pipe
-                            _application.Input.CancelPendingRead();
-                            break;
-                        }
-
                         if (buffer.IsEmpty)
                         {
                             if (result.IsCompleted)
@@ -108,7 +102,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Adapter.Internal
             }
             catch (Exception ex)
             {
-                error = ex;
+                Log.LogError(0, ex, $"{nameof(AdaptedPipeline)}.{nameof(WriteOutputAsync)}");
             }
             finally
             {

src/Kestrel.Core/CoreStrings.resx

@@ -506,4 +506,16 @@ For more information on configuring HTTPS see https://go.microsoft.com/fwlink/?l
   <data name="BadRequest_RequestBodyTimeout" xml:space="preserve">
     <value>Reading the request body timed out due to data arriving too slowly. See MinRequestBodyDataRate.</value>
   </data>
+  <data name="ConnectionAbortedByApplication" xml:space="preserve">
+    <value>The connection was aborted by the application.</value>
+  </data>
+  <data name="ConnectionAbortedDuringServerShutdown" xml:space="preserve">
+    <value>The connection was aborted because the server is shutting down and request processing didn't complete within the time specified by HostOptions.ShutdownTimeout.</value>
+  </data>
+  <data name="ConnectionTimedBecauseResponseMininumDataRateNotSatisfied" xml:space="preserve">
+    <value>The connection was timed out by the server because the response was not read by the client at the specified minimum data rate.</value>
+  </data>
+  <data name="ConnectionTimedOutByServer" xml:space="preserve">
+    <value>The connection was timed out by the server.</value>
+  </data>
 </root>

src/Kestrel.Core/Internal/ConnectionDispatcher.cs

@@ -53,8 +53,6 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal
         {
             using (BeginConnectionScope(connectionContext))
             {
-                Log.ConnectionStart(connectionContext.ConnectionId);
-
                 try
                 {
                     await _connectionDelegate(connectionContext);
@@ -63,8 +61,6 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal
                 {
                     Log.LogCritical(0, ex, $"{nameof(ConnectionDispatcher)}.{nameof(Execute)}() {connectionContext.ConnectionId}");
                 }
-
-                Log.ConnectionStop(connectionContext.ConnectionId);
             }
         }
 

src/Kestrel.Core/Internal/Http/Http1Connection.cs

@@ -45,12 +45,11 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http
             _requestHeadersTimeoutTicks = ServerOptions.Limits.RequestHeadersTimeout.Ticks;
 
             Output = new Http1OutputProducer(
-                _context.Application.Input,
                 _context.Transport.Output,
                 _context.ConnectionId,
+                _context.ConnectionContext,
                 _context.ServiceContext.Log,
                 _context.TimeoutControl,
-                _context.ConnectionFeatures.Get<IConnectionLifetimeFeature>(),
                 _context.ConnectionFeatures.Get<IBytesWrittenFeature>());
         }
 

src/Kestrel.Core/Internal/Http/Http1ConnectionContext.cs

@@ -4,6 +4,7 @@
 using System.Buffers;
 using System.IO.Pipelines;
 using System.Net;
+using Microsoft.AspNetCore.Connections;
 using Microsoft.AspNetCore.Http.Features;
 using Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Infrastructure;
 
@@ -13,6 +14,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http
     {
         public string ConnectionId { get; set; }
         public ServiceContext ServiceContext { get; set; }
+        public ConnectionContext ConnectionContext { get; set; }
         public IFeatureCollection ConnectionFeatures { get; set; }
         public MemoryPool<byte> MemoryPool { get; set; }
         public IPEndPoint RemoteEndPoint { get; set; }

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

@@ -8,6 +8,7 @@ using System.Runtime.CompilerServices;
 using System.Text;
 using System.Threading;
 using System.Threading.Tasks;
+using Microsoft.AspNetCore.Connections;
 using Microsoft.AspNetCore.Connections.Features;
 using Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Infrastructure;
 using Microsoft.AspNetCore.Server.Kestrel.Transport.Abstractions.Internal;
@@ -22,9 +23,9 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http
         private static readonly ReadOnlyMemory<byte> _endChunkedResponseBytes = new ReadOnlyMemory<byte>(Encoding.ASCII.GetBytes("0\r\n\r\n"));
 
         private readonly string _connectionId;
+        private readonly ConnectionContext _connectionContext;
         private readonly ITimeoutControl _timeoutControl;
         private readonly IKestrelTrace _log;
-        private readonly IConnectionLifetimeFeature _lifetimeFeature;
         private readonly IBytesWrittenFeature _transportBytesWrittenFeature;
 
         // This locks access to to all of the below fields
@@ -36,7 +37,6 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http
         private long _totalBytesCommitted;
 
         private readonly PipeWriter _pipeWriter;
-        private readonly PipeReader _outputPipeReader;
 
         // https://github.com/dotnet/corefxlab/issues/1334
         // Pipelines don't support multiple awaiters on flush
@@ -48,21 +48,19 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http
         private ValueTask<FlushResult> _flushTask;
 
         public Http1OutputProducer(
-            PipeReader outputPipeReader,
             PipeWriter pipeWriter,
             string connectionId,
+            ConnectionContext connectionContext,
             IKestrelTrace log,
             ITimeoutControl timeoutControl,
-            IConnectionLifetimeFeature lifetimeFeature,
             IBytesWrittenFeature transportBytesWrittenFeature)
         {
-            _outputPipeReader = outputPipeReader;
             _pipeWriter = pipeWriter;
             _connectionId = connectionId;
+            _connectionContext = connectionContext;
             _timeoutControl = timeoutControl;
             _log = log;
             _flushCompleted = OnFlushCompleted;
-            _lifetimeFeature = lifetimeFeature;
             _transportBytesWrittenFeature = transportBytesWrittenFeature;
         }
 
@@ -169,7 +167,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http
             }
         }
 
-        public void Abort(Exception error)
+        public void Abort(ConnectionAbortedException error)
         {
             // Abort can be called after Dispose if there's a flush timeout.
             // It's important to still call _lifetimeFeature.Abort() in this case.
@@ -181,17 +179,15 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http
                     return;
                 }
 
+                _aborted = true;
+                _connectionContext.Abort(error);
+
                 if (!_completed)
                 {
                     _log.ConnectionDisconnect(_connectionId);
                     _completed = true;
-
+                    _pipeWriter.Complete();
-                    _outputPipeReader.CancelPendingRead();
-                    _pipeWriter.Complete(error);
                 }
-
-                _aborted = true;
-                _lifetimeFeature.Abort();
             }
         }
 

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

@@ -283,7 +283,8 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http
 
         void IHttpRequestLifetimeFeature.Abort()
         {
-            Abort(new ConnectionAbortedException());
+            Log.ApplicationAbortedConnection(ConnectionId, TraceIdentifier);
+            Abort(new ConnectionAbortedException(CoreStrings.ConnectionAbortedByApplication));
         }
     }
 }

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

@@ -39,11 +39,12 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http
 
         protected Streams _streams;
 
-        protected Stack<KeyValuePair<Func<object, Task>, object>> _onStarting;
+        private Stack<KeyValuePair<Func<object, Task>, object>> _onStarting;
-        protected Stack<KeyValuePair<Func<object, Task>, object>> _onCompleted;
+        private Stack<KeyValuePair<Func<object, Task>, object>> _onCompleted;
 
-        protected volatile int _requestAborted;
+        private int _requestAborted;
-        protected CancellationTokenSource _abortedCts;
+        private volatile int _ioCompleted;
+        private CancellationTokenSource _abortedCts;
         private CancellationToken? _manuallySetRequestAbortToken;
 
         protected RequestProcessingStatus _requestProcessingStatus;
@@ -51,15 +52,15 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http
         protected bool _upgradeAvailable;
         private bool _canHaveBody;
         private bool _autoChunk;
-        protected Exception _applicationException;
+        private Exception _applicationException;
         private BadHttpRequestException _requestRejectedException;
 
         protected HttpVersion _httpVersion;
 
         private string _requestId;
-        protected int _requestHeadersParsed;
+        private int _requestHeadersParsed;
 
-        protected long _responseBytesWritten;
+        private long _responseBytesWritten;
 
         private readonly IHttpProtocolContext _context;
 
@@ -247,7 +248,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http
                 var cts = _abortedCts;
                 return
                     cts != null ? cts.Token :
-                    (_requestAborted == 1) ? new CancellationToken(true) :
+                    (_ioCompleted == 1) ? new CancellationToken(true) :
                     RequestAbortedSource.Token;
             }
             set
@@ -272,7 +273,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http
                 var cts = LazyInitializer.EnsureInitialized(ref _abortedCts, () => new CancellationTokenSource())
                             ?? new CancellationTokenSource();
 
-                if (_requestAborted == 1)
+                if (_ioCompleted == 1)
                 {
                     cts.Cancel();
                 }
@@ -410,32 +411,40 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http
             }
         }
 
-        /// <summary>
+        public void OnInputOrOutputCompleted()
-        /// Immediately kill the connection and poison the request and response streams with an error if there is one.
-        /// </summary>
-        public void Abort(Exception error)
         {
-            if (Interlocked.Exchange(ref _requestAborted, 1) != 0)
+            if (Interlocked.Exchange(ref _ioCompleted, 1) != 0)
             {
                 return;
             }
 
             _keepAlive = false;
 
-            // If Abort() isn't called with an exception, there was a FIN. In this case, even though the connection is
+            Output.Dispose();
-            // still closed immediately, we allow the app to drain the data in the request buffer. If the request data
-            // was truncated, MessageBody will complete the RequestBodyPipe with an error.
-            if (error != null)
-            {
-                _streams?.Abort(error);
-            }
-
-            Output.Abort(error);
 
             // Potentially calling user code. CancelRequestAbortedToken logs any exceptions.
             ServiceContext.Scheduler.Schedule(state => ((HttpProtocol)state).CancelRequestAbortedToken(), this);
         }
 
+        /// <summary>
+        /// Immediately kill the connection and poison the request and response streams with an error if there is one.
+        /// </summary>
+        public void Abort(ConnectionAbortedException abortReason)
+        {
+            if (Interlocked.Exchange(ref _requestAborted, 1) != 0)
+            {
+                return;
+            }
+
+            _streams?.Abort(abortReason);
+
+            // Abort output prior to calling OnIOCompleted() to give the transport the chance to
+            // complete the input with the correct error and message.
+            Output.Abort(abortReason);
+
+            OnInputOrOutputCompleted();
+        }
+
         public void OnHeader(Span<byte> name, Span<byte> value)
         {
             _requestHeadersParsed++;
@@ -543,7 +552,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http
                     // Run the application code for this request
                     await application.ProcessRequestAsync(httpContext);
 
-                    if (_requestAborted == 0)
+                    if (_ioCompleted == 0)
                     {
                         VerifyResponseContentLength();
                     }
@@ -579,8 +588,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http
                 // 4XX responses are written by TryProduceInvalidRequestResponse during connection tear down.
                 if (_requestRejectedException == null)
                 {
-                    // If _requestAbort is set, the connection has already been closed.
+                    if (_ioCompleted == 0)
-                    if (_requestAborted == 0)
                     {
                         // Call ProduceEnd() before consuming the rest of the request body to prevent
                         // delaying clients waiting for the chunk terminator:
@@ -612,7 +620,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http
                 application.DisposeContext(httpContext, _applicationException);
 
                 // Even for non-keep-alive requests, try to consume the entire body to avoid RSTs.
-                if (_requestAborted == 0 && _requestRejectedException == null && !messageBody.IsEmpty)
+                if (_ioCompleted == 0 && _requestRejectedException == null && !messageBody.IsEmpty)
                 {
                     await messageBody.ConsumeAsync();
                 }
@@ -1010,8 +1018,8 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http
 
         protected Task TryProduceInvalidRequestResponse()
         {
-            // If _requestAborted is set, the connection has already been closed.
+            // If _ioCompleted is set, the connection has already been closed.
-            if (_requestRejectedException != null && _requestAborted == 0)
+            if (_requestRejectedException != null && _ioCompleted == 0)
             {
                 return ProduceEnd();
             }

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

@@ -5,12 +5,13 @@ using System;
 using System.IO.Pipelines;
 using System.Threading;
 using System.Threading.Tasks;
+using Microsoft.AspNetCore.Connections;
 
 namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http
 {
     public interface IHttpOutputProducer : IDisposable
     {
-        void Abort(Exception error);
+        void Abort(ConnectionAbortedException abortReason);
         Task WriteAsync<T>(Func<PipeWriter, T, long> callback, T state);
         Task FlushAsync(CancellationToken cancellationToken);
         Task Write100ContinueAsync(CancellationToken cancellationToken);

src/Kestrel.Core/Internal/Http2/Http2Connection.cs

@@ -89,7 +89,13 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http2
 
         public IFeatureCollection ConnectionFeatures => _context.ConnectionFeatures;
 
-        public void Abort(Exception ex)
+        public void OnInputOrOutputCompleted()
+        {
+            _stopping = true;
+            _frameWriter.Abort(ex: null);
+        }
+
+        public void Abort(ConnectionAbortedException ex)
         {
             _stopping = true;
             _frameWriter.Abort(ex);
@@ -202,7 +208,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http2
                 {
                     foreach (var stream in _streams.Values)
                     {
-                        stream.Abort(error);
+                        stream.Http2Abort(error);
                     }
 
                     await _frameWriter.WriteGoAwayAsync(_highestOpenedStreamId, errorCode);
@@ -464,7 +470,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http2
 
             if (_streams.TryGetValue(_incomingFrame.StreamId, out var stream))
             {
-                stream.Abort(error: null);
+                stream.Abort(abortReason: null);
             }
 
             return Task.CompletedTask;

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

@@ -5,6 +5,7 @@ using System;
 using System.IO.Pipelines;
 using System.Threading;
 using System.Threading.Tasks;
+using Microsoft.AspNetCore.Connections;
 using Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http;
 using Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Infrastructure;
 
@@ -25,7 +26,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http2
         {
         }
 
-        public void Abort(Exception error)
+        public void Abort(ConnectionAbortedException error)
         {
             // TODO: RST_STREAM?
         }

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

@@ -142,5 +142,16 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http2
                 RequestBodyPipe.Writer.Complete(ex);
             }
         }
+
+        // TODO: The HTTP/2 tests expect the request and response streams to be aborted with
+        // non-ConnectionAbortedExceptions. The abortReasons can include things like
+        // Http2ConnectionErrorException which don't derive from IOException or
+        // OperationCanceledException. This is probably not a good idea.
+        public void Http2Abort(Exception abortReason)
+        {
+            _streams?.Abort(abortReason);
+
+            OnInputOrOutputCompleted();
+        }
     }
 }

src/Kestrel.Core/Internal/HttpConnection.cs

@@ -10,6 +10,8 @@ using System.IO.Pipelines;
 using System.Net;
 using System.Threading;
 using System.Threading.Tasks;
+using Microsoft.AspNetCore.Connections;
+using Microsoft.AspNetCore.Connections.Features;
 using Microsoft.AspNetCore.Hosting.Server;
 using Microsoft.AspNetCore.Http.Features;
 using Microsoft.AspNetCore.Server.Kestrel.Core.Adapter.Internal;
@@ -105,6 +107,10 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal
         {
             try
             {
+                // TODO: When we start tracking all connection middleware for shutdown, go back
+                // to logging connections tart and stop in ConnectionDispatcher so we get these
+                // logs for all connection middleware.
+                Log.ConnectionStart(ConnectionId);
                 KestrelEventSource.Log.ConnectionStart(this);
 
                 AdaptedPipeline adaptedPipeline = null;
@@ -119,9 +125,9 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal
                 if (_context.ConnectionAdapters.Count > 0)
                 {
                     adaptedPipeline = new AdaptedPipeline(_adaptedTransport,
-                                                          application,
                                                           new Pipe(AdaptedInputPipeOptions),
-                                                          new Pipe(AdaptedOutputPipeOptions));
+                                                          new Pipe(AdaptedOutputPipeOptions),
+                                                          Log);
 
                     _adaptedTransport = adaptedPipeline;
                 }
@@ -196,6 +202,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal
                     _context.ServiceContext.ConnectionManager.UpgradedConnectionCount.ReleaseOne();
                 }
 
+                Log.ConnectionStop(ConnectionId);
                 KestrelEventSource.Log.ConnectionStop(this);
             }
         }
@@ -217,6 +224,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal
                 LocalEndPoint = LocalEndPoint,
                 RemoteEndPoint = RemoteEndPoint,
                 ServiceContext = _context.ServiceContext,
+                ConnectionContext = _context.ConnectionContext,
                 TimeoutControl = this,
                 Transport = transport,
                 Application = application
@@ -238,10 +246,8 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal
             });
         }
 
-        public void OnConnectionClosed(Exception ex)
+        public void OnConnectionClosed()
         {
-            Abort(ex);
-
             _socketClosedTcs.TrySetResult(null);
         }
 
@@ -252,15 +258,13 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal
                 switch (_protocolSelectionState)
                 {
                     case ProtocolSelectionState.Initializing:
-                        CloseUninitializedConnection();
+                        CloseUninitializedConnection(abortReason: null);
-                        _protocolSelectionState = ProtocolSelectionState.Stopped;
+                        _protocolSelectionState = ProtocolSelectionState.Aborted;
                         break;
                     case ProtocolSelectionState.Selected:
                         _requestProcessor.StopProcessingNextRequest();
-                        _protocolSelectionState = ProtocolSelectionState.Stopping;
                         break;
-                    case ProtocolSelectionState.Stopping:
+                    case ProtocolSelectionState.Aborted:
-                    case ProtocolSelectionState.Stopped:
                         break;
                 }
             }
@@ -268,28 +272,47 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal
             return _lifetimeTask;
         }
 
-        public void Abort(Exception ex)
+        public void OnInputOrOutputCompleted()
         {
             lock (_protocolSelectionLock)
             {
                 switch (_protocolSelectionState)
                 {
                     case ProtocolSelectionState.Initializing:
-                        CloseUninitializedConnection();
+                        CloseUninitializedConnection(abortReason: null);
+                        _protocolSelectionState = ProtocolSelectionState.Aborted;
                         break;
                     case ProtocolSelectionState.Selected:
-                    case ProtocolSelectionState.Stopping:
+                        _requestProcessor.OnInputOrOutputCompleted();
-                        _requestProcessor.Abort(ex);
                         break;
-                    case ProtocolSelectionState.Stopped:
+                    case ProtocolSelectionState.Aborted:
                         break;
                 }
 
-                _protocolSelectionState = ProtocolSelectionState.Stopped;
             }
         }
 
-        public Task AbortAsync(Exception ex)
+        public void Abort(ConnectionAbortedException ex)
+        {
+            lock (_protocolSelectionLock)
+            {
+                switch (_protocolSelectionState)
+                {
+                    case ProtocolSelectionState.Initializing:
+                        CloseUninitializedConnection(ex);
+                        break;
+                    case ProtocolSelectionState.Selected:
+                        _requestProcessor.Abort(ex);
+                        break;
+                    case ProtocolSelectionState.Aborted:
+                        break;
+                }
+
+                _protocolSelectionState = ProtocolSelectionState.Aborted;
+            }
+        }
+
+        public Task AbortAsync(ConnectionAbortedException ex)
         {
             Abort(ex);
 
@@ -370,7 +393,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal
 
         public void Tick(DateTimeOffset now)
         {
-            if (_protocolSelectionState == ProtocolSelectionState.Stopped)
+            if (_protocolSelectionState == ProtocolSelectionState.Aborted)
             {
                 // It's safe to check for timeouts on a dead connection,
                 // but try not to in order to avoid extraneous logs.
@@ -416,7 +439,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal
                             break;
                         case TimeoutAction.AbortConnection:
                             // This is actually supported with HTTP/2!
-                            Abort(new TimeoutException());
+                            Abort(new ConnectionAbortedException(CoreStrings.ConnectionTimedOutByServer));
                             break;
                     }
                 }
@@ -479,7 +502,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal
                 {
                     RequestTimedOut = true;
                     Log.ResponseMininumDataRateNotSatisfied(_http1Connection.ConnectionIdFeature, _http1Connection.TraceIdentifier);
-                    Abort(new TimeoutException());
+                    Abort(new ConnectionAbortedException(CoreStrings.ConnectionTimedBecauseResponseMininumDataRateNotSatisfied));
                 }
             }
         }
@@ -619,13 +642,11 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal
             ResetTimeout(timeSpan.Ticks, TimeoutAction.AbortConnection);
         }
 
-        private void CloseUninitializedConnection()
+        private void CloseUninitializedConnection(ConnectionAbortedException abortReason)
         {
             Debug.Assert(_adaptedTransport != null);
 
-            // CancelPendingRead signals the transport directly to close the connection
+            _context.ConnectionContext.Abort(abortReason);
-            // without any potential interference from connection adapters.
-            _context.Application.Input.CancelPendingRead();
 
             _adaptedTransport.Input.Complete();
             _adaptedTransport.Output.Complete();
@@ -635,8 +656,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal
         {
             Initializing,
             Selected,
-            Stopping,
+            Aborted
-            Stopped
         }
     }
 }

src/Kestrel.Core/Internal/HttpConnectionMiddleware.cs

@@ -1,11 +1,14 @@
-using System.Collections.Generic;
+// 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.Collections.Generic;
 using System.Net;
 using System.Threading;
 using System.Threading.Tasks;
-using Microsoft.AspNetCore.Hosting.Server;
-using Microsoft.AspNetCore.Http.Features;
 using Microsoft.AspNetCore.Connections;
 using Microsoft.AspNetCore.Connections.Features;
+using Microsoft.AspNetCore.Hosting.Server;
+using Microsoft.AspNetCore.Http.Features;
 using Microsoft.AspNetCore.Server.Kestrel.Core.Adapter.Internal;
 using Microsoft.AspNetCore.Server.Kestrel.Transport.Abstractions.Internal;
 
@@ -30,7 +33,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal
             _connectionAdapters = adapters;
         }
 
-        public Task OnConnectionAsync(ConnectionContext connectionContext)
+        public async Task OnConnectionAsync(ConnectionContext connectionContext)
         {
             // We need the transport feature so that we can cancel the output reader that the transport is using
             // This is a bit of a hack but it preserves the existing semantics
@@ -54,6 +57,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal
             };
 
             var connectionFeature = connectionContext.Features.Get<IHttpConnectionFeature>();
+            var lifetimeFeature = connectionContext.Features.Get<IConnectionLifetimeFeature>();
 
             if (connectionFeature != null)
             {
@@ -72,19 +76,33 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal
 
             var processingTask = connection.StartRequestProcessing(_application);
 
-            connectionContext.Transport.Input.OnWriterCompleted((error, state) =>
+            connectionContext.Transport.Input.OnWriterCompleted(
-            {
+                (_, state) => ((HttpConnection)state).OnInputOrOutputCompleted(),
-                ((HttpConnection)state).Abort(error);
+                connection);
-            },
-            connection);
 
-            connectionContext.Transport.Output.OnReaderCompleted((error, state) =>
+            connectionContext.Transport.Output.OnReaderCompleted(
-            {
+                (_, state) => ((HttpConnection)state).OnInputOrOutputCompleted(),
-                ((HttpConnection)state).OnConnectionClosed(error);
+                connection);
-            },
-            connection);
 
-            return processingTask;
+            await CancellationTokenAsTask(lifetimeFeature.ConnectionClosed);
+
+            connection.OnConnectionClosed();
+
+            await processingTask;
+        }
+
+        private static Task CancellationTokenAsTask(CancellationToken token)
+        {
+            if (token.IsCancellationRequested)
+            {
+                return Task.CompletedTask;
+            }
+
+            // Transports already dispatch prior to tripping ConnectionClosed
+            // since application code can register to this token.
+            var tcs = new TaskCompletionSource<object>();
+            token.Register(() => tcs.SetResult(null));
+            return tcs.Task;
         }
     }
 }

src/Kestrel.Core/Internal/IRequestProcessor.cs

@@ -3,6 +3,7 @@
 
 using System;
 using System.Threading.Tasks;
+using Microsoft.AspNetCore.Connections;
 using Microsoft.AspNetCore.Hosting.Server;
 
 namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal
@@ -11,6 +12,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal
     {
         Task ProcessRequestsAsync<TContext>(IHttpApplication<TContext> application);
         void StopProcessingNextRequest();
-        void Abort(Exception ex);
+        void OnInputOrOutputCompleted();
+        void Abort(ConnectionAbortedException ex);
     }
 }

src/Kestrel.Core/Internal/Infrastructure/HttpConnectionManagerShutdownExtensions.cs

@@ -26,7 +26,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Infrastructure
         public static async Task<bool> AbortAllConnectionsAsync(this HttpConnectionManager connectionManager)
         {
             var abortTasks = new List<Task>();
-            var canceledException = new ConnectionAbortedException();
+            var canceledException = new ConnectionAbortedException(CoreStrings.ConnectionAbortedDuringServerShutdown);
 
             connectionManager.Walk(connection =>
             {

src/Kestrel.Core/Internal/Infrastructure/IKestrelTrace.cs

@@ -52,6 +52,8 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Infrastructure
 
         void ResponseMininumDataRateNotSatisfied(string connectionId, string traceIdentifier);
 
+        void ApplicationAbortedConnection(string connectionId, string traceIdentifier);
+
         void Http2ConnectionError(string connectionId, Http2ConnectionErrorException ex);
 
         void Http2StreamError(string connectionId, Http2StreamErrorException ex);

src/Kestrel.Core/Internal/Infrastructure/KestrelTrace.cs

@@ -65,12 +65,6 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal
         private static readonly Action<ILogger, string, string, double, Exception> _requestBodyMinimumDataRateNotSatisfied =
             LoggerMessage.Define<string, string, double>(LogLevel.Information, new EventId(27, nameof(RequestBodyMininumDataRateNotSatisfied)), @"Connection id ""{ConnectionId}"", Request id ""{TraceIdentifier}"": the request timed out because it was not sent by the client at a minimum of {Rate} bytes/second.");
 
-        private static readonly Action<ILogger, string, string, Exception> _requestBodyNotEntirelyRead =
-            LoggerMessage.Define<string, string>(LogLevel.Information, new EventId(32, nameof(RequestBodyNotEntirelyRead)), @"Connection id ""{ConnectionId}"", Request id ""{TraceIdentifier}"": the application completed without reading the entire request body.");
-
-        private static readonly Action<ILogger, string, string, Exception> _requestBodyDrainTimedOut =
-            LoggerMessage.Define<string, string>(LogLevel.Information, new EventId(33, nameof(RequestBodyDrainTimedOut)), @"Connection id ""{ConnectionId}"", Request id ""{TraceIdentifier}"": automatic draining of the request body timed out after taking over 5 seconds.");
-
         private static readonly Action<ILogger, string, string, Exception> _responseMinimumDataRateNotSatisfied =
             LoggerMessage.Define<string, string>(LogLevel.Information, new EventId(28, nameof(ResponseMininumDataRateNotSatisfied)), @"Connection id ""{ConnectionId}"", Request id ""{TraceIdentifier}"": the connection was closed because the response was not read by the client at the specified minimum data rate.");
 
@@ -83,6 +77,15 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal
         private static readonly Action<ILogger, string, int, Exception> _hpackDecodingError =
             LoggerMessage.Define<string, int>(LogLevel.Information, new EventId(31, nameof(HPackDecodingError)), @"Connection id ""{ConnectionId}"": HPACK decoding error while decoding headers for stream ID {StreamId}.");
 
+        private static readonly Action<ILogger, string, string, Exception> _requestBodyNotEntirelyRead =
+            LoggerMessage.Define<string, string>(LogLevel.Information, new EventId(32, nameof(RequestBodyNotEntirelyRead)), @"Connection id ""{ConnectionId}"", Request id ""{TraceIdentifier}"": the application completed without reading the entire request body.");
+
+        private static readonly Action<ILogger, string, string, Exception> _requestBodyDrainTimedOut =
+            LoggerMessage.Define<string, string>(LogLevel.Information, new EventId(33, nameof(RequestBodyDrainTimedOut)), @"Connection id ""{ConnectionId}"", Request id ""{TraceIdentifier}"": automatic draining of the request body timed out after taking over 5 seconds.");
+
+        private static readonly Action<ILogger, string, string, Exception> _applicationAbortedConnection =
+            LoggerMessage.Define<string, string>(LogLevel.Information, new EventId(34, nameof(RequestBodyDrainTimedOut)), @"Connection id ""{ConnectionId}"", Request id ""{TraceIdentifier}"": the application aborted the connection.");
+
         protected readonly ILogger _logger;
 
         public KestrelTrace(ILogger logger)
@@ -195,6 +198,11 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal
             _responseMinimumDataRateNotSatisfied(_logger, connectionId, traceIdentifier, null);
         }
 
+        public virtual void ApplicationAbortedConnection(string connectionId, string traceIdentifier)
+        {
+            _applicationAbortedConnection(_logger, connectionId, traceIdentifier, null);
+        }
+
         public virtual void Http2ConnectionError(string connectionId, Http2ConnectionErrorException ex)
         {
             _http2ConnectionError(_logger, connectionId, ex);

src/Kestrel.Core/Properties/CoreStrings.Designer.cs

@@ -1820,6 +1820,62 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core
         internal static string FormatBadRequest_RequestBodyTimeout()
             => GetString("BadRequest_RequestBodyTimeout");
 
+        /// <summary>
+        /// The connection was aborted by the application.
+        /// </summary>
+        internal static string ConnectionAbortedByApplication
+        {
+            get => GetString("ConnectionAbortedByApplication");
+        }
+
+        /// <summary>
+        /// The connection was aborted by the application.
+        /// </summary>
+        internal static string FormatConnectionAbortedByApplication()
+            => GetString("ConnectionAbortedByApplication");
+
+        /// <summary>
+        /// The connection was aborted because the server is shutting down and request processing didn't complete within the time specified by HostOptions.ShutdownTimeout.
+        /// </summary>
+        internal static string ConnectionAbortedDuringServerShutdown
+        {
+            get => GetString("ConnectionAbortedDuringServerShutdown");
+        }
+
+        /// <summary>
+        /// The connection was aborted because the server is shutting down and request processing didn't complete within the time specified by HostOptions.ShutdownTimeout.
+        /// </summary>
+        internal static string FormatConnectionAbortedDuringServerShutdown()
+            => GetString("ConnectionAbortedDuringServerShutdown");
+
+        /// <summary>
+        /// The connection was timed out by the server because the response was not read by the client at the specified minimum data rate.
+        /// </summary>
+        internal static string ConnectionTimedBecauseResponseMininumDataRateNotSatisfied
+        {
+            get => GetString("ConnectionTimedBecauseResponseMininumDataRateNotSatisfied");
+        }
+
+        /// <summary>
+        /// The connection was timed out by the server because the response was not read by the client at the specified minimum data rate.
+        /// </summary>
+        internal static string FormatConnectionTimedBecauseResponseMininumDataRateNotSatisfied()
+            => GetString("ConnectionTimedBecauseResponseMininumDataRateNotSatisfied");
+
+        /// <summary>
+        /// The connection was timed out by the server.
+        /// </summary>
+        internal static string ConnectionTimedOutByServer
+        {
+            get => GetString("ConnectionTimedOutByServer");
+        }
+
+        /// <summary>
+        /// The connection was timed out by the server.
+        /// </summary>
+        internal static string FormatConnectionTimedOutByServer()
+            => GetString("ConnectionTimedOutByServer");
+
         private static string GetString(string name, params string[] formatterNames)
         {
             var value = _resourceManager.GetString(name);

src/Kestrel.Transport.Abstractions/Internal/IConnectionDispatcher.cs

@@ -1,8 +1,6 @@
 // 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 Microsoft.AspNetCore.Http.Features;
-
 namespace Microsoft.AspNetCore.Server.Kestrel.Transport.Abstractions.Internal
 {
     public interface IConnectionDispatcher

src/Kestrel.Transport.Abstractions/Internal/TransportConnection.cs

@@ -60,8 +60,15 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Transport.Abstractions.Internal
 
         public CancellationToken ConnectionClosed { get; set; }
 
-        public virtual void Abort()
+        // DO NOT remove this override to ConnectionContext.Abort. Doing so would cause
+        // any TransportConnection that does not override Abort or calls base.Abort
+        // to stack overflow when IConnectionLifetimeFeature.Abort() is called.
+        // That said, all derived types should override this method should override
+        // this implementation of Abort because canceling pending output reads is not
+        // sufficient to abort the connection if there is backpressure.
+        public override void Abort(ConnectionAbortedException abortReason)
         {
+            Output.CancelPendingRead();
         }
     }
 }

src/Kestrel.Transport.Libuv/Internal/LibuvConnection.cs

@@ -5,6 +5,7 @@ using System;
 using System.Buffers;
 using System.IO;
 using System.IO.Pipelines;
+using System.Net;
 using System.Threading;
 using System.Threading.Tasks;
 using Microsoft.AspNetCore.Connections;
@@ -14,7 +15,7 @@ using Microsoft.Extensions.Logging;
 
 namespace Microsoft.AspNetCore.Server.Kestrel.Transport.Libuv.Internal
 {
-    public partial class LibuvConnection : LibuvConnectionContext
+    public partial class LibuvConnection : TransportConnection
     {
         private static readonly int MinAllocBufferSize = KestrelMemoryPool.MinimumSegmentSize / 2;
 
@@ -27,32 +28,31 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Transport.Libuv.Internal
         private readonly UvStreamHandle _socket;
         private readonly CancellationTokenSource _connectionClosedTokenSource = new CancellationTokenSource();
 
+        private volatile ConnectionAbortedException _abortReason;
+
         private MemoryHandle _bufferHandle;
 
-        public LibuvConnection(ListenerContext context, UvStreamHandle socket) : base(context)
+        public LibuvConnection(UvStreamHandle socket, ILibuvTrace log, LibuvThread thread, IPEndPoint remoteEndPoint, IPEndPoint localEndPoint)
         {
             _socket = socket;
 
-            if (_socket is UvTcpHandle tcpHandle)
+            RemoteAddress = remoteEndPoint?.Address;
-            {
+            RemotePort = remoteEndPoint?.Port ?? 0;
-                var remoteEndPoint = tcpHandle.GetPeerIPEndPoint();
-                var localEndPoint = tcpHandle.GetSockIPEndPoint();
 
-                RemoteAddress = remoteEndPoint.Address;
+            LocalAddress = localEndPoint?.Address;
-                RemotePort = remoteEndPoint.Port;
+            LocalPort = localEndPoint?.Port ?? 0;
 
-                LocalAddress = localEndPoint.Address;
+            ConnectionClosed = _connectionClosedTokenSource.Token;
-                LocalPort = localEndPoint.Port;
+            Log = log;
-
+            Thread = thread;
-                ConnectionClosed = _connectionClosedTokenSource.Token;
-            }
         }
 
         public LibuvOutputConsumer OutputConsumer { get; set; }
-
+        private ILibuvTrace Log { get; }
-        private ILibuvTrace Log => ListenerContext.TransportContext.Log;
+        private LibuvThread Thread { get; }
-        private IConnectionDispatcher ConnectionDispatcher => ListenerContext.TransportContext.ConnectionDispatcher;
+        public override MemoryPool<byte> MemoryPool => Thread.MemoryPool;
-        private LibuvThread Thread => ListenerContext.Thread;
+        public override PipeScheduler InputWriterScheduler => Thread;
+        public override PipeScheduler OutputReaderScheduler => Thread;
 
         public override long TotalBytesWritten => OutputConsumer?.TotalBytesWritten ?? 0;
 
@@ -60,13 +60,12 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Transport.Libuv.Internal
         {
             try
             {
-                ConnectionDispatcher.OnConnection(this);
-
                 OutputConsumer = new LibuvOutputConsumer(Output, Thread, _socket, ConnectionId, Log);
 
                 StartReading();
 
-                Exception error = null;
+                Exception inputError = null;
+                Exception outputError = null;
 
                 try
                 {
@@ -77,13 +76,27 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Transport.Libuv.Internal
                 }
                 catch (UvException ex)
                 {
-                    error = new IOException(ex.Message, ex);
+                    // The connection reset/error has already been logged by LibuvOutputConsumer
+                    if (ex.StatusCode == LibuvConstants.ECANCELED)
+                    {
+                        // Connection was aborted.
+                    }
+                    else if (LibuvConstants.IsConnectionReset(ex.StatusCode))
+                    {
+                        // Don't cause writes to throw for connection resets.
+                        inputError = new ConnectionResetException(ex.Message, ex);
+                    }
+                    else
+                    {
+                        inputError = ex;
+                        outputError = ex;
+                    }
                 }
                 finally
                 {
                     // Now, complete the input so that no more reads can happen
-                    Input.Complete(error ?? new ConnectionAbortedException());
+                    Input.Complete(inputError ?? _abortReason ?? new ConnectionAbortedException());
-                    Output.Complete(error);
+                    Output.Complete(outputError);
 
                     // Make sure it isn't possible for a paused read to resume reading after calling uv_close
                     // on the stream handle
@@ -103,8 +116,11 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Transport.Libuv.Internal
             }
         }
 
-        public override void Abort()
+        public override void Abort(ConnectionAbortedException abortReason)
         {
+            _abortReason = abortReason;
+            Output.CancelPendingRead();
+            
             // This cancels any pending I/O.
             Thread.Post(s => s.Dispose(), _socket);
         }
@@ -156,7 +172,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Transport.Libuv.Internal
                 // Given a negative status, it's possible that OnAlloc wasn't called.
                 _socket.ReadStop();
 
-                IOException error = null;
+                Exception error = null;
 
                 if (status == LibuvConstants.EOF)
                 {
@@ -165,18 +181,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Transport.Libuv.Internal
                 else
                 {
                     handle.Libuv.Check(status, out var uvError);
-
+                    error = LogAndWrapReadError(uvError);
-                    // Log connection resets at a lower (Debug) level.
-                    if (LibuvConstants.IsConnectionReset(status))
-                    {
-                        Log.ConnectionReset(ConnectionId);
-                        error = new ConnectionResetException(uvError.Message, uvError);
-                    }
-                    else
-                    {
-                        Log.ConnectionError(ConnectionId, uvError);
-                        error = new IOException(uvError.Message, uvError);
-                    }
                 }
 
                 // Complete after aborting the connection
@@ -209,10 +214,27 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Transport.Libuv.Internal
             {
                 // ReadStart() can throw a UvException in some cases (e.g. socket is no longer connected).
                 // This should be treated the same as OnRead() seeing a negative status.
-                Log.ConnectionReadFin(ConnectionId);
+                Input.Complete(LogAndWrapReadError(ex));
-                var error = new IOException(ex.Message, ex);
+            }
+        }
 
-                Input.Complete(error);
+        private Exception LogAndWrapReadError(UvException uvError)
+        {
+            if (uvError.StatusCode == LibuvConstants.ECANCELED)
+            {
+                // The operation was canceled by the server not the client. No need for additional logs.
+                return new ConnectionAbortedException(uvError.Message, uvError);
+            }
+            else if (LibuvConstants.IsConnectionReset(uvError.StatusCode))
+            {
+                // Log connection resets at a lower (Debug) level.
+                Log.ConnectionReset(ConnectionId);
+                return new ConnectionResetException(uvError.Message, uvError);
+            }
+            else
+            {
+                Log.ConnectionError(ConnectionId, uvError);
+                return new IOException(uvError.Message, uvError);
             }
         }
 
@@ -221,7 +243,6 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Transport.Libuv.Internal
             try
             {
                 _connectionClosedTokenSource.Cancel();
-                _connectionClosedTokenSource.Dispose();
             }
             catch (Exception ex)
             {

src/Kestrel.Transport.Libuv/Internal/LibuvConnectionContext.cs

@@ -1,23 +0,0 @@
-// 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.Buffers;
-using System.IO.Pipelines;
-using Microsoft.AspNetCore.Server.Kestrel.Transport.Abstractions.Internal;
-
-namespace Microsoft.AspNetCore.Server.Kestrel.Transport.Libuv.Internal
-{
-    public class LibuvConnectionContext : TransportConnection
-    {
-        public LibuvConnectionContext(ListenerContext context)
-        {
-            ListenerContext = context;
-        }
-
-        public ListenerContext ListenerContext { get; set; }
-
-        public override MemoryPool<byte> MemoryPool => ListenerContext.Thread.MemoryPool;
-        public override PipeScheduler InputWriterScheduler => ListenerContext.Thread;
-        public override PipeScheduler OutputReaderScheduler => ListenerContext.Thread;
-    }
-}

src/Kestrel.Transport.Libuv/Internal/LibuvConstants.cs

@@ -15,11 +15,14 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Transport.Libuv.Internal
         public static readonly int? EADDRINUSE = GetEADDRINUSE();
         public static readonly int? ENOTSUP = GetENOTSUP();
         public static readonly int? EPIPE = GetEPIPE();
+        public static readonly int? ECANCELED = GetECANCELED();
+        public static readonly int? ENOTCONN = GetENOTCONN();
+        public static readonly int? EINVAL = GetEINVAL();
 
         [MethodImpl(MethodImplOptions.AggressiveInlining)]
         public static bool IsConnectionReset(int errno)
         {
-            return errno == ECONNRESET || errno == EPIPE;
+            return errno == ECONNRESET || errno == EPIPE || errno == ENOTCONN | errno == EINVAL;
         }
 
         private static int? GetECONNRESET()
@@ -41,11 +44,52 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Transport.Libuv.Internal
 
         private static int? GetEPIPE()
         {
-            if (RuntimeInformation.IsOSPlatform(OSPlatform.Linux))
+            if (RuntimeInformation.IsOSPlatform(OSPlatform.Windows))
+            {
+                return -4047;
+            }
+            else if (RuntimeInformation.IsOSPlatform(OSPlatform.Linux))
             {
                 return -32;
             }
+            else if (RuntimeInformation.IsOSPlatform(OSPlatform.OSX))
+            {
+                return -32;
+            }
+            return null;
+        }
 
+        private static int? GetENOTCONN()
+        {
+            if (RuntimeInformation.IsOSPlatform(OSPlatform.Windows))
+            {
+                return -4053;
+            }
+            else if (RuntimeInformation.IsOSPlatform(OSPlatform.Linux))
+            {
+                return -107;
+            }
+            else if (RuntimeInformation.IsOSPlatform(OSPlatform.OSX))
+            {
+                return -57;
+            }
+            return null;
+        }
+
+        private static int? GetEINVAL()
+        {
+            if (RuntimeInformation.IsOSPlatform(OSPlatform.Windows))
+            {
+                return -4071;
+            }
+            else if (RuntimeInformation.IsOSPlatform(OSPlatform.Linux))
+            {
+                return -22;
+            }
+            else if (RuntimeInformation.IsOSPlatform(OSPlatform.OSX))
+            {
+                return -22;
+            }
             return null;
         }
 
@@ -78,5 +122,22 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Transport.Libuv.Internal
             }
             return null;
         }
+
+        private static int? GetECANCELED()
+        {
+            if (RuntimeInformation.IsOSPlatform(OSPlatform.Windows))
+            {
+                return -4081;
+            }
+            else if (RuntimeInformation.IsOSPlatform(OSPlatform.Linux))
+            {
+                return -125;
+            }
+            else if (RuntimeInformation.IsOSPlatform(OSPlatform.OSX))
+            {
+                return -89;
+            }
+            return null;
+        }
     }
 }

src/Kestrel.Transport.Libuv/Internal/LibuvOutputConsumer.cs

@@ -41,17 +41,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Transport.Libuv.Internal
 
             while (true)
             {
-                ReadResult result;
+                var result = await _pipe.ReadAsync();
-
-                try
-                {
-                    result = await _pipe.ReadAsync();
-                }
-                catch
-                {
-                    // Handled in LibuvConnection.Abort()
-                    return;
-                }
 
                 var buffer = result.Buffer;
                 var consumed = buffer.End;
@@ -120,7 +110,11 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Transport.Libuv.Internal
             else
             {
                 // Log connection resets at a lower (Debug) level.
-                if (LibuvConstants.IsConnectionReset(status))
+                if (status == LibuvConstants.ECANCELED)
+                {
+                    // Connection was aborted.
+                }
+                else if (LibuvConstants.IsConnectionReset(status))
                 {
                     _log.ConnectionReset(_connectionId);
                 }

src/Kestrel.Transport.Libuv/Internal/Listener.cs

@@ -93,7 +93,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Transport.Libuv.Internal
             try
             {
                 pipe.Init(Thread.Loop, Thread.QueueCloseHandle, false);
-                
+
                 if (!useFileHandle)
                 {
                     pipe.Bind(EndPointInformation.SocketPath);
@@ -172,6 +172,11 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Transport.Libuv.Internal
                 listenSocket.Accept(acceptSocket);
                 DispatchConnection(acceptSocket);
             }
+            catch (UvException ex) when (LibuvConstants.IsConnectionReset(ex.StatusCode))
+            {
+                Log.ConnectionReset("(null)");
+                acceptSocket?.Dispose();
+            }
             catch (UvException ex)
             {
                 Log.LogError(0, ex, "Listener.OnConnection");
@@ -181,8 +186,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Transport.Libuv.Internal
 
         protected virtual void DispatchConnection(UvStreamHandle socket)
         {
-            var connection = new LibuvConnection(this, socket);
+            HandleConnectionAsync(socket);
-            _ = connection.Start();
         }
 
         public virtual async Task DisposeAsync()

src/Kestrel.Transport.Libuv/Internal/ListenerContext.cs

@@ -2,8 +2,11 @@
 // Licensed under the Apache License, Version 2.0. See License.txt in the project root for license information.
 
 using System;
+using System.Net;
+using System.Threading.Tasks;
 using Microsoft.AspNetCore.Server.Kestrel.Transport.Abstractions.Internal;
 using Microsoft.AspNetCore.Server.Kestrel.Transport.Libuv.Internal.Networking;
+using Microsoft.Extensions.Logging;
 
 namespace Microsoft.AspNetCore.Server.Kestrel.Transport.Libuv.Internal
 {
@@ -38,6 +41,38 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Transport.Libuv.Internal
             }
         }
 
+        protected void HandleConnectionAsync(UvStreamHandle socket)
+        {
+            try
+            {
+                IPEndPoint remoteEndPoint = null;
+                IPEndPoint localEndPoint = null;
+
+                if (socket is UvTcpHandle tcpHandle)
+                {
+                    try
+                    {
+                        remoteEndPoint = tcpHandle.GetPeerIPEndPoint();
+                        localEndPoint = tcpHandle.GetSockIPEndPoint();
+                    }
+                    catch (UvException ex) when (LibuvConstants.IsConnectionReset(ex.StatusCode))
+                    {
+                        TransportContext.Log.ConnectionReset("(null)");
+                        socket.Dispose();
+                        return;
+                    }
+                }
+
+                var connection = new LibuvConnection(socket, TransportContext.Log, Thread, remoteEndPoint, localEndPoint);
+                TransportContext.ConnectionDispatcher.OnConnection(connection);
+                _ = connection.Start();
+            }
+            catch (Exception ex)
+            {
+                TransportContext.Log.LogCritical(ex, $"Unexpected exception in {nameof(ListenerContext)}.{nameof(HandleConnectionAsync)}.");
+            }
+        }
+
         private UvTcpHandle AcceptTcp()
         {
             var socket = new UvTcpHandle(TransportContext.Log);

src/Kestrel.Transport.Libuv/Internal/ListenerSecondary.cs

@@ -151,23 +151,17 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Transport.Libuv.Internal
             try
             {
                 DispatchPipe.Accept(acceptSocket);
+                HandleConnectionAsync(acceptSocket);
+            }
+            catch (UvException ex) when (LibuvConstants.IsConnectionReset(ex.StatusCode))
+            {
+                Log.ConnectionReset("(null)");
+                acceptSocket.Dispose();
             }
             catch (UvException ex)
             {
                 Log.LogError(0, ex, "DispatchPipe.Accept");
                 acceptSocket.Dispose();
-                return;
-            }
-
-            try
-            {
-                var connection = new LibuvConnection(this, acceptSocket);
-                _ = connection.Start();
-            }
-            catch (UvException ex)
-            {
-                Log.LogError(0, ex, "ListenerSecondary.OnConnection");
-                acceptSocket.Dispose();
             }
         }
 

src/Kestrel.Transport.Sockets/Internal/SocketConnection.cs

@@ -31,6 +31,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Transport.Sockets.Internal
 
         private readonly object _shutdownLock = new object();
         private volatile bool _aborted;
+        private volatile ConnectionAbortedException _abortReason;
         private long _totalBytesWritten;
 
         internal SocketConnection(Socket socket, MemoryPool<byte> memoryPool, PipeScheduler scheduler, ISocketsTrace trace)
@@ -69,12 +70,10 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Transport.Sockets.Internal
         public override PipeScheduler OutputReaderScheduler => _scheduler;
         public override long TotalBytesWritten => Interlocked.Read(ref _totalBytesWritten);
 
-        public async Task StartAsync(IConnectionDispatcher connectionDispatcher)
+        public async Task StartAsync()
         {
             try
             {
-                connectionDispatcher.OnConnection(this);
-
                 // Spawn send and receive logic
                 var receiveTask = DoReceive();
                 var sendTask = DoSend();
@@ -93,8 +92,11 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Transport.Sockets.Internal
             }
         }
 
-        public override void Abort()
+        public override void Abort(ConnectionAbortedException abortReason)
         {
+            _abortReason = abortReason;
+            Output.CancelPendingRead();
+
             // Try to gracefully close the socket to match libuv behavior.
             Shutdown();
         }
@@ -107,20 +109,20 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Transport.Sockets.Internal
             {
                 await ProcessReceives();
             }
-            catch (SocketException ex) when (ex.SocketErrorCode == SocketError.ConnectionReset)
+            catch (SocketException ex) when (IsConnectionResetError(ex.SocketErrorCode))
             {
-                error = new ConnectionResetException(ex.Message, ex);
+                // A connection reset can be reported as SocketError.ConnectionAborted on Windows
-                _trace.ConnectionReset(ConnectionId);
+                if (!_aborted)
+                {
+                    error = new ConnectionResetException(ex.Message, ex);
+                    _trace.ConnectionReset(ConnectionId);
+                }
             }
-            catch (SocketException ex) when (ex.SocketErrorCode == SocketError.OperationAborted ||
+            catch (SocketException ex) when (IsConnectionAbortError(ex.SocketErrorCode))
-                                             ex.SocketErrorCode == SocketError.ConnectionAborted ||
-                                             ex.SocketErrorCode == SocketError.Interrupted ||
-                                             ex.SocketErrorCode == SocketError.InvalidArgument)
             {
                 if (!_aborted)
                 {
                     // Calling Dispose after ReceiveAsync can cause an "InvalidArgument" error on *nix.
-                    error = new ConnectionAbortedException();
                     _trace.ConnectionError(ConnectionId, error);
                 }
             }
@@ -128,7 +130,6 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Transport.Sockets.Internal
             {
                 if (!_aborted)
                 {
-                    error = new ConnectionAbortedException();
                     _trace.ConnectionError(ConnectionId, error);
                 }
             }
@@ -146,7 +147,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Transport.Sockets.Internal
             {
                 if (_aborted)
                 {
-                    error = error ?? new ConnectionAbortedException();
+                    error = error ?? _abortReason ?? new ConnectionAbortedException();
                 }
 
                 Input.Complete(error);
@@ -199,7 +200,13 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Transport.Sockets.Internal
             {
                 await ProcessSends();
             }
-            catch (SocketException ex) when (ex.SocketErrorCode == SocketError.OperationAborted)
+            catch (SocketException ex) when (IsConnectionResetError(ex.SocketErrorCode))
+            {
+                // A connection reset can be reported as SocketError.ConnectionAborted on Windows
+                error = null;
+                _trace.ConnectionReset(ConnectionId);
+            }
+            catch (SocketException ex) when (IsConnectionAbortError(ex.SocketErrorCode))
             {
                 error = null;
             }
@@ -210,10 +217,12 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Transport.Sockets.Internal
             catch (IOException ex)
             {
                 error = ex;
+                _trace.ConnectionError(ConnectionId, error);
             }
             catch (Exception ex)
             {
                 error = new IOException(ex.Message, ex);
+                _trace.ConnectionError(ConnectionId, error);
             }
             finally
             {
@@ -228,8 +237,8 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Transport.Sockets.Internal
         {
             while (true)
             {
-                // Wait for data to write from the pipe producer
                 var result = await Output.ReadAsync();
+
                 var buffer = result.Buffer;
 
                 if (result.IsCanceled)
@@ -289,12 +298,25 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Transport.Sockets.Internal
             try
             {
                 _connectionClosedTokenSource.Cancel();
-                _connectionClosedTokenSource.Dispose();
             }
             catch (Exception ex)
             {
                 _trace.LogError(0, ex, $"Unexpected exception in {nameof(SocketConnection)}.{nameof(CancelConnectionClosedToken)}.");
             }
         }
+
+        private static bool IsConnectionResetError(SocketError errorCode)
+        {
+            return errorCode == SocketError.ConnectionReset ||
+                   errorCode == SocketError.ConnectionAborted ||
+                   errorCode == SocketError.Shutdown;
+        }
+
+        private static bool IsConnectionAbortError(SocketError errorCode)
+        {
+            return errorCode == SocketError.OperationAborted ||
+                   errorCode == SocketError.Interrupted ||
+                   errorCode == SocketError.InvalidArgument;
+        }
     }
 }

src/Kestrel.Transport.Sockets/SocketTransport.cs

@@ -155,17 +155,12 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Transport.Sockets
                             acceptSocket.NoDelay = _endPointInformation.NoDelay;
 
                             var connection = new SocketConnection(acceptSocket, _memoryPool, _schedulers[schedulerIndex], _trace);
-                            _ = connection.StartAsync(_dispatcher);
+                            HandleConnectionAsync(connection);
                         }
-                        catch (SocketException ex) when (ex.SocketErrorCode == SocketError.ConnectionReset)
+                        catch (SocketException) when (!_unbinding)
                         {
-                            // REVIEW: Should there be a separate log message for a connection reset this early?
                             _trace.ConnectionReset(connectionId: "(null)");
                         }
-                        catch (SocketException ex) when (!_unbinding)
-                        {
-                            _trace.ConnectionError(connectionId: "(null)", ex);
-                        }
                     }
                 }
             }
@@ -177,7 +172,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Transport.Sockets
                 }
                 else
                 {
-                    _trace.LogCritical(ex, $"Unexpected exeption in {nameof(SocketTransport)}.{nameof(RunAcceptLoopAsync)}.");
+                    _trace.LogCritical(ex, $"Unexpected exception in {nameof(SocketTransport)}.{nameof(RunAcceptLoopAsync)}.");
                     _listenException = ex;
 
                     // Request shutdown so we can rethrow this exception
@@ -187,6 +182,19 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Transport.Sockets
             }
         }
 
+        private void HandleConnectionAsync(SocketConnection connection)
+        {
+            try
+            {
+                _dispatcher.OnConnection(connection);
+                _ = connection.StartAsync();
+            }
+            catch (Exception ex)
+            {
+                _trace.LogCritical(ex, $"Unexpected exception in {nameof(SocketTransport)}.{nameof(HandleConnectionAsync)}.");
+            }
+        }
+
         [DllImport("libc", SetLastError = true)]
         private static extern int setsockopt(int socket, int level, int option_name, IntPtr option_value, uint option_len);
 

test/Kestrel.Core.Tests/ConnectionDispatcherTests.cs

@@ -1,12 +1,13 @@
-using System;
+// Copyright (c) .NET Foundation. All rights reserved.
-using System.Buffers;
+// Licensed under the Apache License, Version 2.0. See License.txt in the project root for license information.
+
 using System.Collections.Generic;
-using System.IO.Pipelines;
 using System.Linq;
 using System.Threading.Tasks;
 using Microsoft.AspNetCore.Server.Kestrel.Core.Internal;
 using Microsoft.AspNetCore.Server.Kestrel.Transport.Abstractions.Internal;
 using Microsoft.AspNetCore.Testing;
+using Moq;
 using Xunit;
 
 namespace Microsoft.AspNetCore.Server.Kestrel.Core.Tests
@@ -20,7 +21,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Tests
             var tcs = new TaskCompletionSource<object>();
             var dispatcher = new ConnectionDispatcher(serviceContext, _ => tcs.Task);
 
-            var connection = new TransportConnection();
+            var connection = Mock.Of<TransportConnection>();
 
             dispatcher.OnConnection(connection);
 

test/Kestrel.Core.Tests/Http1ConnectionTests.cs

@@ -11,6 +11,7 @@ using System.Linq;
 using System.Text;
 using System.Threading;
 using System.Threading.Tasks;
+using Microsoft.AspNetCore.Connections;
 using Microsoft.AspNetCore.Connections.Features;
 using Microsoft.AspNetCore.Http;
 using Microsoft.AspNetCore.Http.Features;
@@ -58,6 +59,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Tests
             _http1ConnectionContext = new Http1ConnectionContext
             {
                 ServiceContext = _serviceContext,
+                ConnectionContext = Mock.Of<ConnectionContext>(),
                 ConnectionFeatures = connectionFeatures,
                 MemoryPool = _pipelineFactory,
                 TimeoutControl = _timeoutControl.Object,
@@ -596,7 +598,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Tests
             _http1Connection.StopProcessingNextRequest();
             Assert.IsNotType<Task<Task>>(requestProcessingTask);
 
-            await requestProcessingTask.TimeoutAfter(TestConstants.DefaultTimeout);
+            await requestProcessingTask.DefaultTimeout();
             _application.Output.Complete();
         }
 
@@ -732,7 +734,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Tests
             Assert.Equal(header0Count + header1Count, _http1Connection.RequestHeaders.Count);
 
             await _application.Output.WriteAsync(Encoding.ASCII.GetBytes("\r\n"));
-            await requestProcessingTask.TimeoutAfter(TestConstants.DefaultTimeout);
+            await requestProcessingTask.DefaultTimeout();
         }
 
         [Theory]
@@ -809,7 +811,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Tests
 
             var data = Encoding.ASCII.GetBytes("POST / HTTP/1.1\r\nHost:\r\nConnection: close\r\ncontent-length: 1\r\n\r\n");
             await _application.Output.WriteAsync(data);
-            await requestProcessingTask.TimeoutAfter(TestConstants.DefaultTimeout);
+            await requestProcessingTask.DefaultTimeout();
 
             mockMessageBody.Verify(body => body.ConsumeAsync(), Times.Once);
         }

test/Kestrel.Core.Tests/Http2ConnectionTests.cs

@@ -213,7 +213,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Tests
                     sem.Release();
                 });
 
-                await sem.WaitAsync().TimeoutAfter(TestConstants.DefaultTimeout);
+                await sem.WaitAsync().DefaultTimeout();
             };
 
             _largeHeadersApplication = context =>
@@ -241,7 +241,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Tests
                     sem.Release();
                 });
 
-                await sem.WaitAsync().TimeoutAfter(TestConstants.DefaultTimeout);
+                await sem.WaitAsync().DefaultTimeout();
 
                 _runningStreams[streamIdFeature.StreamId].TrySetResult(null);
             };
@@ -261,7 +261,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Tests
                     sem.Release();
                 });
 
-                await sem.WaitAsync().TimeoutAfter(TestConstants.DefaultTimeout);
+                await sem.WaitAsync().DefaultTimeout();
 
                 await context.Response.Body.FlushAsync();
 
@@ -2459,7 +2459,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Tests
 
         private Task WaitForAllStreamsAsync()
         {
-            return Task.WhenAll(_runningStreams.Values.Select(tcs => tcs.Task)).TimeoutAfter(TestConstants.DefaultTimeout);
+            return Task.WhenAll(_runningStreams.Values.Select(tcs => tcs.Task)).DefaultTimeout();
         }
 
         private Task SendAsync(ReadOnlySpan<byte> span)

test/Kestrel.Core.Tests/HttpConnectionTests.cs

@@ -7,6 +7,7 @@ using System.Collections.Generic;
 using System.IO.Pipelines;
 using System.Threading;
 using System.Threading.Tasks;
+using Microsoft.AspNetCore.Connections;
 using Microsoft.AspNetCore.Connections.Features;
 using Microsoft.AspNetCore.Http.Features;
 using Microsoft.AspNetCore.Server.Kestrel.Core.Adapter.Internal;
@@ -38,6 +39,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Tests
             _httpConnectionContext = new HttpConnectionContext
             {
                 ConnectionId = "0123456789",
+                ConnectionContext = Mock.Of<ConnectionContext>(),
                 ConnectionAdapters = new List<IConnectionAdapter>(),
                 ConnectionFeatures = connectionFeatures,
                 MemoryPool = _memoryPool,

test/Kestrel.Core.Tests/KestrelServerTests.cs

@@ -263,7 +263,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Tests
             unbind.Release();
             stop.Release();
 
-            await Task.WhenAll(new[] { stopTask1, stopTask2, stopTask3 }).TimeoutAfter(TestConstants.DefaultTimeout);
+            await Task.WhenAll(new[] { stopTask1, stopTask2, stopTask3 }).DefaultTimeout();
 
             mockTransport.Verify(transport => transport.UnbindAsync(), Times.Once);
             mockTransport.Verify(transport => transport.StopAsync(), Times.Once);

test/Kestrel.Core.Tests/MessageBodyTests.cs

@@ -148,7 +148,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Tests
 
                 input.Add("\r\r\r\nHello\r\n0\r\n\r\n");
 
-                Assert.Equal(5, await readTask.TimeoutAfter(TestConstants.DefaultTimeout));
+                Assert.Equal(5, await readTask.DefaultTimeout());
                 Assert.Equal(0, await stream.ReadAsync(buffer, 0, buffer.Length));
 
                 await body.StopAsync();

test/Kestrel.Core.Tests/OutputProducerTests.cs

@@ -5,6 +5,7 @@ using System;
 using System.Buffers;
 using System.IO.Pipelines;
 using System.Threading;
+using Microsoft.AspNetCore.Connections;
 using Microsoft.AspNetCore.Connections.Features;
 using Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http;
 using Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Infrastructure;
@@ -61,39 +62,38 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Tests
         [Fact]
         public void AbortsTransportEvenAfterDispose()
         {
-            var mockLifetimeFeature = new Mock<IConnectionLifetimeFeature>();
+            var mockConnectionContext = new Mock<ConnectionContext>();
 
-            var outputProducer = CreateOutputProducer(lifetimeFeature: mockLifetimeFeature.Object);
+            var outputProducer = CreateOutputProducer(connectionContext: mockConnectionContext.Object);
 
             outputProducer.Dispose();
 
-            mockLifetimeFeature.Verify(f => f.Abort(), Times.Never());
+            mockConnectionContext.Verify(f => f.Abort(It.IsAny<ConnectionAbortedException>()), Times.Never());
 
             outputProducer.Abort(null);
 
-            mockLifetimeFeature.Verify(f => f.Abort(), Times.Once());
+            mockConnectionContext.Verify(f => f.Abort(null), Times.Once());
 
             outputProducer.Abort(null);
 
-            mockLifetimeFeature.Verify(f => f.Abort(), Times.Once());
+            mockConnectionContext.Verify(f => f.Abort(null), Times.Once());
         }
 
         private Http1OutputProducer CreateOutputProducer(
             PipeOptions pipeOptions = null,
-            IConnectionLifetimeFeature lifetimeFeature = null)
+            ConnectionContext connectionContext = null)
         {
             pipeOptions = pipeOptions ?? new PipeOptions();
-            lifetimeFeature = lifetimeFeature ?? Mock.Of<IConnectionLifetimeFeature>();
+            connectionContext = connectionContext ?? Mock.Of<ConnectionContext>();
 
             var pipe = new Pipe(pipeOptions);
             var serviceContext = new TestServiceContext();
             var socketOutput = new Http1OutputProducer(
-                pipe.Reader,
                 pipe.Writer,
                 "0",
+                connectionContext,
                 serviceContext.Log,
                 Mock.Of<ITimeoutControl>(),
-                lifetimeFeature,
                 Mock.Of<IBytesWrittenFeature>());
 
             return socketOutput;

test/Kestrel.Core.Tests/TestInput.cs

@@ -6,6 +6,7 @@ using System.Buffers;
 using System.IO.Pipelines;
 using System.Text;
 using System.Threading.Tasks;
+using Microsoft.AspNetCore.Connections;
 using Microsoft.AspNetCore.Connections.Features;
 using Microsoft.AspNetCore.Http.Features;
 using Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http;
@@ -35,6 +36,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Tests
             Http1ConnectionContext = new Http1ConnectionContext
             {
                 ServiceContext = new TestServiceContext(),
+                ConnectionContext = Mock.Of<ConnectionContext>(),
                 ConnectionFeatures = connectionFeatures,
                 Application = Application,
                 Transport = Transport,

test/Kestrel.FunctionalTests/ConnectionLimitTests.cs

@@ -38,11 +38,11 @@ namespace Microsoft.AspNetCore.Server.Kestrel.FunctionalTests
             {
                 await connection.SendEmptyGetAsKeepAlive(); ;
                 await connection.Receive("HTTP/1.1 200 OK");
-                Assert.True(await lockedTcs.Task.TimeoutAfter(TestConstants.DefaultTimeout));
+                Assert.True(await lockedTcs.Task.DefaultTimeout());
                 requestTcs.TrySetResult(null);
             }
 
-            await releasedTcs.Task.TimeoutAfter(TestConstants.DefaultTimeout);
+            await releasedTcs.Task.DefaultTimeout();
         }
 
         [Fact]
@@ -86,7 +86,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.FunctionalTests
                     catch { }
 
                     // connection should close without sending any data
-                    await rejected.WaitForConnectionClose().TimeoutAfter(TestConstants.DefaultTimeout);
+                    await rejected.WaitForConnectionClose().DefaultTimeout();
                 }
             }
         }
@@ -126,7 +126,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.FunctionalTests
                         catch { }
 
                         // connection should close without sending any data
-                        await connection.WaitForConnectionClose().TimeoutAfter(TestConstants.DefaultTimeout);
+                        await connection.WaitForConnectionClose().DefaultTimeout();
                     }
                 }
 

test/Kestrel.FunctionalTests/EventSourceTests.cs

@@ -44,7 +44,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.FunctionalTests
                         "Host:",
                         "",
                         "")
-                        .TimeoutAfter(TestConstants.DefaultTimeout);
+                        .DefaultTimeout();
                     await connection.Receive("HTTP/1.1 200");
                 }
             }

test/Kestrel.FunctionalTests/HttpProtocolSelectionTests.cs

@@ -85,7 +85,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.FunctionalTests
 
                 using (var connection = new TestConnection(host.GetPort()))
                 {
-                    await connection.WaitForConnectionClose().TimeoutAfter(TestConstants.DefaultTimeout);
+                    await connection.WaitForConnectionClose().DefaultTimeout();
                 }
             }
 

test/Kestrel.FunctionalTests/HttpsTests.cs

@@ -142,7 +142,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.FunctionalTests
                     // Close socket immediately
                 }
 
-                await loggerProvider.FilterLogger.LogTcs.Task.TimeoutAfter(TestConstants.DefaultTimeout);
+                await loggerProvider.FilterLogger.LogTcs.Task.DefaultTimeout();
             }
 
             Assert.Equal(1, loggerProvider.FilterLogger.LastEventId.Id);
@@ -179,7 +179,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.FunctionalTests
                     await stream.WriteAsync(new byte[10], 0, 10);
                 }
 
-                await loggerProvider.FilterLogger.LogTcs.Task.TimeoutAfter(TestConstants.DefaultTimeout);
+                await loggerProvider.FilterLogger.LogTcs.Task.DefaultTimeout();
             }
 
             Assert.Equal(1, loggerProvider.FilterLogger.LastEventId.Id);
@@ -292,7 +292,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.FunctionalTests
                 }
             }
 
-            await tcs.Task.TimeoutAfter(TestConstants.DefaultTimeout);
+            await tcs.Task.DefaultTimeout();
         }
 
         // Regression test for https://github.com/aspnet/KestrelHttpServer/issues/1693
@@ -391,11 +391,11 @@ namespace Microsoft.AspNetCore.Server.Kestrel.FunctionalTests
                 using (var stream = new NetworkStream(socket, ownsSocket: false))
                 {
                     // No data should be sent and the connection should be closed in well under 30 seconds.
-                    Assert.Equal(0, await stream.ReadAsync(new byte[1], 0, 1).TimeoutAfter(TestConstants.DefaultTimeout));
+                    Assert.Equal(0, await stream.ReadAsync(new byte[1], 0, 1).DefaultTimeout());
                 }
             }
 
-            await loggerProvider.FilterLogger.LogTcs.Task.TimeoutAfter(TestConstants.DefaultTimeout);
+            await loggerProvider.FilterLogger.LogTcs.Task.DefaultTimeout();
             Assert.Equal(2, loggerProvider.FilterLogger.LastEventId);
             Assert.Equal(LogLevel.Debug, loggerProvider.FilterLogger.LastLogLevel);
         }

test/Kestrel.FunctionalTests/LoggingConnectionAdapterTests.cs

@@ -44,7 +44,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.FunctionalTests
                 await host.StartAsync();
 
                 var response = await HttpClientSlim.GetStringAsync($"https://localhost:{host.GetPort()}/", validateCertificate: false)
-                                                   .TimeoutAfter(TimeSpan.FromSeconds(10));
+                                                   .DefaultTimeout();
 
                 Assert.Equal("Hello World!", response);
             }

test/Kestrel.FunctionalTests/RequestTests.cs

@@ -312,8 +312,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.FunctionalTests
                     // Wait until connection is established
                     Assert.True(await connectionStarted.WaitAsync(TestConstants.DefaultTimeout));
 
-                    // Force a reset
+                    connection.Reset();
-                    connection.Socket.LingerState = new LingerOption(true, 0);
                 }
 
                 // If the reset is correctly logged as Debug, the wait below should complete shortly.
@@ -381,8 +380,8 @@ namespace Microsoft.AspNetCore.Server.Kestrel.FunctionalTests
                         "",
                         "");
 
+                    connection.Reset();
                     // Force a reset
-                    connection.Socket.LingerState = new LingerOption(true, 0);
                 }
 
                 // If the reset is correctly logged as Debug, the wait below should complete shortly.
@@ -450,8 +449,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.FunctionalTests
                     // Wait until connection is established
                     Assert.True(await requestStarted.WaitAsync(TestConstants.DefaultTimeout), "request should have started");
 
-                    // Force a reset
+                    connection.Reset();
-                    connection.Socket.LingerState = new LingerOption(true, 0);
                 }
 
                 // If the reset is correctly logged as Debug, the wait below should complete shortly.
@@ -528,7 +526,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.FunctionalTests
 
                     var token = context.RequestAborted;
                     token.Register(() => requestAborted.Release(2));
-                    await requestAborted.WaitAsync().TimeoutAfter(TestConstants.DefaultTimeout);
+                    await requestAborted.WaitAsync().DefaultTimeout();
                 }));
 
             using (var host = builder.Build())
@@ -541,7 +539,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.FunctionalTests
                     socket.Send(Encoding.ASCII.GetBytes("GET / HTTP/1.1\r\nHost:\r\n\r\n"));
                     await appStarted.WaitAsync();
                     socket.Shutdown(SocketShutdown.Send);
-                    await requestAborted.WaitAsync().TimeoutAfter(TestConstants.DefaultTimeout);
+                    await requestAborted.WaitAsync().DefaultTimeout();
                 }
             }
         }
@@ -599,11 +597,11 @@ namespace Microsoft.AspNetCore.Server.Kestrel.FunctionalTests
                         "",
                         "");
 
-                    await appStartedTcs.Task.TimeoutAfter(TestConstants.DefaultTimeout);
+                    await appStartedTcs.Task.DefaultTimeout();
 
-                    connection.Socket.Shutdown(SocketShutdown.Send);
+                    connection.Shutdown(SocketShutdown.Send);
 
-                    await connectionClosedTcs.Task.TimeoutAfter(TestConstants.DefaultTimeout);
+                    await connectionClosedTcs.Task.DefaultTimeout();
                 }
             }
         }
@@ -632,7 +630,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.FunctionalTests
                         "",
                         "");
 
-                    await connectionClosedTcs.Task.TimeoutAfter(TestConstants.DefaultTimeout);
+                    await connectionClosedTcs.Task.DefaultTimeout();
 
                     await connection.ReceiveEnd($"HTTP/1.1 200 OK",
                         "Connection: close",
@@ -669,7 +667,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.FunctionalTests
                         "",
                         "");
 
-                    await connectionClosedTcs.Task.TimeoutAfter(TestConstants.DefaultTimeout);
+                    await connectionClosedTcs.Task.DefaultTimeout();
                     await connection.ReceiveForcedEnd();
                 }
             }
@@ -728,9 +726,9 @@ namespace Microsoft.AspNetCore.Server.Kestrel.FunctionalTests
                         "",
                         "4",
                         "Done")
-                        .TimeoutAfter(TestConstants.DefaultTimeout);
+                        .DefaultTimeout();
 
-                    await Task.WhenAll(pathTcs.Task, rawTargetTcs.Task, queryTcs.Task).TimeoutAfter(TestConstants.DefaultTimeout);
+                    await Task.WhenAll(pathTcs.Task, rawTargetTcs.Task, queryTcs.Task).DefaultTimeout();
                     Assert.Equal(new PathString(expectedPath), pathTcs.Task.Result);
                     Assert.Equal(requestUrl, rawTargetTcs.Task.Result);
                     if (queryValue == null)
@@ -756,7 +754,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.FunctionalTests
             }, new TestServiceContext(LoggerFactory)))
             {
                 var requestId = await HttpClientSlim.GetStringAsync($"http://{server.EndPoint}")
-                    .TimeoutAfter(TestConstants.DefaultTimeout);
+                    .DefaultTimeout();
                 Assert.Equal(knownId, requestId);
             }
         }
@@ -797,7 +795,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.FunctionalTests
                            $"Date: {server.Context.DateHeaderValue}",
                            $"Content-Length: {identifierLength}",
                            "",
-                           "").TimeoutAfter(TestConstants.DefaultTimeout);
+                           "").DefaultTimeout();
 
                         var read = await connection.Reader.ReadAsync(buffer, 0, identifierLength);
                         Assert.Equal(identifierLength, read);
@@ -1058,7 +1056,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.FunctionalTests
             using (var server = new TestServer(async httpContext =>
             {
                 // This will hang if 0 content length is not assumed by the server
-                Assert.Equal(0, await httpContext.Request.Body.ReadAsync(new byte[1], 0, 1).TimeoutAfter(TestConstants.DefaultTimeout));
+                Assert.Equal(0, await httpContext.Request.Body.ReadAsync(new byte[1], 0, 1).DefaultTimeout());
             }, testContext, listenOptions))
             {
                 using (var connection = server.CreateConnection())
@@ -1133,6 +1131,8 @@ namespace Microsoft.AspNetCore.Server.Kestrel.FunctionalTests
         [MemberData(nameof(ConnectionAdapterData))]
         public async Task RequestsCanBeAbortedMidRead(ListenOptions listenOptions)
         {
+            const int applicationAbortedConnectionId = 34;
+
             var testContext = new TestServiceContext(LoggerFactory);
 
             var readTcs = new TaskCompletionSource<object>(TaskCreationOptions.RunContinuationsAsynchronously);
@@ -1207,6 +1207,9 @@ namespace Microsoft.AspNetCore.Server.Kestrel.FunctionalTests
             // The cancellation token for only the last request should be triggered.
             var abortedRequestId = await registrationTcs.Task;
             Assert.Equal(2, abortedRequestId);
+
+            Assert.Single(TestSink.Writes.Where(w => w.LoggerName == "Microsoft.AspNetCore.Server.Kestrel" &&
+                                                     w.EventId == applicationAbortedConnectionId));
         }
 
         [Theory]
@@ -1291,12 +1294,12 @@ namespace Microsoft.AspNetCore.Server.Kestrel.FunctionalTests
                     var ignore = connection.Stream.WriteAsync(scratchBuffer, 0, scratchBuffer.Length);
 
                     // Wait until the read callback is no longer hooked up so that the connection disconnect isn't observed.
-                    await readCallbackUnwired.Task.TimeoutAfter(TestConstants.DefaultTimeout);
+                    await readCallbackUnwired.Task.DefaultTimeout();
                 }
 
                 clientClosedConnection.SetResult(null);
 
-                await appFuncCompleted.Task.TimeoutAfter(TestConstants.DefaultTimeout);
+                await appFuncCompleted.Task.DefaultTimeout();
             }
 
             mockKestrelTrace.Verify(t => t.ConnectionStop(It.IsAny<string>()), Times.Once());
@@ -1306,7 +1309,8 @@ namespace Microsoft.AspNetCore.Server.Kestrel.FunctionalTests
         [MemberData(nameof(ConnectionAdapterData))]
         public async Task AppCanHandleClientAbortingConnectionMidRequest(ListenOptions listenOptions)
         {
-            var readTcs = new TaskCompletionSource<Exception>(TaskCreationOptions.RunContinuationsAsynchronously);
+            var readTcs = new TaskCompletionSource<object>(TaskCreationOptions.RunContinuationsAsynchronously);
+            var appStartedTcs = new TaskCompletionSource<object>(TaskCreationOptions.RunContinuationsAsynchronously);
 
             var mockKestrelTrace = new Mock<KestrelTrace>(Logger) { CallBase = true };
             var testContext = new TestServiceContext()
@@ -1318,6 +1322,8 @@ namespace Microsoft.AspNetCore.Server.Kestrel.FunctionalTests
 
             using (var server = new TestServer(async context =>
             {
+                appStartedTcs.SetResult(null);
+
                 try
                 {
                     await context.Request.Body.CopyToAsync(Stream.Null);;
@@ -1341,10 +1347,14 @@ namespace Microsoft.AspNetCore.Server.Kestrel.FunctionalTests
                         "",
                         "");
 
+                    await appStartedTcs.Task.DefaultTimeout();
+
                     await connection.Stream.WriteAsync(scratchBuffer, 0, scratchBuffer.Length);
+
+                    connection.Reset();
                 }
 
-                await Assert.ThrowsAnyAsync<IOException>(() => readTcs.Task).TimeoutAfter(TestConstants.DefaultTimeout);
+                await Assert.ThrowsAnyAsync<IOException>(() => readTcs.Task).DefaultTimeout();
             }
 
             mockKestrelTrace.Verify(t => t.ConnectionStop(It.IsAny<string>()), Times.Once());
@@ -1418,7 +1428,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.FunctionalTests
                 var read = 0;
                 while (read < message.Length)
                 {
-                    read += await duplexStream.ReadAsync(buffer, read, buffer.Length - read).TimeoutAfter(TestConstants.DefaultTimeout);
+                    read += await duplexStream.ReadAsync(buffer, read, buffer.Length - read).DefaultTimeout();
                 }
 
                 await duplexStream.WriteAsync(buffer, 0, read);

test/Kestrel.FunctionalTests/ResponseTests.cs

@@ -176,7 +176,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.FunctionalTests
 
                     Assert.Equal(HttpStatusCode.InternalServerError, response.StatusCode);
                     Assert.False(onStartingCalled);
-                    await onCompletedTcs.Task.TimeoutAfter(TestConstants.DefaultTimeout);
+                    await onCompletedTcs.Task.DefaultTimeout();
                 }
             }
         }
@@ -403,7 +403,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.FunctionalTests
                 }
             }
 
-            await onCompletedTcs.Task.TimeoutAfter(TestConstants.DefaultTimeout);
+            await onCompletedTcs.Task.DefaultTimeout();
         }
 
         private static async Task ResponseStatusCodeSetBeforeHttpContextDispose(
@@ -479,7 +479,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.FunctionalTests
                     }
                 }
 
-                var disposedStatusCode = await disposedTcs.Task.TimeoutAfter(TestConstants.DefaultTimeout);
+                var disposedStatusCode = await disposedTcs.Task.DefaultTimeout();
                 Assert.Equal(expectedServerStatusCode, (HttpStatusCode)disposedStatusCode);
             }
 
@@ -647,7 +647,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.FunctionalTests
                     // Wait for message to be logged before disposing the socket.
                     // Disposing the socket will abort the connection and HttpProtocol._requestAborted
                     // might be 1 by the time ProduceEnd() gets called and the message is logged.
-                    await logTcs.Task.TimeoutAfter(TestConstants.DefaultTimeout);
+                    await logTcs.Task.DefaultTimeout();
                 }
             }
 
@@ -685,7 +685,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.FunctionalTests
                         "",
                         "hello,");
 
-                    await connection.WaitForConnectionClose().TimeoutAfter(TestConstants.DefaultTimeout);
+                    await connection.WaitForConnectionClose().DefaultTimeout();
                 }
             }
 
@@ -843,10 +843,10 @@ namespace Microsoft.AspNetCore.Server.Kestrel.FunctionalTests
                         "hello, world");
 
                     // Wait for error message to be logged.
-                    await logTcs.Task.TimeoutAfter(TestConstants.DefaultTimeout);
+                    await logTcs.Task.DefaultTimeout();
 
                     // The server should close the connection in this situation.
-                    await connection.WaitForConnectionClose().TimeoutAfter(TestConstants.DefaultTimeout);
+                    await connection.WaitForConnectionClose().DefaultTimeout();
                 }
             }
 
@@ -1210,12 +1210,12 @@ namespace Microsoft.AspNetCore.Server.Kestrel.FunctionalTests
 
                     requestStarted.Wait();
                     connection.Shutdown(SocketShutdown.Send);
-                    await connection.WaitForConnectionClose().TimeoutAfter(TestConstants.DefaultTimeout);
+                    await connection.WaitForConnectionClose().DefaultTimeout();
                 }
 
                 connectionClosed.Set();
 
-                await tcs.Task.TimeoutAfter(TestConstants.DefaultTimeout);
+                await tcs.Task.DefaultTimeout();
             }
         }
 
@@ -1249,7 +1249,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.FunctionalTests
                         "Transfer-Encoding: chunked",
                         "",
                         "gg");
-                    await responseWritten.WaitAsync().TimeoutAfter(TestConstants.DefaultTimeout);
+                    await responseWritten.WaitAsync().DefaultTimeout();
                     await connection.ReceiveEnd(
                         "HTTP/1.1 400 Bad Request",
                         $"Date: {server.Context.DateHeaderValue}",
@@ -1860,7 +1860,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.FunctionalTests
 
             using (var server = new TestServer(async httpContext =>
             {
-                Assert.Equal(0, await httpContext.Request.Body.ReadAsync(new byte[1], 0, 1).TimeoutAfter(TestConstants.DefaultTimeout));
+                Assert.Equal(0, await httpContext.Request.Body.ReadAsync(new byte[1], 0, 1).DefaultTimeout());
             }, testContext, listenOptions))
             {
                 using (var connection = server.CreateConnection())
@@ -2321,7 +2321,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.FunctionalTests
 
                 // Write failed - can throw TaskCanceledException or OperationCanceledException,
                 // dependending on how far the canceled write goes.
-                await Assert.ThrowsAnyAsync<OperationCanceledException>(async () => await writeTcs.Task).TimeoutAfter(TestConstants.DefaultTimeout);
+                await Assert.ThrowsAnyAsync<OperationCanceledException>(async () => await writeTcs.Task).DefaultTimeout();
 
                 // RequestAborted tripped
                 Assert.True(requestAbortedWh.Wait(TestConstants.DefaultTimeout));
@@ -2338,7 +2338,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.FunctionalTests
             var requestAborted = false;
             var readCallbackUnwired = new TaskCompletionSource<object>(TaskCreationOptions.RunContinuationsAsynchronously);
             var clientClosedConnection = new TaskCompletionSource<object>(TaskCreationOptions.RunContinuationsAsynchronously);
-            var writeTcs = new TaskCompletionSource<Exception>(TaskCreationOptions.RunContinuationsAsynchronously);
+            var writeTcs = new TaskCompletionSource<object>(TaskCreationOptions.RunContinuationsAsynchronously);
 
             var mockKestrelTrace = new Mock<KestrelTrace>(Logger) { CallBase = true };
             var mockLogger = new Mock<ILogger>();
@@ -2380,7 +2380,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.FunctionalTests
                 }
             };
 
-            var scratchBuffer = new byte[4096];
+            var scratchBuffer = new byte[maxRequestBufferSize * 8];
 
             using (var server = new TestServer(async context =>
             {
@@ -2420,12 +2420,12 @@ namespace Microsoft.AspNetCore.Server.Kestrel.FunctionalTests
                     var ignore = connection.Stream.WriteAsync(scratchBuffer, 0, scratchBuffer.Length);
 
                     // Wait until the read callback is no longer hooked up so that the connection disconnect isn't observed.
-                    await readCallbackUnwired.Task.TimeoutAfter(TestConstants.DefaultTimeout);
+                    await readCallbackUnwired.Task.DefaultTimeout();
                 }
 
                 clientClosedConnection.SetResult(null);
 
-                await Assert.ThrowsAnyAsync<OperationCanceledException>(() => writeTcs.Task).TimeoutAfter(TestConstants.DefaultTimeout);
+                await Assert.ThrowsAnyAsync<OperationCanceledException>(() => writeTcs.Task).DefaultTimeout();
             }
 
             mockKestrelTrace.Verify(t => t.ConnectionStop(It.IsAny<string>()), Times.Once());
@@ -2436,18 +2436,15 @@ namespace Microsoft.AspNetCore.Server.Kestrel.FunctionalTests
         [MemberData(nameof(ConnectionAdapterData))]
         public async Task AppCanHandleClientAbortingConnectionMidResponse(ListenOptions listenOptions)
         {
+            const int connectionResetEventId = 19;
+            const int connectionFinEventId = 6;
+            //const int connectionStopEventId = 2;
+
             const int responseBodySegmentSize = 65536;
             const int responseBodySegmentCount = 100;
-            const int responseBodySize = responseBodySegmentSize * responseBodySegmentCount;
 
-            var requestAborted = false;
             var appCompletedTcs = new TaskCompletionSource<object>(TaskCreationOptions.RunContinuationsAsynchronously);
-
+            var requestAborted = false;
-            var mockKestrelTrace = new Mock<KestrelTrace>(Logger) { CallBase = true };
-            var testContext = new TestServiceContext()
-            {
-                Log = mockKestrelTrace.Object,
-            };
 
             var scratchBuffer = new byte[responseBodySegmentSize];
 
@@ -2458,23 +2455,14 @@ namespace Microsoft.AspNetCore.Server.Kestrel.FunctionalTests
                     requestAborted = true;
                 });
 
-                context.Response.ContentLength = responseBodySize;
+                for (var i = 0; i < responseBodySegmentCount; i++)
+                {
+                    await context.Response.Body.WriteAsync(scratchBuffer, 0, scratchBuffer.Length);
+                    await Task.Delay(10);
+                }
 
-                try
+                appCompletedTcs.SetResult(null);
-                {
+            }, new TestServiceContext(LoggerFactory), listenOptions))
-                    for (var i = 0; i < responseBodySegmentCount; i++)
-                    {
-                        await context.Response.Body.WriteAsync(scratchBuffer, 0, scratchBuffer.Length);
-                        await Task.Delay(10);
-                    }
-                }
-                finally
-                {
-                    // WriteAsync shouldn't throw without a CancellationToken passed in. Unfortunately a ECONNRESET UvException sometimes gets thrown.
-                    // This will be fixed by https://github.com/aspnet/KestrelHttpServer/pull/2547
-                    appCompletedTcs.SetResult(null);
-                }
-            }, testContext, listenOptions))
             {
                 using (var connection = server.CreateConnection())
                 {
@@ -2484,23 +2472,69 @@ namespace Microsoft.AspNetCore.Server.Kestrel.FunctionalTests
                         "",
                         "");
 
-                    var readCount = 0;
-
                     // Read just part of the response and close the connection.
                     // https://github.com/aspnet/KestrelHttpServer/issues/2554
-                    for (var i = 0; i < responseBodySegmentCount / 10; i++)
+                    await connection.Stream.ReadAsync(scratchBuffer, 0, scratchBuffer.Length);
+
+                    connection.Reset();
+                }
+
+                await appCompletedTcs.Task.DefaultTimeout();
+
+                // After the app is done with the write loop, the connection reset should be logged.
+                // On Linux and macOS, the connection close is still sometimes observed as a FIN despite the LingerState.
+                var presShutdownTransportLogs = TestSink.Writes.Where(
+                    w => w.LoggerName == "Microsoft.AspNetCore.Server.Kestrel.Transport.Libuv" ||
+                         w.LoggerName == "Microsoft.AspNetCore.Server.Kestrel.Transport.Sockets");
+                var connectionResetLogs = presShutdownTransportLogs.Where(
+                    w => w.EventId == connectionResetEventId ||
+                         (!TestPlatformHelper.IsWindows && w.EventId == connectionFinEventId));
+
+                Assert.NotEmpty(connectionResetLogs);
+            }
+
+            // TODO: Figure out what the following assertion is flaky. The server shouldn't shutdown before all
+            // the connections are closed, yet sometimes the connection stop log isn't observed here.
+            //var coreLogs = TestSink.Writes.Where(w => w.LoggerName == "Microsoft.AspNetCore.Server.Kestrel");
+            //Assert.Single(coreLogs.Where(w => w.EventId == connectionStopEventId));
+
+            Assert.True(requestAborted, "RequestAborted token didn't fire.");
+
+            var transportLogs = TestSink.Writes.Where(w => w.LoggerName == "Microsoft.AspNetCore.Server.Kestrel.Transport.Libuv" ||
+                                                           w.LoggerName == "Microsoft.AspNetCore.Server.Kestrel.Transport.Sockets");
+            Assert.Empty(transportLogs.Where(w => w.LogLevel > LogLevel.Debug));
+        }
+
+        [Theory]
+        [MemberData(nameof(ConnectionAdapterData))]
+        public async Task ClientAbortingConnectionImmediatelyIsNotLoggedHigherThanDebug(ListenOptions listenOptions)
+        {
+            // Attempt multiple connections to be extra sure the resets are consistently logged appropriately.
+            const int numConnections = 10;
+
+            // There's not guarantee that the app even gets invoked in this test. The connection reset can be observed
+            // as early as accept.
+            using (var server = new TestServer(context => Task.CompletedTask, new TestServiceContext(LoggerFactory), listenOptions))
+            {
+                for (var i = 0; i < numConnections; i++)
+                {
+                    using (var connection = server.CreateConnection())
                     {
-                        readCount += await connection.Stream.ReadAsync(scratchBuffer, 0, scratchBuffer.Length);
+                        await connection.Send(
+                            "GET / HTTP/1.1",
+                            "Host:",
+                            "",
+                            "");
+
+                        connection.Reset();
                     }
-
-                    connection.Socket.Shutdown(SocketShutdown.Send);
-
-                    await appCompletedTcs.Task.TimeoutAfter(TestConstants.DefaultTimeout);
                 }
             }
 
-            mockKestrelTrace.Verify(t => t.ConnectionStop(It.IsAny<string>()), Times.Once());
+            var transportLogs = TestSink.Writes.Where(w => w.LoggerName == "Microsoft.AspNetCore.Server.Kestrel.Transport.Libuv" ||
-            Assert.True(requestAborted);
+                                                           w.LoggerName == "Microsoft.AspNetCore.Server.Kestrel.Transport.Sockets");
+
+            Assert.Empty(transportLogs.Where(w => w.LogLevel > LogLevel.Debug));
         }
 
         [Theory]
@@ -2654,7 +2688,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.FunctionalTests
                         "hello, world");
 
                     // Wait for all callbacks to be called.
-                    await onStartingTcs.Task.TimeoutAfter(TestConstants.DefaultTimeout);
+                    await onStartingTcs.Task.DefaultTimeout();
                 }
             }
 
@@ -2706,7 +2740,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.FunctionalTests
                         "hello, world");
 
                     // Wait for all callbacks to be called.
-                    await onCompletedTcs.Task.TimeoutAfter(TestConstants.DefaultTimeout);
+                    await onCompletedTcs.Task.DefaultTimeout();
                 }
             }
 
@@ -2884,10 +2918,10 @@ namespace Microsoft.AspNetCore.Server.Kestrel.FunctionalTests
                         var sw = Stopwatch.StartNew();
                         logger.LogInformation("Waiting for connection to abort.");
 
-                        await requestAborted.Task.TimeoutAfter(TimeSpan.FromSeconds(60));
+                        await requestAborted.Task.DefaultTimeout();
-                        await responseRateTimeoutMessageLogged.Task.TimeoutAfter(TimeSpan.FromSeconds(10));
+                        await responseRateTimeoutMessageLogged.Task.DefaultTimeout();
-                        await connectionStopMessageLogged.Task.TimeoutAfter(TimeSpan.FromSeconds(10));
+                        await connectionStopMessageLogged.Task.DefaultTimeout();
-                        await appFuncCompleted.Task.TimeoutAfter(TimeSpan.FromSeconds(10));
+                        await appFuncCompleted.Task.DefaultTimeout();
                         await AssertStreamAborted(connection.Reader.BaseStream, chunkSize * chunks);
 
                         sw.Stop();
@@ -2971,10 +3005,10 @@ namespace Microsoft.AspNetCore.Server.Kestrel.FunctionalTests
                         var request = Encoding.ASCII.GetBytes("GET / HTTP/1.1\r\nHost:\r\n\r\n");
                         await sslStream.WriteAsync(request, 0, request.Length);
 
-                        await aborted.Task.TimeoutAfter(TimeSpan.FromSeconds(60));
+                        await aborted.Task.DefaultTimeout();
-                        await responseRateTimeoutMessageLogged.Task.TimeoutAfter(TimeSpan.FromSeconds(10));
+                        await responseRateTimeoutMessageLogged.Task.DefaultTimeout();
-                        await connectionStopMessageLogged.Task.TimeoutAfter(TimeSpan.FromSeconds(10));
+                        await connectionStopMessageLogged.Task.DefaultTimeout();
-                        await appFuncCompleted.Task.TimeoutAfter(TimeSpan.FromSeconds(10));
+                        await appFuncCompleted.Task.DefaultTimeout();
 
                         // Temporary workaround for a deadlock when reading from an aborted client SslStream on Mac and Linux.
                         if (TestPlatformHelper.IsWindows)
@@ -2993,15 +3027,15 @@ namespace Microsoft.AspNetCore.Server.Kestrel.FunctionalTests
         [Fact]
         public async Task ConnectionClosedWhenBothRequestAndResponseExperienceBackPressure()
         {
-            const int bufferSize = 1024;
+            const int bufferSize = 65536;
-            const int bufferCount = 256 * 1024;
+            const int bufferCount = 100;
             var responseSize = bufferCount * bufferSize;
             var buffer = new byte[bufferSize];
 
             var responseRateTimeoutMessageLogged = new TaskCompletionSource<object>(TaskCreationOptions.RunContinuationsAsynchronously);
             var connectionStopMessageLogged = new TaskCompletionSource<object>(TaskCreationOptions.RunContinuationsAsynchronously);
             var requestAborted = new TaskCompletionSource<object>(TaskCreationOptions.RunContinuationsAsynchronously);
-            var appFuncCompleted = new TaskCompletionSource<object>(TaskCreationOptions.RunContinuationsAsynchronously);
+            var copyToAsyncCts = new TaskCompletionSource<object>(TaskCreationOptions.RunContinuationsAsynchronously);
 
             var mockKestrelTrace = new Mock<IKestrelTrace>();
             mockKestrelTrace
@@ -3039,13 +3073,13 @@ namespace Microsoft.AspNetCore.Server.Kestrel.FunctionalTests
                 {
                     await context.Request.Body.CopyToAsync(context.Response.Body);
                 }
-                catch
+                catch (Exception ex)
                 {
-                    // This should always throw an OperationCanceledException. Unfortunately a ECONNRESET UvException sometimes gets thrown.
+                    copyToAsyncCts.SetException(ex);
-                    // This will be fixed by https://github.com/aspnet/KestrelHttpServer/pull/2547
-                    appFuncCompleted.SetResult(null);
                     throw;
                 }
+
+                copyToAsyncCts.SetException(new Exception("This shouldn't be reached."));
             }
 
             using (var server = new TestServer(App, testContext, listenOptions))
@@ -3065,13 +3099,16 @@ namespace Microsoft.AspNetCore.Server.Kestrel.FunctionalTests
                         for (var i = 0; i < bufferCount; i++)
                         {
                             await connection.Stream.WriteAsync(buffer, 0, buffer.Length);
+                            await Task.Delay(10);
                         }
                     });
 
-                    await requestAborted.Task.TimeoutAfter(TimeSpan.FromSeconds(60));
+                    await requestAborted.Task.DefaultTimeout();
-                    await responseRateTimeoutMessageLogged.Task.TimeoutAfter(TimeSpan.FromSeconds(10));
+                    await responseRateTimeoutMessageLogged.Task.DefaultTimeout();
-                    await connectionStopMessageLogged.Task.TimeoutAfter(TimeSpan.FromSeconds(10));
+                    await connectionStopMessageLogged.Task.DefaultTimeout();
-                    await appFuncCompleted.Task.TimeoutAfter(TimeSpan.FromSeconds(10));
+
+                    // Expect OperationCanceledException instead of IOException because the server initiated the abort due to a response rate timeout.
+                    await Assert.ThrowsAnyAsync<OperationCanceledException>(() => copyToAsyncCts.Task).DefaultTimeout();
                     await AssertStreamAborted(connection.Stream, responseSize);
                 }
             }
@@ -3135,7 +3172,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.FunctionalTests
                     // Make sure consuming a single chunk exceeds the 2 second timeout.
                     var targetBytesPerSecond = chunkSize / 4;
                     await AssertStreamCompleted(connection.Reader.BaseStream, minTotalOutputSize, targetBytesPerSecond);
-                    await appFuncCompleted.Task.TimeoutAfter(TimeSpan.FromSeconds(10));
+                    await appFuncCompleted.Task.DefaultTimeout();
 
                     mockKestrelTrace.Verify(t => t.ResponseMininumDataRateNotSatisfied(It.IsAny<string>(), It.IsAny<string>()), Times.Never());
                     mockKestrelTrace.Verify(t => t.ConnectionStop(It.IsAny<string>()), Times.Once());
@@ -3257,7 +3294,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.FunctionalTests
             {
                 while (totalReceived < totalBytes)
                 {
-                    var bytes = await stream.ReadAsync(receiveBuffer, 0, receiveBuffer.Length).TimeoutAfter(TimeSpan.FromSeconds(10));
+                    var bytes = await stream.ReadAsync(receiveBuffer, 0, receiveBuffer.Length).DefaultTimeout();
 
                     if (bytes == 0)
                     {

test/Kestrel.FunctionalTests/UpgradeTests.cs

@@ -47,7 +47,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.FunctionalTests
                         "");
 
                     await connection.Receive("New protocol data");
-                    await upgrade.Task.TimeoutAfter(TestConstants.DefaultTimeout);
+                    await upgrade.Task.DefaultTimeout();
                 }
             }
         }
@@ -67,7 +67,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.FunctionalTests
                     var stream = await feature.UpgradeAsync();
 
                     var buffer = new byte[128];
-                    var read = await context.Request.Body.ReadAsync(buffer, 0, 128).TimeoutAfter(TestConstants.DefaultTimeout);
+                    var read = await context.Request.Body.ReadAsync(buffer, 0, 128).DefaultTimeout();
                     Assert.Equal(0, read);
 
                     using (var reader = new StreamReader(stream))
@@ -101,7 +101,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.FunctionalTests
                     await connection.Send(send + "\r\n");
                     await connection.Receive(recv);
 
-                    await upgrade.Task.TimeoutAfter(TestConstants.DefaultTimeout);
+                    await upgrade.Task.DefaultTimeout();
                 }
             }
         }
@@ -138,10 +138,10 @@ namespace Microsoft.AspNetCore.Server.Kestrel.FunctionalTests
                     $"Date: {server.Context.DateHeaderValue}",
                     "",
                     "");
-                await connection.WaitForConnectionClose().TimeoutAfter(TestConstants.DefaultTimeout);
+                await connection.WaitForConnectionClose().DefaultTimeout();
             }
 
-            var ex = await Assert.ThrowsAsync<InvalidOperationException>(async () => await upgradeTcs.Task.TimeoutAfter(TestConstants.DefaultTimeout));
+            var ex = await Assert.ThrowsAsync<InvalidOperationException>(async () => await upgradeTcs.Task.DefaultTimeout());
             Assert.Equal(CoreStrings.UpgradeCannotBeCalledMultipleTimes, ex.Message);
         }
 
@@ -243,7 +243,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.FunctionalTests
                 }
             }
 
-            var ex = await Assert.ThrowsAsync<InvalidOperationException>(async () => await upgradeTcs.Task).TimeoutAfter(TestConstants.DefaultTimeout);
+            var ex = await Assert.ThrowsAsync<InvalidOperationException>(async () => await upgradeTcs.Task).DefaultTimeout();
             Assert.Equal(CoreStrings.CannotUpgradeNonUpgradableRequest, ex.Message);
         }
 
@@ -331,7 +331,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.FunctionalTests
                         "");
                 }
 
-                await appCompletedTcs.Task.TimeoutAfter(TestConstants.DefaultTimeout);
+                await appCompletedTcs.Task.DefaultTimeout();
             }
         }
     }

test/Kestrel.Transport.Libuv.Tests/LibuvConnectionTests.cs

@@ -35,7 +35,8 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Transport.Libuv.Tests
                         Thread = thread
                     };
                     var socket = new MockSocket(mockLibuv, Thread.CurrentThread.ManagedThreadId, transportContext.Log);
-                    var connection = new LibuvConnection(listenerContext, socket);
+                    var connection = new LibuvConnection(socket, listenerContext.TransportContext.Log, thread, null, null);
+                    listenerContext.TransportContext.ConnectionDispatcher.OnConnection(connection);
                     _ = connection.Start();
 
                     mockLibuv.AllocCallback(socket.InternalGetHandle(), 2048, out var ignored);
@@ -85,7 +86,8 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Transport.Libuv.Tests
                         Thread = thread
                     };
                     var socket = new MockSocket(mockLibuv, Thread.CurrentThread.ManagedThreadId, transportContext.Log);
-                    var connection = new LibuvConnection(listenerContext, socket);
+                    var connection = new LibuvConnection(socket, listenerContext.TransportContext.Log, thread, null, null);
+                    listenerContext.TransportContext.ConnectionDispatcher.OnConnection(connection);
                     connectionTask = connection.Start();
 
                     mockLibuv.AllocCallback(socket.InternalGetHandle(), 2048, out var ignored);
@@ -100,7 +102,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Transport.Libuv.Tests
                 // Now complete the output writer so that the connection closes
                 mockConnectionDispatcher.Output.Writer.Complete();
 
-                await connectionTask.TimeoutAfter(TestConstants.DefaultTimeout);
+                await connectionTask.DefaultTimeout();
 
                 // Assert that we don't try to start reading
                 Assert.Null(mockLibuv.AllocCallback);
@@ -150,7 +152,8 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Transport.Libuv.Tests
                         Thread = thread
                     };
                     var socket = new MockSocket(mockLibuv, Thread.CurrentThread.ManagedThreadId, transportContext.Log);
-                    var connection = new LibuvConnection(listenerContext, socket);
+                    var connection = new LibuvConnection(socket, listenerContext.TransportContext.Log, thread, null, null);
+                    listenerContext.TransportContext.ConnectionDispatcher.OnConnection(connection);
                     connectionTask = connection.Start();
 
                     mockLibuv.AllocCallback(socket.InternalGetHandle(), 2048, out var ignored);
@@ -181,7 +184,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Transport.Libuv.Tests
                 // Now complete the output writer and wait for the connection to close
                 mockConnectionDispatcher.Output.Writer.Complete();
 
-                await connectionTask.TimeoutAfter(TestConstants.DefaultTimeout);
+                await connectionTask.DefaultTimeout();
 
                 // Assert that we don't try to start reading
                 Assert.Null(mockLibuv.AllocCallback);
@@ -212,7 +215,8 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Transport.Libuv.Tests
                         Thread = thread
                     };
                     var socket = new MockSocket(mockLibuv, Thread.CurrentThread.ManagedThreadId, transportContext.Log);
-                    var connection = new LibuvConnection(listenerContext, socket);
+                    var connection = new LibuvConnection(socket, listenerContext.TransportContext.Log, thread, null, null);
+                    listenerContext.TransportContext.ConnectionDispatcher.OnConnection(connection);
                     _ = connection.Start();
 
                     var ignored = new LibuvFunctions.uv_buf_t();

test/Kestrel.Transport.Libuv.Tests/LibuvOutputConsumerTests.cs

@@ -7,6 +7,7 @@ using System.Collections.Concurrent;
 using System.IO.Pipelines;
 using System.Threading;
 using System.Threading.Tasks;
+using Microsoft.AspNetCore.Connections;
 using Microsoft.AspNetCore.Connections.Features;
 using Microsoft.AspNetCore.Http.Features;
 using Microsoft.AspNetCore.Server.Kestrel.Core;
@@ -84,7 +85,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Transport.Libuv.Tests
                 var writeTask = outputProducer.WriteDataAsync(buffer);
 
                 // Assert
-                await writeTask.TimeoutAfter(TestConstants.DefaultTimeout);
+                await writeTask.DefaultTimeout();
             }
         }
 
@@ -122,7 +123,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Transport.Libuv.Tests
                 var writeTask = outputProducer.WriteDataAsync(buffer);
 
                 // Assert
-                await writeTask.TimeoutAfter(TestConstants.DefaultTimeout);
+                await writeTask.DefaultTimeout();
 
                 // Cleanup
                 outputProducer.Dispose();
@@ -181,7 +182,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Transport.Libuv.Tests
                 await _libuvThread.PostAsync(cb => cb(0), triggerNextCompleted);
 
                 // Assert
-                await writeTask.TimeoutAfter(TestConstants.DefaultTimeout);
+                await writeTask.DefaultTimeout();
 
                 // Cleanup
                 outputProducer.Dispose();
@@ -245,7 +246,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Transport.Libuv.Tests
                 await _libuvThread.PostAsync(cb => cb(0), triggerNextCompleted);
 
                 // Finishing the first write should allow the second write to pre-complete.
-                await writeTask2.TimeoutAfter(TestConstants.DefaultTimeout);
+                await writeTask2.DefaultTimeout();
 
                 // Cleanup
                 outputProducer.Dispose();
@@ -738,6 +739,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Transport.Libuv.Tests
             var http1Connection = new Http1Connection(new Http1ConnectionContext
             {
                 ServiceContext = serviceContext,
+                ConnectionContext = Mock.Of<ConnectionContext>(),
                 ConnectionFeatures = connectionFeatures,
                 MemoryPool = _memoryPool,
                 TimeoutControl = Mock.Of<ITimeoutControl>(),
@@ -764,12 +766,12 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Transport.Libuv.Tests
                 // Without ConfigureAwait(false), xunit will dispatch.
                 await consumer.WriteOutputAsync().ConfigureAwait(false);
 
-                http1Connection.Abort(error: null);
+                http1Connection.Abort(abortReason: null);
                 outputReader.Complete();
             }
             catch (UvException ex)
             {
-                http1Connection.Abort(ex);
+                http1Connection.Abort(new ConnectionAbortedException(ex.Message, ex));
                 outputReader.Complete(ex);
             }
         }

test/Kestrel.Transport.Libuv.Tests/TestHelpers/MockConnectionDispatcher.cs

@@ -4,9 +4,6 @@
 using System;
 using System.Buffers;
 using System.IO.Pipelines;
-using Microsoft.AspNetCore.Http.Features;
-using Microsoft.AspNetCore.Connections;
-using Microsoft.AspNetCore.Connections.Features;
 using Microsoft.AspNetCore.Server.Kestrel.Transport.Abstractions.Internal;
 
 namespace Microsoft.AspNetCore.Server.Kestrel.Transport.Libuv.Tests.TestHelpers

test/shared/TaskTimeoutExtensions.cs

@@ -0,0 +1,20 @@
+// 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 Microsoft.AspNetCore.Testing;
+
+namespace System.Threading.Tasks
+{
+    public static class TaskTimeoutExtensions
+    {
+        public static Task<T> DefaultTimeout<T>(this Task<T> task)
+        {
+            return task.TimeoutAfter(TestConstants.DefaultTimeout);
+        }
+
+        public static Task DefaultTimeout(this Task task)
+        {
+            return task.TimeoutAfter(TestConstants.DefaultTimeout);
+        }
+    }
+}

test/shared/TestConnection.cs

@@ -47,8 +47,6 @@ namespace Microsoft.AspNetCore.Testing
             _reader = new StreamReader(_stream, Encoding.ASCII);
         }
 
-        public Socket Socket => _socket;
-
         public Stream Stream => _stream;
 
         public StreamReader Reader => _reader;
@@ -215,6 +213,12 @@ namespace Microsoft.AspNetCore.Testing
             _socket.Shutdown(how);
         }
 
+        public void Reset()
+        {
+            _socket.LingerState = new LingerOption(true, 0);
+            _socket.Dispose();
+        }
+
         public Task WaitForConnectionClose()
         {
             var tcs = new TaskCompletionSource<object>(TaskCreationOptions.RunContinuationsAsynchronously);