From 92efb7611ffc803f29de69aa5659a09e036004e2 Mon Sep 17 00:00:00 2001 From: Diogo Martins Date: Wed, 7 Oct 2026 21:40:28 +0100 Subject: [PATCH 1/2] pg, redis: the timeout sweep removes the connection that timed out, not whatever its index holds afterwards Sweep failed a timed-out connection's waiters through CheckTimeout and then removed and disposed _connections[i]. The waiters resume inline, so a handler that queries again picks from the pool inside the sweep, and Pick evicts the timed-out connection, which is broken, shifting the list. Sweep then disposed whatever had moved into index i: a live connection holding the queries just placed on it, which failed with "send failed: errno 9". When the timed-out connection was the last entry, the index was past the end and the ticker faulted, skipping the rest of that sweep. Sweep now holds the connection it checked, removes it by reference (a no-op if a Pick already did), and clamps the index to the list's current length. No new state. New tests, PgSweepTests and RedisSweepTests: a pool of 2 with a 1000 ms command timeout, a pg_sleep(30) or BLPOP for 30 s on index 0, and a handler that issues two more commands when the timeout fails it. The handler records that the timed-out connection is still listed when it runs, which shows it ran inline in the sweep. Confirmed both ways: on f298af0 both commands fail with errno 9 (pg and redis), and pass with the fix. With the sleep moved to index 1, f298af0 logs "ticker faulted: Index was out of range" and the fix does not. --- src/clients/ioxide.pg/PgPool.cs | 8 +- src/clients/ioxide.redis/RedisPool.cs | 8 +- tests/Ioxide.Tests.Pg/Ioxide.Tests.Pg.csproj | 1 + tests/Ioxide.Tests.Pg/PgSweepTests.cs | 100 ++++++++++++++++++ tests/Ioxide.Tests.Pg/Program.cs | 1 + .../Ioxide.Tests.Redis.csproj | 1 + tests/Ioxide.Tests.Redis/Program.cs | 1 + tests/Ioxide.Tests.Redis/RedisSweepTests.cs | 98 +++++++++++++++++ 8 files changed, 212 insertions(+), 6 deletions(-) create mode 100644 tests/Ioxide.Tests.Pg/PgSweepTests.cs create mode 100644 tests/Ioxide.Tests.Redis/RedisSweepTests.cs diff --git a/src/clients/ioxide.pg/PgPool.cs b/src/clients/ioxide.pg/PgPool.cs index 4590755d..e0ce921b 100644 --- a/src/clients/ioxide.pg/PgPool.cs +++ b/src/clients/ioxide.pg/PgPool.cs @@ -177,11 +177,13 @@ private void Sweep() { for (int i = _connections.Count - 1; i >= 0; i--) { - if (_connections[i].CheckTimeout(now, timeoutMs, _options.Host, _options.Port)) + PgConnection c = _connections[i]; + if (c.CheckTimeout(now, timeoutMs, _options.Host, _options.Port)) { - PgConnection c = _connections[i]; - _connections.RemoveAt(i); + // Its failed waiters resumed inline, and a Pick of theirs may have removed it or others. + _connections.Remove(c); c.Dispose(); // closes the fd - unsticks the connection's stuck send/recv loops + i = Math.Min(i, _connections.Count); } } } diff --git a/src/clients/ioxide.redis/RedisPool.cs b/src/clients/ioxide.redis/RedisPool.cs index 3748c3ad..a07b97ac 100644 --- a/src/clients/ioxide.redis/RedisPool.cs +++ b/src/clients/ioxide.redis/RedisPool.cs @@ -181,11 +181,13 @@ private void Sweep() { for (int i = _connections.Count - 1; i >= 0; i--) { - if (_connections[i].CheckTimeout(now, timeoutMs, _options.Host, _options.Port)) + RedisConnection c = _connections[i]; + if (c.CheckTimeout(now, timeoutMs, _options.Host, _options.Port)) { - RedisConnection c = _connections[i]; - _connections.RemoveAt(i); + // Its failed waiters resumed inline, and a Pick of theirs may have removed it or others. + _connections.Remove(c); c.Dispose(); // closes the fd - unsticks the connection's stuck send/recv loops + i = Math.Min(i, _connections.Count); } } } diff --git a/tests/Ioxide.Tests.Pg/Ioxide.Tests.Pg.csproj b/tests/Ioxide.Tests.Pg/Ioxide.Tests.Pg.csproj index ab66308e..aa394e5f 100644 --- a/tests/Ioxide.Tests.Pg/Ioxide.Tests.Pg.csproj +++ b/tests/Ioxide.Tests.Pg/Ioxide.Tests.Pg.csproj @@ -13,6 +13,7 @@ + diff --git a/tests/Ioxide.Tests.Pg/PgSweepTests.cs b/tests/Ioxide.Tests.Pg/PgSweepTests.cs new file mode 100644 index 00000000..e131bace --- /dev/null +++ b/tests/Ioxide.Tests.Pg/PgSweepTests.cs @@ -0,0 +1,100 @@ +using ioxide; +using ioxide.pg; +using ioxide.timer; + +namespace Ioxide.Tests; + +/// The pool's timeout sweep, when the waiters it fails query the pool again from inside it. +internal static class PgSweepTests +{ + public static void Register(Runner runner, (string Host, int Port) pg, bool pgUp) + { + // The failed waiter resumes inside the sweep, and its picks evict the timed-out connection from the list the sweep is indexing. + runner.Test("pg: queries made by a timed-out query's handler run on the live connection the sweep keeps", () => + { + int port = TestServer.Start(RequeryOnTimeout, r => PgPool.Start(r, Options(pg))); + + (int status, string body) = Client.Get(port, "/timeout", timeoutMs: 15_000); + Assert.Equal(200, status); + Assert.Equal("listed=2 first=1 second=2", body); + + (int afterStatus, string afterBody) = Client.Get(port, "/select"); + Assert.Equal(200, afterStatus); + Assert.Equal("42", afterBody); + }, skip: !pgUp); + } + + private static async Task RequeryOnTimeout(Reactor r, TcpConnection conn) + { + PgPool pool = r.GetService(); + + try + { + RecvSnapshot snapshot = await conn.ReadAsync(); + string path = Wire.ReadPath(conn, snapshot); + + string body; + if (path == "/timeout") + { + // Both open before the first pick, so the sleep lands on index 0 with a live connection after it. + var timer = new RingTimer(r); + while (pool.ConnectionCount < 2) + { + await timer.DelayAsync(10); + } + + try + { + await pool.QueryAsync("SELECT pg_sleep(30)"); + body = "the sleep returned"; + } + catch (PgException e) when (e.Message.Contains("timed out")) + { + // Still listed here only if this runs inline in the sweep, before it removes the connection. + int listed = pool.ConnectionCount; + ValueTask first = pool.QueryAsync("SELECT 1"); + ValueTask second = pool.QueryAsync("SELECT 2"); + body = $"listed={listed} first={await Outcome(first)} second={await Outcome(second)}"; + } + } + else if (path == "/select") + { + body = (await pool.QueryAsync("SELECT 42")).Value ?? ""; + } + else + { + return; // the harness's listen probe: it must not pick, or the sleep moves off index 0 + } + + Wire.Write(conn, 200, body); + await conn.FlushAsync(); + } + finally + { + conn.DecRef(); + } + } + + private static async Task Outcome(ValueTask query) + { + try + { + return (await query).Value ?? "null"; + } + catch (PgException e) + { + return e.Message; + } + } + + private static PgOptions Options((string Host, int Port) pg) => new() + { + Host = pg.Host, + Port = (ushort)pg.Port, + User = Environment.GetEnvironmentVariable("EXAMPLES_PG_USER") ?? "bench", + Database = Environment.GetEnvironmentVariable("EXAMPLES_PG_DB") ?? "bench", + Password = Environment.GetEnvironmentVariable("EXAMPLES_PG_PASSWORD"), + PoolSize = 2, + CommandTimeoutMs = 1000, + }; +} diff --git a/tests/Ioxide.Tests.Pg/Program.cs b/tests/Ioxide.Tests.Pg/Program.cs index 18e089c4..c76cadc3 100644 --- a/tests/Ioxide.Tests.Pg/Program.cs +++ b/tests/Ioxide.Tests.Pg/Program.cs @@ -20,6 +20,7 @@ private static int Main() Console.WriteLine($"postgres {(up ? "up" : "down")} ({pg.Host}:{pg.Port})\n"); PgTests.Register(runner, pg, up); + PgSweepTests.Register(runner, pg, up); return runner.Summary(); } } diff --git a/tests/Ioxide.Tests.Redis/Ioxide.Tests.Redis.csproj b/tests/Ioxide.Tests.Redis/Ioxide.Tests.Redis.csproj index 6bc9a7ce..8954775e 100644 --- a/tests/Ioxide.Tests.Redis/Ioxide.Tests.Redis.csproj +++ b/tests/Ioxide.Tests.Redis/Ioxide.Tests.Redis.csproj @@ -13,6 +13,7 @@ + diff --git a/tests/Ioxide.Tests.Redis/Program.cs b/tests/Ioxide.Tests.Redis/Program.cs index 727bcdcb..3e4acb5e 100644 --- a/tests/Ioxide.Tests.Redis/Program.cs +++ b/tests/Ioxide.Tests.Redis/Program.cs @@ -18,6 +18,7 @@ private static int Main() Console.WriteLine($"redis {(up ? "up" : "down")} ({redis.Host}:{redis.Port})\n"); RedisTests.Register(runner, redis, up); + RedisSweepTests.Register(runner, redis, up); return runner.Summary(); } } diff --git a/tests/Ioxide.Tests.Redis/RedisSweepTests.cs b/tests/Ioxide.Tests.Redis/RedisSweepTests.cs new file mode 100644 index 00000000..8933005a --- /dev/null +++ b/tests/Ioxide.Tests.Redis/RedisSweepTests.cs @@ -0,0 +1,98 @@ +using ioxide; +using ioxide.redis; +using ioxide.timer; + +namespace Ioxide.Tests; + +/// The pool's timeout sweep, when the waiters it fails command the pool again from inside it. +internal static class RedisSweepTests +{ + public static void Register(Runner runner, (string Host, int Port) redis, bool redisUp) + { + // The failed waiter resumes inside the sweep, and its picks evict the timed-out connection from the list the sweep is indexing. + runner.Test("redis: commands made by a timed-out command's handler run on the live connection the sweep keeps", () => + { + int port = TestServer.Start(RecommandOnTimeout, r => RedisPool.Start(r, Options(redis))); + + (int status, string body) = Client.Get(port, "/timeout", timeoutMs: 15_000); + Assert.Equal(200, status); + Assert.Equal("listed=2 first=PONG second=PONG", body); + + (int afterStatus, string afterBody) = Client.Get(port, "/ping"); + Assert.Equal(200, afterStatus); + Assert.Equal("PONG", afterBody); + }, skip: !redisUp); + } + + private static async Task RecommandOnTimeout(Reactor r, TcpConnection conn) + { + RedisPool pool = r.GetService(); + + try + { + RecvSnapshot snapshot = await conn.ReadAsync(); + string path = Wire.ReadPath(conn, snapshot); + + string body; + if (path == "/timeout") + { + // Both open before the first pick, so the block lands on index 0 with a live connection after it. + var timer = new RingTimer(r); + while (pool.ConnectionCount < 2) + { + await timer.DelayAsync(10); + } + + try + { + await pool.ExecuteAsync("BLPOP", "sweep:never-pushed", 30); + body = "the block returned"; + } + catch (RedisException e) when (e.Message.Contains("timed out")) + { + // Still listed here only if this runs inline in the sweep, before it removes the connection. + int listed = pool.ConnectionCount; + ValueTask first = pool.ExecuteAsync("PING"); + ValueTask second = pool.ExecuteAsync("PING"); + body = $"listed={listed} first={await Outcome(first)} second={await Outcome(second)}"; + } + } + else if (path == "/ping") + { + body = (await pool.ExecuteAsync("PING")).AsString() ?? ""; + } + else + { + return; // the harness's listen probe: it must not pick, or the block moves off index 0 + } + + Wire.Write(conn, 200, body); + await conn.FlushAsync(); + } + finally + { + conn.DecRef(); + } + } + + private static async Task Outcome(ValueTask command) + { + try + { + return (await command).AsString() ?? "null"; + } + catch (RedisException e) + { + return e.Message; + } + } + + private static RedisOptions Options((string Host, int Port) redis) => new() + { + Host = redis.Host, + Port = (ushort)redis.Port, + Password = Environment.GetEnvironmentVariable("EXAMPLES_REDIS_PASSWORD"), + PoolSize = 2, + CommandTimeoutMs = 1000, + }; +} From 8358daab7109006d87d41fe43470bbe86d546f36 Mon Sep 17 00:00:00 2001 From: Diogo Martins Date: Mon, 12 Oct 2026 15:20:24 +0100 Subject: [PATCH 2/2] pg-only: the Redis half goes with ioxide.redis (#337) --- src/clients/ioxide.redis/RedisPool.cs | 8 +- .../Ioxide.Tests.Redis.csproj | 1 - tests/Ioxide.Tests.Redis/Program.cs | 1 - tests/Ioxide.Tests.Redis/RedisSweepTests.cs | 98 ------------------- 4 files changed, 3 insertions(+), 105 deletions(-) delete mode 100644 tests/Ioxide.Tests.Redis/RedisSweepTests.cs diff --git a/src/clients/ioxide.redis/RedisPool.cs b/src/clients/ioxide.redis/RedisPool.cs index 92f4c465..a7f27bd6 100644 --- a/src/clients/ioxide.redis/RedisPool.cs +++ b/src/clients/ioxide.redis/RedisPool.cs @@ -181,13 +181,11 @@ private void Sweep() { for (int i = _connections.Count - 1; i >= 0; i--) { - RedisConnection c = _connections[i]; - if (c.CheckTimeout(now, timeoutMs, _options.Host, _options.Port)) + if (_connections[i].CheckTimeout(now, timeoutMs, _options.Host, _options.Port)) { - // Its failed waiters resumed inline, and a Pick of theirs may have removed it or others. - _connections.Remove(c); + RedisConnection c = _connections[i]; + _connections.RemoveAt(i); c.Dispose(); // closes the fd - unsticks the connection's stuck send/recv loops - i = Math.Min(i, _connections.Count); } } } diff --git a/tests/Ioxide.Tests.Redis/Ioxide.Tests.Redis.csproj b/tests/Ioxide.Tests.Redis/Ioxide.Tests.Redis.csproj index 8954775e..6bc9a7ce 100644 --- a/tests/Ioxide.Tests.Redis/Ioxide.Tests.Redis.csproj +++ b/tests/Ioxide.Tests.Redis/Ioxide.Tests.Redis.csproj @@ -13,7 +13,6 @@ - diff --git a/tests/Ioxide.Tests.Redis/Program.cs b/tests/Ioxide.Tests.Redis/Program.cs index 3e4acb5e..727bcdcb 100644 --- a/tests/Ioxide.Tests.Redis/Program.cs +++ b/tests/Ioxide.Tests.Redis/Program.cs @@ -18,7 +18,6 @@ private static int Main() Console.WriteLine($"redis {(up ? "up" : "down")} ({redis.Host}:{redis.Port})\n"); RedisTests.Register(runner, redis, up); - RedisSweepTests.Register(runner, redis, up); return runner.Summary(); } } diff --git a/tests/Ioxide.Tests.Redis/RedisSweepTests.cs b/tests/Ioxide.Tests.Redis/RedisSweepTests.cs deleted file mode 100644 index 8933005a..00000000 --- a/tests/Ioxide.Tests.Redis/RedisSweepTests.cs +++ /dev/null @@ -1,98 +0,0 @@ -using ioxide; -using ioxide.redis; -using ioxide.timer; - -namespace Ioxide.Tests; - -/// The pool's timeout sweep, when the waiters it fails command the pool again from inside it. -internal static class RedisSweepTests -{ - public static void Register(Runner runner, (string Host, int Port) redis, bool redisUp) - { - // The failed waiter resumes inside the sweep, and its picks evict the timed-out connection from the list the sweep is indexing. - runner.Test("redis: commands made by a timed-out command's handler run on the live connection the sweep keeps", () => - { - int port = TestServer.Start(RecommandOnTimeout, r => RedisPool.Start(r, Options(redis))); - - (int status, string body) = Client.Get(port, "/timeout", timeoutMs: 15_000); - Assert.Equal(200, status); - Assert.Equal("listed=2 first=PONG second=PONG", body); - - (int afterStatus, string afterBody) = Client.Get(port, "/ping"); - Assert.Equal(200, afterStatus); - Assert.Equal("PONG", afterBody); - }, skip: !redisUp); - } - - private static async Task RecommandOnTimeout(Reactor r, TcpConnection conn) - { - RedisPool pool = r.GetService(); - - try - { - RecvSnapshot snapshot = await conn.ReadAsync(); - string path = Wire.ReadPath(conn, snapshot); - - string body; - if (path == "/timeout") - { - // Both open before the first pick, so the block lands on index 0 with a live connection after it. - var timer = new RingTimer(r); - while (pool.ConnectionCount < 2) - { - await timer.DelayAsync(10); - } - - try - { - await pool.ExecuteAsync("BLPOP", "sweep:never-pushed", 30); - body = "the block returned"; - } - catch (RedisException e) when (e.Message.Contains("timed out")) - { - // Still listed here only if this runs inline in the sweep, before it removes the connection. - int listed = pool.ConnectionCount; - ValueTask first = pool.ExecuteAsync("PING"); - ValueTask second = pool.ExecuteAsync("PING"); - body = $"listed={listed} first={await Outcome(first)} second={await Outcome(second)}"; - } - } - else if (path == "/ping") - { - body = (await pool.ExecuteAsync("PING")).AsString() ?? ""; - } - else - { - return; // the harness's listen probe: it must not pick, or the block moves off index 0 - } - - Wire.Write(conn, 200, body); - await conn.FlushAsync(); - } - finally - { - conn.DecRef(); - } - } - - private static async Task Outcome(ValueTask command) - { - try - { - return (await command).AsString() ?? "null"; - } - catch (RedisException e) - { - return e.Message; - } - } - - private static RedisOptions Options((string Host, int Port) redis) => new() - { - Host = redis.Host, - Port = (ushort)redis.Port, - Password = Environment.GetEnvironmentVariable("EXAMPLES_REDIS_PASSWORD"), - PoolSize = 2, - CommandTimeoutMs = 1000, - }; -}