Complete the transport pipes after connection middleware runs (#2735)

Author: davidfowl

src/Kestrel.Core/Internal/ConnectionDispatcher.cs

@@ -59,6 +59,12 @@ private async Task Execute(ConnectionContext connectionContext)
                 {
                     Log.LogCritical(0, ex, $"{nameof(ConnectionDispatcher)}.{nameof(Execute)}() {connectionContext.ConnectionId}");
                 }
+                finally
+                {
+                    // Complete the transport PipeReader and PipeWriter after calling into application code
+                    connectionContext.Transport.Input.Complete();
+                    connectionContext.Transport.Output.Complete();
+                }
             }
         }
 

test/Kestrel.Core.Tests/ConnectionDispatcherTests.cs

@@ -1,7 +1,9 @@
 // Copyright (c) .NET Foundation. All rights reserved.
 // Licensed under the Apache License, Version 2.0. See License.txt in the project root for license information.
 
+using System;
 using System.Collections.Generic;
+using System.IO.Pipelines;
 using System.Linq;
 using System.Threading.Tasks;
 using Microsoft.AspNetCore.Server.Kestrel.Core.Internal;
@@ -42,5 +44,27 @@ public void OnConnectionCreatesLogScopeWithConnectionId()
             // Verify the scope was disposed after request processing completed
             Assert.True(((TestKestrelTrace)serviceContext.Log).Logger.Scopes.IsEmpty);
         }
+
+        [Fact]
+        public async Task OnConnectionCompletesTransportPipesAfterReturning()
+        {
+            var serviceContext = new TestServiceContext();
+            var tcs = new TaskCompletionSource<object>();
+            var dispatcher = new ConnectionDispatcher(serviceContext, _ => Task.CompletedTask);
+
+            var mockConnection = new Mock<TransportConnection>();
+            var mockPipeReader = new Mock<PipeReader>();
+            var mockPipeWriter = new Mock<PipeWriter>();
+            var mockPipe = new Mock<IDuplexPipe>();
+            mockPipe.Setup(m => m.Input).Returns(mockPipeReader.Object);
+            mockPipe.Setup(m => m.Output).Returns(mockPipeWriter.Object);
+            mockConnection.Setup(m => m.Transport).Returns(mockPipe.Object);
+            var connection = mockConnection.Object;
+
+            await dispatcher.OnConnection(connection);
+
+            mockPipeWriter.Verify(m => m.Complete(It.IsAny<Exception>()), Times.Once());
+            mockPipeReader.Verify(m => m.Complete(It.IsAny<Exception>()), Times.Once());
+        }
     }
 }