pg: a timed-out connection is shut down, and freed only once its armed recv has returned

src/clients/ioxide.pg/PgConnection.cs

@@ -305,6 +305,11 @@ private async Task SenderLoopAsync()
             _sending = false;
             FailAll(ex);
         }
+
+        if (IsBroken)
+        {
+            Dispose();   // a Dispose that found this send in flight left the teardown to here
+        }
     }
 
     // Move [from, from+length) to the front of the send buffer. CopyTo is memmove-safe for the
@@ -387,6 +392,11 @@ private async Task ReaderLoopAsync()
             _reading = false;
             FailAll(ex is PgException p ? p : new PgException(ex.Message));
         }
+
+        if (IsBroken)
+        {
+            Dispose();   // a Dispose that found this recv in flight left the teardown to here
+        }
     }
 
     private void FailAll(Exception ex)
@@ -400,7 +410,7 @@ private void FailAll(Exception ex)
 
     // Reactor-thread (pool ticker): if the oldest in-flight command is older than the timeout, tear
     // the connection down - fail its waiters with a diagnostic error and mark it broken. The pool then
-    // disposes it, and closing the fd cancels the stuck ring recv/send. Returns true if it timed out.
+    // disposes it, and shutting the socket down completes the stuck ring recv/send. Returns true if it timed out.
     internal bool CheckTimeout(long nowMs, int timeoutMs, string host, ushort port)
     {
         if (timeoutMs <= 0 || _inflight.Count == 0)
@@ -677,6 +687,15 @@ private PgException ReadServerError(in Message message)
 
     public unsafe void Dispose()
     {
+        IsBroken = true;
+
+        // close() would leave an armed op holding the socket open and using the buffers freed below.
+        if (_reading || _sending)
+        {
+            Native.shutdown(_socket.Fd, Native.SHUT_RDWR);
+            return;
+        }
+
         _socket.Dispose();
 
         if (_send != 0)

src/clients/ioxide.pg/PgPool.cs

@@ -181,7 +181,7 @@ private void Sweep()
                 {
                     PgConnection c = _connections[i];
                     _connections.RemoveAt(i);
-                    c.Dispose();   // closes the fd - unsticks the connection's stuck send/recv loops
+                    c.Dispose();   // shuts the socket down - unsticks the connection's stuck send/recv loops
                 }
             }
         }

tests/Ioxide.Tests.Harness/TcpSockets.cs

@@ -0,0 +1,101 @@
+namespace Ioxide.Tests;
+
+/// <summary>This process's TCP connections from /proc/net/tcp, which unlike a descriptor count shows a socket outliving its close().</summary>
+public static class TcpSockets
+{
+    private const string Established = "01";   // the "st" column
+
+    /// <summary>One IPv4 connection, named by both of its ends exactly as /proc/net/tcp prints them.</summary>
+    public readonly record struct Connection(string Local, string Remote);
+
+    /// <summary>This process's IPv4 connections that are ESTABLISHED to <paramref name="remotePort"/>.</summary>
+    public static HashSet<Connection> EstablishedTo(int remotePort)
+    {
+        HashSet<string> mine = OwnSocketInodes();
+        var found = new HashSet<Connection>();
+
+        foreach (string[] row in Rows())
+        {
+            if (row[3] == Established && PortOf(row[2]) == remotePort && mine.Contains(row[9]))
+            {
+                found.Add(new Connection(row[1], row[2]));
+            }
+        }
+
+        return found;
+    }
+
+    /// <summary>Whether <paramref name="connection"/> is still ESTABLISHED, whether or not a descriptor still holds it.</summary>
+    public static bool StillEstablished(Connection connection)
+        => Rows().Any(row => row[1] == connection.Local && row[2] == connection.Remote && row[3] == Established);
+
+    /// <summary>Waits for <paramref name="count"/> connections to <paramref name="remotePort"/> beyond <paramref name="earlier"/>, and returns what it found.</summary>
+    public static HashSet<Connection> WaitForNew(int remotePort, HashSet<Connection> earlier, int count, int timeoutMs = 10_000)
+    {
+        long deadlineMs = Environment.TickCount64 + timeoutMs;
+
+        while (true)
+        {
+            HashSet<Connection> fresh = EstablishedTo(remotePort);
+            fresh.ExceptWith(earlier);
+
+            if (fresh.Count >= count || Environment.TickCount64 > deadlineMs)
+            {
+                return fresh;
+            }
+
+            Thread.Sleep(20);
+        }
+    }
+
+    /// <summary>Waits until any of <paramref name="connections"/> stops being ESTABLISHED, and returns the ones that have.</summary>
+    public static List<Connection> WaitForAnyToEnd(IReadOnlyCollection<Connection> connections, int timeoutMs)
+    {
+        long deadlineMs = Environment.TickCount64 + timeoutMs;
+
+        while (true)
+        {
+            List<Connection> ended = connections.Where(c => !StillEstablished(c)).ToList();
+
+            if (ended.Count > 0 || Environment.TickCount64 > deadlineMs)
+            {
+                return ended;
+            }
+
+            Thread.Sleep(20);
+        }
+    }
+
+    // Columns: sl, local_address, rem_address, st, tx:rx queues, tr:when, retrnsmt, uid, timeout, inode.
+    private static IEnumerable<string[]> Rows()
+        => File.ReadAllLines("/proc/net/tcp")
+            .Skip(1)
+            .Select(line => line.Split(' ', StringSplitOptions.RemoveEmptyEntries))
+            .Where(row => row.Length > 9);
+
+    private static int PortOf(string endpoint) => Convert.ToInt32(endpoint[(endpoint.IndexOf(':') + 1)..], 16);
+
+    // A socket descriptor links to "socket:[inode]", the number in the inode column.
+    private static HashSet<string> OwnSocketInodes()
+    {
+        var inodes = new HashSet<string>();
+
+        foreach (string fd in Directory.EnumerateFileSystemEntries("/proc/self/fd"))
+        {
+            try
+            {
+                string? target = new FileInfo(fd).LinkTarget;
+                if (target is not null && target.StartsWith("socket:[", StringComparison.Ordinal))
+                {
+                    inodes.Add(target["socket:[".Length..^1]);
+                }
+            }
+            catch (IOException)
+            {
+                // Closed between the listing and the read.
+            }
+        }
+
+        return inodes;
+    }
+}

