From 2bb9983755b38151debb5a90ce1c8380ef536c4d Mon Sep 17 00:00:00 2001 From: Diogo Martins Date: Wed, 7 Oct 2026 19:49:35 +0100 Subject: [PATCH 1/2] pg, redis: a timed-out connection is shut down, and freed only once its armed recv has returned A command timeout disposed the connection while its reader was parked on a ring recv: close(fd), then the buffers freed. close() does not end a socket an io_uring op still holds, so the socket stayed ESTABLISHED and the recv stayed armed until the server finally answered, and the kernel then wrote that reply into the freed receive buffer - which the allocator had already handed to the replacement connection (same address, same fd number). Pick's dispose of a connection broken by one loop could hit the other loop's op the same way. Dispose now shuts the socket down while a loop is parked on an op, which completes it, and the loop that exits last closes the fd and frees the buffers. Dispose marks the connection broken, so it takes no new commands in between. "pg:" and "redis: a timed-out connection's socket ends at the timeout, not when the late reply comes" time a command out on the sidecar (pg_sleep(30), BLPOP 30) and require the pool connection that timed out to leave ESTABLISHED within 10 s, while its sibling stays ESTABLISHED and the pool still answers. Confirmed both ways: both fail on the unmodified source (still ESTABLISHED after 10 s) and pass with the fix. --- src/clients/ioxide.pg/PgConnection.cs | 21 +++- src/clients/ioxide.pg/PgPool.cs | 2 +- src/clients/ioxide.redis/RedisConnection.cs | 21 +++- src/clients/ioxide.redis/RedisPool.cs | 2 +- tests/Ioxide.Tests.Harness/TcpSockets.cs | 101 ++++++++++++++++++++ tests/Ioxide.Tests.Pg/PgHandlers.cs | 7 +- tests/Ioxide.Tests.Pg/PgTests.cs | 30 ++++++ tests/Ioxide.Tests.Redis/RedisHandlers.cs | 12 +++ tests/Ioxide.Tests.Redis/RedisTests.cs | 33 ++++++- 9 files changed, 223 insertions(+), 6 deletions(-) create mode 100644 tests/Ioxide.Tests.Harness/TcpSockets.cs diff --git a/src/clients/ioxide.pg/PgConnection.cs b/src/clients/ioxide.pg/PgConnection.cs index ce5f3dfa..2f771c5f 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 4590755d..e3facd27 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/src/clients/ioxide.redis/RedisConnection.cs b/src/clients/ioxide.redis/RedisConnection.cs index 34799ea3..8d92378c 100644 --- a/src/clients/ioxide.redis/RedisConnection.cs +++ b/src/clients/ioxide.redis/RedisConnection.cs @@ -211,6 +211,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 + } } private async Task ReaderLoopAsync() @@ -230,6 +235,11 @@ private async Task ReaderLoopAsync() _reading = false; FailAll(ex); } + + if (IsBroken) + { + Dispose(); // a Dispose that found this recv in flight left the teardown to here + } } private void FailAll(Exception ex) @@ -243,7 +253,7 @@ private void FailAll(Exception ex) // Reactor-thread (pool ticker): tear the connection down if its oldest in-flight command is // overdue - fail waiters with a diagnostic error and mark it broken; the pool disposes it, and - // closing the fd cancels the stuck ring recv/send. Returns true if it timed out. + // 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) @@ -356,6 +366,15 @@ private unsafe void EnsureRecvSpace() 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.redis/RedisPool.cs b/src/clients/ioxide.redis/RedisPool.cs index 3748c3ad..47e82f65 100644 --- a/src/clients/ioxide.redis/RedisPool.cs +++ b/src/clients/ioxide.redis/RedisPool.cs @@ -185,7 +185,7 @@ private void Sweep() { RedisConnection 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() diff --git a/tests/Ioxide.Tests.Redis/RedisHandlers.cs b/tests/Ioxide.Tests.Redis/RedisHandlers.cs index 81cc93d4..f63d50e1 100644 --- a/tests/Ioxide.Tests.Redis/RedisHandlers.cs +++ b/tests/Ioxide.Tests.Redis/RedisHandlers.cs @@ -31,6 +31,18 @@ public static async Task Redis(Reactor r, TcpConnection conn) new RedisCommand("GET", "e2e:p")); Wire.Write(conn, 200, replies[2].AsString() ?? ""); } + else if (path.StartsWith("/block/", StringComparison.Ordinal) && int.TryParse(path["/block/".Length..], out int seconds)) + { + try + { + await pool.ExecuteAsync("BLPOP", "e2e:never-pushed", seconds); // nothing pushes it: blocks the full time + Wire.Write(conn, 200, "unblocked"); + } + catch (RedisException e) + { + Wire.Write(conn, 500, e.Message); + } + } else { await pool.SetExAsync("e2e:k", "hello", 60); diff --git a/tests/Ioxide.Tests.Redis/RedisTests.cs b/tests/Ioxide.Tests.Redis/RedisTests.cs index bd686e23..ad73e835 100644 --- a/tests/Ioxide.Tests.Redis/RedisTests.cs +++ b/tests/Ioxide.Tests.Redis/RedisTests.cs @@ -30,13 +30,44 @@ public static void Register(Runner runner, (string Host, int Port) redis, bool r Assert.Equal(200, status); Assert.Equal("2", body); }, skip: !redisUp); + + // A socket outliving its close() is the visible half of a recv still armed into the freed buffer. + runner.Test("redis: a timed-out connection's socket ends at the timeout, not when the late reply comes", () => + { + HashSet earlier = TcpSockets.EstablishedTo(redis.Port); + int port = TestServer.Start(RedisHandlers.Redis, r => RedisPool.Start(r, RedisOpts(redis, commandTimeoutMs: 1000))); + + (int okStatus, string okBody) = Client.Get(port, "/"); + Assert.Equal(200, okStatus); + Assert.Equal("hello", okBody); + + HashSet pool = TcpSockets.WaitForNew(redis.Port, earlier, count: 2); + Assert.True(pool.Count == 2, $"expected the pool's 2 connections ESTABLISHED to :{redis.Port}, found {pool.Count}"); + + // The block outlasts the wait below, so its reply cannot be what ends the socket. + (int status, string body) = Client.Get(port, "/block/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("hello", afterBody); + }, skip: !redisUp); } - private static RedisOptions RedisOpts((string Host, int Port) redis) => new() + private static RedisOptions RedisOpts((string Host, int Port) redis, int commandTimeoutMs = 30_000) => new() { Host = redis.Host, Port = (ushort)redis.Port, Password = Environment.GetEnvironmentVariable("EXAMPLES_REDIS_PASSWORD"), PoolSize = 2, + CommandTimeoutMs = commandTimeoutMs, }; } From 7543293a5d174db93d39cf55752b50af50665ab8 Mon Sep 17 00:00:00 2001 From: Diogo Martins Date: Mon, 12 Oct 2026 15:19:50 +0100 Subject: [PATCH 2/2] pg-only: the Redis half goes with ioxide.redis (#337) --- src/clients/ioxide.redis/RedisConnection.cs | 21 +------------ src/clients/ioxide.redis/RedisPool.cs | 2 +- tests/Ioxide.Tests.Redis/RedisHandlers.cs | 12 -------- tests/Ioxide.Tests.Redis/RedisTests.cs | 33 +-------------------- 4 files changed, 3 insertions(+), 65 deletions(-) diff --git a/src/clients/ioxide.redis/RedisConnection.cs b/src/clients/ioxide.redis/RedisConnection.cs index 5f7bab80..62a80acc 100644 --- a/src/clients/ioxide.redis/RedisConnection.cs +++ b/src/clients/ioxide.redis/RedisConnection.cs @@ -211,11 +211,6 @@ private async Task SenderLoopAsync() _sending = false; FailAll(ex); } - - if (IsBroken) - { - Dispose(); // a Dispose that found this send in flight left the teardown to here - } } private async Task ReaderLoopAsync() @@ -235,11 +230,6 @@ private async Task ReaderLoopAsync() _reading = false; FailAll(ex); } - - if (IsBroken) - { - Dispose(); // a Dispose that found this recv in flight left the teardown to here - } } private void FailAll(Exception ex) @@ -253,7 +243,7 @@ private void FailAll(Exception ex) // Reactor-thread (pool ticker): tear the connection down if its oldest in-flight command is // overdue - fail waiters with a diagnostic error and mark it broken; the pool disposes it, and - // shutting the socket down completes the stuck ring recv/send. Returns true if it timed out. + // closing the fd cancels 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) @@ -366,15 +356,6 @@ private unsafe void EnsureRecvSpace() 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.redis/RedisPool.cs b/src/clients/ioxide.redis/RedisPool.cs index ac63d1e4..a7f27bd6 100644 --- a/src/clients/ioxide.redis/RedisPool.cs +++ b/src/clients/ioxide.redis/RedisPool.cs @@ -185,7 +185,7 @@ private void Sweep() { RedisConnection c = _connections[i]; _connections.RemoveAt(i); - c.Dispose(); // shuts the socket down - unsticks the connection's stuck send/recv loops + c.Dispose(); // closes the fd - unsticks the connection's stuck send/recv loops } } } diff --git a/tests/Ioxide.Tests.Redis/RedisHandlers.cs b/tests/Ioxide.Tests.Redis/RedisHandlers.cs index f63d50e1..81cc93d4 100644 --- a/tests/Ioxide.Tests.Redis/RedisHandlers.cs +++ b/tests/Ioxide.Tests.Redis/RedisHandlers.cs @@ -31,18 +31,6 @@ public static async Task Redis(Reactor r, TcpConnection conn) new RedisCommand("GET", "e2e:p")); Wire.Write(conn, 200, replies[2].AsString() ?? ""); } - else if (path.StartsWith("/block/", StringComparison.Ordinal) && int.TryParse(path["/block/".Length..], out int seconds)) - { - try - { - await pool.ExecuteAsync("BLPOP", "e2e:never-pushed", seconds); // nothing pushes it: blocks the full time - Wire.Write(conn, 200, "unblocked"); - } - catch (RedisException e) - { - Wire.Write(conn, 500, e.Message); - } - } else { await pool.SetExAsync("e2e:k", "hello", 60); diff --git a/tests/Ioxide.Tests.Redis/RedisTests.cs b/tests/Ioxide.Tests.Redis/RedisTests.cs index ad73e835..bd686e23 100644 --- a/tests/Ioxide.Tests.Redis/RedisTests.cs +++ b/tests/Ioxide.Tests.Redis/RedisTests.cs @@ -30,44 +30,13 @@ public static void Register(Runner runner, (string Host, int Port) redis, bool r Assert.Equal(200, status); Assert.Equal("2", body); }, skip: !redisUp); - - // A socket outliving its close() is the visible half of a recv still armed into the freed buffer. - runner.Test("redis: a timed-out connection's socket ends at the timeout, not when the late reply comes", () => - { - HashSet earlier = TcpSockets.EstablishedTo(redis.Port); - int port = TestServer.Start(RedisHandlers.Redis, r => RedisPool.Start(r, RedisOpts(redis, commandTimeoutMs: 1000))); - - (int okStatus, string okBody) = Client.Get(port, "/"); - Assert.Equal(200, okStatus); - Assert.Equal("hello", okBody); - - HashSet pool = TcpSockets.WaitForNew(redis.Port, earlier, count: 2); - Assert.True(pool.Count == 2, $"expected the pool's 2 connections ESTABLISHED to :{redis.Port}, found {pool.Count}"); - - // The block outlasts the wait below, so its reply cannot be what ends the socket. - (int status, string body) = Client.Get(port, "/block/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("hello", afterBody); - }, skip: !redisUp); } - private static RedisOptions RedisOpts((string Host, int Port) redis, int commandTimeoutMs = 30_000) => new() + private static RedisOptions RedisOpts((string Host, int Port) redis) => new() { Host = redis.Host, Port = (ushort)redis.Port, Password = Environment.GetEnvironmentVariable("EXAMPLES_REDIS_PASSWORD"), PoolSize = 2, - CommandTimeoutMs = commandTimeoutMs, }; }