diff --git a/src/clients/ioxide.pg/PgConnection.cs b/src/clients/ioxide.pg/PgConnection.cs
index 0abca19f..d4496e7d 100644
--- a/src/clients/ioxide.pg/PgConnection.cs
+++ b/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)
diff --git a/src/clients/ioxide.pg/PgPool.cs b/src/clients/ioxide.pg/PgPool.cs
index 7c108e28..5e022514 100644
--- a/src/clients/ioxide.pg/PgPool.cs
+++ b/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
}
}
}
diff --git a/tests/Ioxide.Tests.Harness/TcpSockets.cs b/tests/Ioxide.Tests.Harness/TcpSockets.cs
new file mode 100644
index 00000000..e1a55bdd
--- /dev/null
+++ b/tests/Ioxide.Tests.Harness/TcpSockets.cs
@@ -0,0 +1,101 @@
+namespace Ioxide.Tests;
+
+/// This process's TCP connections from /proc/net/tcp, which unlike a descriptor count shows a socket outliving its close().
+public static class TcpSockets
+{
+ private const string Established = "01"; // the "st" column
+
+ /// One IPv4 connection, named by both of its ends exactly as /proc/net/tcp prints them.
+ public readonly record struct Connection(string Local, string Remote);
+
+ /// This process's IPv4 connections that are ESTABLISHED to .
+ public static HashSet EstablishedTo(int remotePort)
+ {
+ HashSet mine = OwnSocketInodes();
+ var found = new HashSet();
+
+ 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;
+ }
+
+ /// Whether is still ESTABLISHED, whether or not a descriptor still holds it.
+ public static bool StillEstablished(Connection connection)
+ => Rows().Any(row => row[1] == connection.Local && row[2] == connection.Remote && row[3] == Established);
+
+ /// Waits for connections to beyond , and returns what it found.
+ public static HashSet WaitForNew(int remotePort, HashSet earlier, int count, int timeoutMs = 10_000)
+ {
+ long deadlineMs = Environment.TickCount64 + timeoutMs;
+
+ while (true)
+ {
+ HashSet fresh = EstablishedTo(remotePort);
+ fresh.ExceptWith(earlier);
+
+ if (fresh.Count >= count || Environment.TickCount64 > deadlineMs)
+ {
+ return fresh;
+ }
+
+ Thread.Sleep(20);
+ }
+ }
+
+ /// Waits until any of stops being ESTABLISHED, and returns the ones that have.
+ public static List WaitForAnyToEnd(IReadOnlyCollection connections, int timeoutMs)
+ {
+ long deadlineMs = Environment.TickCount64 + timeoutMs;
+
+ while (true)
+ {
+ List 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 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 OwnSocketInodes()
+ {
+ var inodes = new HashSet();
+
+ 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;
+ }
+}
diff --git a/tests/Ioxide.Tests.Pg/PgHandlers.cs b/tests/Ioxide.Tests.Pg/PgHandlers.cs
index 153e9d5b..114cf0a5 100644
--- a/tests/Ioxide.Tests.Pg/PgHandlers.cs
+++ b/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();
diff --git a/tests/Ioxide.Tests.Pg/PgTests.cs b/tests/Ioxide.Tests.Pg/PgTests.cs
index 772eb7a1..99c94502 100644
--- a/tests/Ioxide.Tests.Pg/PgTests.cs
+++ b/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 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 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 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()