Send the event socket's close code instead of aborting the socket

The events endpoint cancelled its pump token and only then wrote the close
frame. The token was also the one passed to socket.ReceiveAsync and
SendAsync, and cancelling a pending operation on the ManagedWebSocket that
Kestrel hands over aborts it. CloseAsync then saw an Aborted socket and
returned, so in production a client whose token expired got a TCP reset
rather than 4401 "reconnect with a fresh token", and a shutdown sent no
1001 either.

The tests missed it because TestServer's in-memory socket survives that
cancellation: the same scenario through TestServer closes with 4401.

PumpAsync now does what the relay's pump does. No WebSocket operation is
cancelled; the token stops only the outbound queue, the clock and the
access refresh. The sender is let finish its last write, the close goes
out, and the receiver ends on the client's answering close. Each wait is
bounded by the close timeout, and the socket is aborted only when a peer
stops answering.

ApiFixture gains ConnectEventsOverTcpAsync, which runs a second host on
real Kestrel over loopback, sharing the container and identity stub. The
new AnExpiringToken_ClosesTheSocketWith4401 fails with a reset before this
change and passes after it.

src/DodoSSH.Api/Features/Events/EventsEndpoint.cs

@@ -211,10 +211,23 @@ internal sealed class VaultEventsEndpoint(
 
     /// <summary>Runs the socket until something ends it, then closes it politely.</summary>
     /// <remarks>
+    /// <para>
     /// Three loops rather than one: reading a socket and writing to it are independent waits, and the
-    /// clock is a third. They are joined by <see cref="Task.WhenAny(Task[])"/> and then <em>all</em>
-    /// awaited before the close is written, because a close frame racing a notice frame is a protocol
+    /// clock is a third. They are joined by <see cref="Task.WhenAny(Task[])"/>, and the sender is
+    /// finished before the close is written, because a close frame racing a notice frame is a protocol
     /// violation that presents as a client dropping its connection for no visible reason.
+    /// </para>
+    /// <para>
+    /// <b>No operation on the WebSocket is ever cancelled.</b> Cancelling a pending receive or send on
+    /// the socket Kestrel hands over aborts it, and an aborted socket cannot send the close frame that
+    /// is the only way a client learns <em>why</em> it was dropped — 4401 "reconnect with a fresh
+    /// token" against 1001 "the server is going away" against a failure to back off from. This was
+    /// once the case here, and it went unnoticed because <c>TestServer</c>'s in-memory socket survives
+    /// the cancellation that a real one does not. So the token stops the queue and the clock; the
+    /// sender is let finish its last write; the close goes out; and the receiver ends on the client's
+    /// answering close. Each wait is bounded, and the socket is aborted only when a peer has stopped
+    /// answering altogether. The relay's pump does the same, for the same reason.
+    /// </para>
     /// </remarks>
     private async Task PumpAsync(
         WebSocket socket,
@@ -236,15 +249,21 @@ internal sealed class VaultEventsEndpoint(
             VaultCount: connection.VaultCount));
 
         var sending = SendAsync(socket, connection, pump.Token);
-        var receiving = ReceiveAsync(socket, connection, pump.Token);
+        var receiving = ReceiveAsync(socket, connection);
         var minding = MindAsync(connection, closure, pump.Token);
 
         await Task.WhenAny(sending, receiving, minding).ConfigureAwait(false);
 
         await pump.CancelAsync().ConfigureAwait(false);
 
-        // Nothing may still be mid-send when the close frame goes out.
-        await Task.WhenAll(Settled(sending), Settled(receiving), Settled(minding)).ConfigureAwait(false);
+        // Nothing may still be mid-send when the close frame goes out. A send that does not finish is
+        // a client that has stopped reading, which will not read a close frame either.
+        if (!await SettlesAsync(sending).ConfigureAwait(false))
+        {
+            socket.Abort();
+        }
+
+        await SettlesAsync(minding).ConfigureAwait(false);
 
         if (lifetime.ApplicationStopping.IsCancellationRequested)
         {
@@ -256,6 +275,14 @@ internal sealed class VaultEventsEndpoint(
         EventsLog.ClosingConnection(logger, connection.UserId, closure.Reason);
 
         await CloseAsync(socket, closure).ConfigureAwait(false);
+
+        // The client's answering close is what ends the receiver. Returning before it arrives would
+        // dispose the socket under a close frame still in flight.
+        if (!await SettlesAsync(receiving).ConfigureAwait(false))
+        {
+            socket.Abort();
+            await SettlesAsync(receiving).ConfigureAwait(false);
+        }
     }
 
     /// <summary>
@@ -287,8 +314,9 @@ internal sealed class VaultEventsEndpoint(
 
             var bytes = JsonSerializer.SerializeToUtf8Bytes(frame, DodoSshJsonContext.Default.VaultEvent);
 
+            // Not cancellable, deliberately. See PumpAsync.
             await socket
-                .SendAsync(bytes, WebSocketMessageType.Text, endOfMessage: true, cancellationToken)
+                .SendAsync(bytes, WebSocketMessageType.Text, endOfMessage: true, CancellationToken.None)
                 .ConfigureAwait(false);
         }
     }
@@ -297,21 +325,26 @@ internal sealed class VaultEventsEndpoint(
     /// Reads what the client sends, which in this version is heartbeats and a close.
     /// </summary>
     /// <remarks>
+    /// <para>
     /// A client cannot ask to follow a vault, and that is deliberate rather than unfinished: a
     /// <c>subscribe(vaultId)</c> frame is an existence oracle for vault ids, which is the disclosure
     /// <c>SyncPullEndpoint</c> answers 404 rather than 403 to avoid. Subscription is decided from the
     /// caller's access and nothing else.
+    /// </para>
+    /// <para>
+    /// It takes no token: it ends on the client's close, a frame it refuses, or the socket failing —
+    /// and when the server is the one ending the connection, the close it sends is what prompts the
+    /// first of those. See <see cref="PumpAsync"/>.
+    /// </para>
     /// </remarks>
-    private static async Task ReceiveAsync(
-        WebSocket socket,
-        VaultEventConnection connection,
-        CancellationToken cancellationToken)
+    private static async Task ReceiveAsync(WebSocket socket, VaultEventConnection connection)
     {
         var buffer = new byte[MaxInboundFrameBytes];
 
-        while (!cancellationToken.IsCancellationRequested)
+        while (true)
         {
-            var received = await socket.ReceiveAsync(buffer, cancellationToken).ConfigureAwait(false);
+            // Not cancellable, deliberately. See PumpAsync.
+            var received = await socket.ReceiveAsync(buffer, CancellationToken.None).ConfigureAwait(false);
 
             if (received.MessageType == WebSocketMessageType.Close)
             {
@@ -492,31 +525,39 @@ internal sealed class VaultEventsEndpoint(
     }
 
     /// <summary>
-    /// Awaits a pump loop, treating its cancellation and its socket faults as the ordinary end.
+    /// Waits for a pump loop to finish, for at most the close timeout.
     /// </summary>
+    /// <returns>Whether it finished.</returns>
     /// <remarks>
     /// Every one of these loops ends by being cancelled or by the socket going away, so an exception
     /// here is the expected shape of "this connection is over" rather than a fault to propagate — and
     /// propagating it would skip the close frame the other side is waiting for.
     /// </remarks>
-    private static async Task Settled(Task loop)
+    private static async Task<bool> SettlesAsync(Task loop)
     {
         try
         {
-            await loop.ConfigureAwait(false);
+            await loop.WaitAsync(CloseTimeout).ConfigureAwait(false);
+
+            return true;
+        }
+        catch (TimeoutException)
+        {
+            return false;
         }
         catch (Exception exception)
             when (exception is OperationCanceledException or WebSocketException or ObjectDisposedException)
         {
-            // Expected.
+            return true;
         }
     }
 
     /// <summary>Why the socket is being closed, decided by whichever loop ended first.</summary>
     /// <remarks>
-    /// Mutable and shared, and safe without a lock for one specific reason: it is written by the pump
-    /// loops and read only after <see cref="Task.WhenAll(Task[])"/> over all of them, which is a
-    /// memory barrier. Writes race only with each other, and any of them is a true answer.
+    /// Mutable and shared, and safe without a lock for one specific reason: it is written only by
+    /// <see cref="MindAsync"/>, and read only after that loop has been awaited, which is a memory
+    /// barrier. If that loop outlived the close timeout, a late write races the read, and either
+    /// answer is still a true one.
     /// </remarks>
     private sealed class Closure(int code, string reason)
     {

tests/DodoSSH.Api.Tests/ApiFixture.cs

@@ -1,9 +1,12 @@
+using System.Net;
 using System.Net.Http.Headers;
 using System.Net.WebSockets;
 using DodoSSH.Api.Features.Relay;
 using DodoSSH.Contracts;
 using DodoSSH.Infrastructure;
 using Microsoft.AspNetCore.Hosting;
+using Microsoft.AspNetCore.Hosting.Server;
+using Microsoft.AspNetCore.Hosting.Server.Features;
 using Microsoft.AspNetCore.Mvc.Testing;
 using Microsoft.AspNetCore.TestHost;
 using Microsoft.EntityFrameworkCore;
@@ -29,6 +32,13 @@ public sealed class ApiFixture : WebApplicationFactory<Program>, IAsyncLifetime
         .WithPassword("test")
         .Build();
 
+    private readonly Lazy<(WebApplicationFactory<Program> Host, Uri Address)> kestrel;
+
+    public ApiFixture()
+    {
+        kestrel = new Lazy<(WebApplicationFactory<Program>, Uri)>(StartKestrel);
+    }
+
     /// <summary>The stubbed identity provider.</summary>
     public StubIdentityProvider IdentityProvider { get; } = new();
 
@@ -45,6 +55,11 @@ public sealed class ApiFixture : WebApplicationFactory<Program>, IAsyncLifetime
     /// <inheritdoc />
     public override async ValueTask DisposeAsync()
     {
+        if (kestrel.IsValueCreated)
+        {
+            await kestrel.Value.Host.DisposeAsync();
+        }
+
         await base.DisposeAsync();
         await container.DisposeAsync();
         IdentityProvider.Dispose();
@@ -162,6 +177,64 @@ public sealed class ApiFixture : WebApplicationFactory<Program>, IAsyncLifetime
             new Uri(Server.BaseAddress, VaultEvents.Path.TrimStart('/')),
             cancellationToken);
     }
+
+    /// <summary>
+    /// Opens the event socket with the supplied raw token, over real Kestrel and a real TCP connection.
+    /// </summary>
+    /// <remarks>
+    /// <para>
+    /// For what <c>TestServer</c> cannot show: how the socket behaves when it ends. Kestrel hands the
+    /// endpoint a <c>ManagedWebSocket</c>, which aborts itself when a pending operation is cancelled,
+    /// and an aborted socket cannot send the close frame that tells a client why it was dropped. The
+    /// in-memory socket <c>TestServer</c> provides does not reproduce that, so a close-code test run
+    /// through it can pass against a server that never sends one.
+    /// </para>
+    /// <para>
+    /// A second host on the same database, started on first use. It listens on loopback only, for the
+    /// same Windows Firewall reason as <see cref="StubIdentityProvider"/>.
+    /// </para>
+    /// </remarks>
+    public async Task<WebSocket> ConnectEventsOverTcpAsync(string token, CancellationToken cancellationToken)
+    {
+        var socket = new ClientWebSocket();
+
+        try
+        {
+            socket.Options.AddSubProtocol(VaultEvents.SubProtocol);
+            socket.Options.SetRequestHeader("Authorization", $"Bearer {token}");
+
+            var address = new UriBuilder(new Uri(kestrel.Value.Address, VaultEvents.Path.TrimStart('/')))
+            {
+                Scheme = Uri.UriSchemeWs,
+            };
+
+            await socket.ConnectAsync(address.Uri, cancellationToken);
+
+            return socket;
+        }
+        catch
+        {
+            socket.Dispose();
+            throw;
+        }
+    }
+
+    private (WebApplicationFactory<Program> Host, Uri Address) StartKestrel()
+    {
+        // WithWebHostBuilder rather than a second fixture class, so the host inherits ConfigureWebHost
+        // above and therefore this container, this identity provider and this signing key.
+        var host = WithWebHostBuilder(_ => { });
+
+        host.UseKestrel(options => options.Listen(IPAddress.Loopback, 0));
+        host.StartServer();
+
+        var address = host.Services
+            .GetRequiredService<IServer>()
+            .Features.Get<IServerAddressesFeature>()!
+            .Addresses.Single();
+
+        return (host, new Uri(address));
+    }
 }
 
 /// <summary>Shares one host and container across every test class in the assembly.</summary>

tests/DodoSSH.Api.Tests/EventsEndpointTests.cs

@@ -304,6 +304,31 @@ public sealed class EventsEndpointTests(ApiFixture fixture)
         socket.State.ShouldBe(WebSocketState.Open);
     }
 
+    // ---- The close ----
+
+    [Fact]
+    public async Task AnExpiringToken_ClosesTheSocketWith4401()
+    {
+        // The code is the whole message: 4401 tells a client to reconnect at once with a fresh token,
+        // where anything else — or a connection that simply drops — tells it to back off. Over real
+        // Kestrel, because TestServer's in-memory socket survived the server cancelling its own pending
+        // receive, which is what hid a server that aborted the socket instead of closing it.
+        var (subject, _) = await SeedUserWithVaultAsync();
+
+        // Seconds, not minutes: the bearer handler's clock skew lets a token this close to expiry
+        // through the upgrade, and the socket's own deadline is the token's exp without skew.
+        var token = fixture.IdentityProvider.MintToken(
+            subject, expires: TimeProvider.System.GetUtcNow().UtcDateTime.AddSeconds(2));
+
+        using var timeout = Timeout();
+        using var socket = await fixture.ConnectEventsOverTcpAsync(token, timeout.Token);
+
+        var closed = await ReadCloseAsync(socket, timeout.Token);
+
+        ((int?)closed.CloseStatus).ShouldBe(VaultEvents.TokenExpiredCloseCode);
+        closed.CloseStatusDescription.ShouldNotBeNullOrEmpty();
+    }
+
     // ---- Helpers ----
 
     /// <summary>A token that gives up rather than letting a broken build hang the suite.</summary>
@@ -340,6 +365,31 @@ public sealed class EventsEndpointTests(ApiFixture fixture)
         return Encoding.UTF8.GetString(buffer, 0, received.Count);
     }
 
+    /// <summary>Reads past every frame to the server's close, and answers it.</summary>
+    /// <remarks>
+    /// A socket the server aborted rather than closed throws here instead of returning, which is the
+    /// failure this exists to catch.
+    /// </remarks>
+    private static async Task<WebSocketReceiveResult> ReadCloseAsync(
+        WebSocket socket,
+        CancellationToken cancellationToken)
+    {
+        var buffer = new byte[8 * 1024];
+
+        while (true)
+        {
+            var received = await socket.ReceiveAsync(buffer, cancellationToken);
+
+            if (received.MessageType == WebSocketMessageType.Close)
+            {
+                await socket.CloseOutputAsync(
+                    WebSocketCloseStatus.NormalClosure, string.Empty, cancellationToken);
+
+                return received;
+            }
+        }
+    }
+
     /// <summary>Reads past the frames a test does not care about — hello, and heartbeats.</summary>
     private static async Task<VaultEvent> ReadUntilAsync(
         WebSocket socket,