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);