tests/Ioxide.Tests.Pg/PgHandlers.cs

@@ -35,6 +35,11 @@ public static async Task Pg(Reactor r, TcpConnection conn)
                         await pool.QueryAsync("SELECT * FROM no_such_table");
                         Wire.Write(conn, 200, "unreachable");
                     }
+                    else if (path.StartsWith("/sleep/", StringComparison.Ordinal) && int.TryParse(path["/sleep/".Length..], out int seconds))
+                    {
+                        await pool.QueryAsync($"SELECT pg_sleep({seconds})");
+                        Wire.Write(conn, 200, "slept");
+                    }
                     else
                     {
                         PgResult result = await pool.QueryAsync("SELECT 42");
@@ -43,7 +48,7 @@ public static async Task Pg(Reactor r, TcpConnection conn)
                 }
                 catch (PgException e)
                 {
-                    Wire.Write(conn, 500, e.SqlState ?? "error");
+                    Wire.Write(conn, 500, e.SqlState ?? e.Message);
                 }
 
                 await conn.FlushAsync();

tests/Ioxide.Tests.Pg/PgTests.cs

@@ -59,6 +59,36 @@ public static void Register(Runner runner, (string Host, int Port) pg, bool pgUp
             (int status, _) = Client.Get(port, "/slow", timeoutMs: 8000);
             Assert.Equal(503, status);
         }, skip: !pgUp);
+
+        // A socket outliving its close() is the visible half of a recv still armed into the freed buffer.
+        runner.Test("pg: a timed-out connection's socket ends at the timeout, not when the late reply comes", () =>
+        {
+            HashSet<TcpSockets.Connection> earlier = TcpSockets.EstablishedTo(pg.Port);
+            int port = TestServer.Start(PgHandlers.Pg, r => PgPool.Start(r, PgOpts(pg, commandTimeoutMs: 1000)));
+
+            (int okStatus, string okBody) = Client.Get(port, "/");
+            Assert.Equal(200, okStatus);
+            Assert.Equal("42", okBody);
+
+            HashSet<TcpSockets.Connection> pool = TcpSockets.WaitForNew(pg.Port, earlier, count: 2);
+            Assert.True(pool.Count == 2, $"expected the pool's 2 connections ESTABLISHED to :{pg.Port}, found {pool.Count}");
+
+            // The sleep outlasts the wait below, so its reply cannot be what ends the socket.
+            (int status, string body) = Client.Get(port, "/sleep/30", timeoutMs: 15_000);
+            Assert.Equal(500, status);
+            Assert.True(body.Contains("timed out"), $"expected the command timeout, got [{body}]");
+
+            List<TcpSockets.Connection> ended = TcpSockets.WaitForAnyToEnd(pool, timeoutMs: 10_000);
+            Assert.True(ended.Count == 1,
+                $"{ended.Count} of the pool's 2 connections ended within 10 s of the timeout; the timed-out one is still ESTABLISHED");
+
+            // Control: the connection that did not time out is untouched, and the pool still answers.
+            Assert.True(pool.Where(c => !ended.Contains(c)).All(TcpSockets.StillEstablished),
+                "the connection that did not time out ended too");
+            (int afterStatus, string afterBody) = Client.Get(port, "/");
+            Assert.Equal(200, afterStatus);
+            Assert.Equal("42", afterBody);
+        }, skip: !pgUp);
     }
 
     private static PgOptions PgOpts((string Host, int Port) pg, int commandTimeoutMs = 30_000) => new()