[Named pipes] Disconnect + delayed client read test cases (dotnet#46166)

Co-authored-by: James Newton-King <[email protected]>

Author: simonferquel

src/Servers/Kestrel/Transport.NamedPipes/src/Internal/NamedPipeConnection.cs

@@ -24,9 +24,10 @@ internal sealed class NamedPipeConnection : TransportConnection, IConnectionName
     private bool _connectionDisposed;
     private Exception? _shutdownReason;
     private readonly object _shutdownLock = new object();
-    
-    private Task _receivingTask = Task.CompletedTask;
-    private Task _sendingTask = Task.CompletedTask;
+
+    // Internal for testing.
+    internal Task _receivingTask = Task.CompletedTask;
+    internal Task _sendingTask = Task.CompletedTask;
 
     public NamedPipeConnection(
         NamedPipeServerStream stream,

src/Servers/Kestrel/Transport.NamedPipes/test/NamedPipeConnectionTests.cs

@@ -4,6 +4,7 @@
 using System.Text;
 using Microsoft.AspNetCore.Connections;
 using Microsoft.AspNetCore.Internal;
+using Microsoft.AspNetCore.Server.Kestrel.Transport.NamedPipes.Internal;
 using Microsoft.AspNetCore.Testing;
 
 namespace Microsoft.AspNetCore.Server.Kestrel.Transport.NamedPipes.Tests;
@@ -39,7 +40,7 @@ public async Task InputReadAsync_ServerAborted_ThrowError()
         var clientStream = NamedPipeTestHelpers.CreateClientStream(connectionListener.EndPoint);
         await clientStream.ConnectAsync().DefaultTimeout();
         await clientStream.WriteAsync(TestData).DefaultTimeout();
-        
+
         var serverConnection = await connectionListener.AcceptAsync().DefaultTimeout();
         var readResult = await serverConnection.Transport.Input.ReadAtLeastAsync(TestData.Length).DefaultTimeout();
         serverConnection.Transport.Input.AdvanceTo(readResult.Buffer.End);
@@ -103,4 +104,70 @@ public async Task OutputWriteAsync_ServerAborted_ThrowError()
         // Complete writing.
         await serverConnection.Transport.Output.CompleteAsync();
     }
+
+    [ConditionalFact]
+    [OSSkipCondition(OperatingSystems.Linux | OperatingSystems.MacOSX, SkipReason = "Non-OS implementations use UDS with different transport behavior.")]
+    public async Task CompleteTransport_BeforeClientHasReadLastBytes_DontLoseData()
+    {
+        var options = new NamedPipeTransportOptions();
+        await using var connectionListener = await NamedPipeTestHelpers.CreateConnectionListenerFactory(LoggerFactory, options: options);
+
+        var clientStream = NamedPipeTestHelpers.CreateClientStream(connectionListener.EndPoint);
+        var connectTask = clientStream.ConnectAsync();
+
+        var connection = (NamedPipeConnection)await connectionListener.AcceptAsync();
+        await connection.Transport.Output.WriteAsync(new byte[] { 41 });
+        byte[] readBuffer = new byte[1];
+        var readAmount = await clientStream.ReadAsync(readBuffer);
+        Assert.Equal(1, readAmount);
+        Assert.Equal(41, readBuffer[0]);
+        await connection.Transport.Output.WriteAsync(new byte[] { 42 });
+
+        connection.Transport.Output.Complete();
+
+        // Wait for a short amount of time after completing transport before asserting the server is waiting for the client to finish reading.
+        // Actual stream disconnection should happen after the last client read happens.
+        await Task.Delay(100);
+        Assert.False(connection._sendingTask.IsCompleted, "Sending isn't complete until the client reads data.");
+
+        readAmount = await clientStream.ReadAsync(readBuffer);
+        Assert.Equal(1, readAmount);
+        Assert.Equal(42, readBuffer[0]);
+
+        await connection._sendingTask.DefaultTimeout();
+    }
+
+    [ConditionalFact]
+    [OSSkipCondition(OperatingSystems.Linux | OperatingSystems.MacOSX, SkipReason = "Non-OS implementations use UDS with different transport behavior.")]
+    public async Task DisposeAsync_BeforeClientHasReadLastBytes_DontLoseData()
+    {
+        var options = new NamedPipeTransportOptions();
+        await using var connectionListener = await NamedPipeTestHelpers.CreateConnectionListenerFactory(LoggerFactory, options: options);
+
+        var clientStream = NamedPipeTestHelpers.CreateClientStream(connectionListener.EndPoint);
+        var connectTask = clientStream.ConnectAsync();
+
+        var connection = (NamedPipeConnection)await connectionListener.AcceptAsync();
+        await connection.Transport.Output.WriteAsync(new byte[] { 41 });
+        byte[] readBuffer = new byte[1];
+        var readAmount = await clientStream.ReadAsync(readBuffer);
+        Assert.Equal(1, readAmount);
+        Assert.Equal(41, readBuffer[0]);
+        await connection.Transport.Output.WriteAsync(new byte[] { 42 });
+
+        var disposeTask = connection.DisposeAsync();
+
+        // Wait for a short amount of time after completing transport before asserting the server is waiting for the client to finish reading.
+        // Actual stream disconnection should happen after the last client read happens.
+        await Task.Delay(100);
+        Assert.False(connection._sendingTask.IsCompleted, "Sending isn't complete until the client reads data.");
+        Assert.False(disposeTask.IsCompleted, "Sending isn't complete until the client reads data.");
+
+        readAmount = await clientStream.ReadAsync(readBuffer);
+        Assert.Equal(1, readAmount);
+        Assert.Equal(42, readBuffer[0]);
+
+        await connection._sendingTask.DefaultTimeout();
+        await disposeTask.DefaultTimeout();
+    }
 }