diff --git a/src/clients/ioxide.pg/PgPool.cs b/src/clients/ioxide.pg/PgPool.cs index 5e022514..01008387 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(); // shuts the socket down - 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 2c17ea76..1221dd12 100644 --- a/tests/Ioxide.Tests.Pg/Program.cs +++ b/tests/Ioxide.Tests.Pg/Program.cs @@ -18,6 +18,7 @@ private static int Main() bool up = Sidecars.Reachable(pg.Host, pg.Port); Console.WriteLine($"postgres {(up ? "up" : "down")} ({pg.Host}:{pg.Port})\n"); + PgSweepTests.Register(runner, pg, up); PgTests.Register(runner, pg, up); PgPreparedTests.Register(runner, pg, up);