From 449f729c1eabd621cdc26545745fb7f6cda64790 Mon Sep 17 00:00:00 2001 From: rslowinski Date: Thu, 23 Jul 2026 13:53:40 +0200 Subject: [PATCH] fix(pool): bound the on-release ping so a dead connection can't leak its permit MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit When a PoolConnection is dropped, the pool spawns a task that pings the connection before returning it to the idle queue. If the peer has gone away silently (no RST/FIN — e.g. a NAT/firewall dropped the flow), that ping is sent successfully and then waits forever for a response that never arrives. Because the ping runs inside the drop-spawned task, the unbounded wait strands that task's DecrementSizeGuard, permanently leaking the connection's pool permit. Once this has happened max_connections times, every subsequent acquire() fails with PoolTimedOut and the pool never recovers. Bound the on-release ping with a timeout, mirroring the existing close-on-drop path (CLOSE_ON_DROP_TIMEOUT). On timeout the connection is discarded via close_hard() (a local socket shutdown that cannot block), which releases the permit so the pool can open a replacement. Adds a regression test that drives a real PgPool against an in-process fake server which completes the handshake then goes silent; it needs no DATABASE_URL and runs on the tokio runtime cells. --- Cargo.toml | 7 ++ sqlx-core/src/pool/connection.rs | 51 +++++++++--- tests/postgres/pool.rs | 138 +++++++++++++++++++++++++++++++ 3 files changed, 183 insertions(+), 13 deletions(-) create mode 100644 tests/postgres/pool.rs diff --git a/Cargo.toml b/Cargo.toml index b7ed7ac2cb..d540c55ab8 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -449,6 +449,13 @@ name = "postgres-error" path = "tests/postgres/error.rs" required-features = ["postgres"] +# Uses a tokio fake server + `#[tokio::test]`, so it only runs on the tokio +# runtime; skipped under the other runtime matrix cells. +[[test]] +name = "postgres-pool" +path = "tests/postgres/pool.rs" +required-features = ["postgres", "runtime-tokio"] + [[test]] name = "postgres-test-attr" path = "tests/postgres/test-attr.rs" diff --git a/sqlx-core/src/pool/connection.rs b/sqlx-core/src/pool/connection.rs index 7912b12aa1..754f7e3580 100644 --- a/sqlx-core/src/pool/connection.rs +++ b/sqlx-core/src/pool/connection.rs @@ -15,6 +15,12 @@ use crate::pool::options::PoolConnectionMetadata; const CLOSE_ON_DROP_TIMEOUT: Duration = Duration::from_secs(5); +/// Upper bound on the on-release `ping()` used to test a connection before it is +/// returned to the pool. This runs in a drop-spawned task, so it must never block +/// indefinitely (e.g. on a silently-dropped TCP flow) or it would leak the pool +/// permit held by that task. +const RETURN_TO_POOL_PING_TIMEOUT: Duration = Duration::from_secs(5); + /// A connection managed by a [`Pool`][crate::pool::Pool]. /// /// Will be returned to the pool on-drop. @@ -311,19 +317,38 @@ impl Floating> { // returned to the pool; also of course, if it was dropped due to an error // this is simply a band-aid as SQLx-next connections should be able // to recover from cancellations - if let Err(error) = self.raw.ping().await { - tracing::warn!( - %error, - "error occurred while testing the connection on-release", - ); - - // Connection is broken, don't try to gracefully close. - self.close_hard().await; - false - } else { - // if the connection is still viable, release it to the pool - self.release(); - true + // + // The ping is bounded by a timeout: this runs in the drop-spawned + // return-to-pool task, and on a connection whose peer has silently gone + // away (e.g. a NAT/firewall dropped the flow) `ping()` can send its + // request successfully and then wait forever for a response that never + // comes. An unbounded wait here strands this task's `DecrementSizeGuard`, + // leaking the pool permit permanently; enough such leaks exhaust the pool. + // Follow the `close_on_drop` path (see `take_and_close`) and cap the wait. + match crate::rt::timeout(RETURN_TO_POOL_PING_TIMEOUT, self.raw.ping()).await { + Ok(Ok(())) => { + // if the connection is still viable, release it to the pool + self.release(); + true + } + Ok(Err(error)) => { + tracing::warn!( + %error, + "error occurred while testing the connection on-release", + ); + + // Connection is broken, don't try to gracefully close. + self.close_hard().await; + false + } + Err(_) => { + tracing::warn!("timed out testing the connection on-release; discarding it"); + + // Connection is unresponsive; discard it. `close_hard` only shuts + // the socket down locally, so it does not risk blocking as well. + self.close_hard().await; + false + } } } diff --git a/tests/postgres/pool.rs b/tests/postgres/pool.rs new file mode 100644 index 0000000000..2a9203d4bc --- /dev/null +++ b/tests/postgres/pool.rs @@ -0,0 +1,138 @@ +//! Pool behavior tests that do not require a live database: they drive a real +//! `PgPool` against an in-process fake server, so they run anywhere `cargo test` +//! runs without a `DATABASE_URL`. + +use std::net::SocketAddr; +use std::sync::atomic::{AtomicUsize, Ordering}; +use std::sync::Arc; +use std::time::Duration; + +use sqlx::postgres::PgPoolOptions; +use tokio::io::{AsyncReadExt, AsyncWriteExt}; +use tokio::net::TcpListener; + +/// Spawn a fake PostgreSQL server that completes the startup handshake and then +/// goes silent: it keeps reading (and thus ACKing) whatever the client sends but +/// never sends another byte back. This is the observable state of a connection +/// whose peer has silently gone away — e.g. a NAT/firewall that dropped the flow +/// without sending RST/FIN. Any request the client makes (including the pool's +/// on-release `ping`) will be written successfully and then awaited forever. +/// +/// Returns the bound address and a counter of accepted physical connections. +async fn spawn_silent_postgres() -> (SocketAddr, Arc) { + let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind"); + let addr = listener.local_addr().expect("local_addr"); + let connections = Arc::new(AtomicUsize::new(0)); + let accepted = connections.clone(); + + tokio::spawn(async move { + loop { + let Ok((mut socket, _)) = listener.accept().await else { + break; + }; + accepted.fetch_add(1, Ordering::SeqCst); + + tokio::spawn(async move { + // Read the client's StartupMessage (length-prefixed, no type byte). + let mut len_buf = [0u8; 4]; + if socket.read_exact(&mut len_buf).await.is_err() { + return; + } + let len = (u32::from_be_bytes(len_buf) as usize).saturating_sub(4); + let mut body = vec![0u8; len]; + if socket.read_exact(&mut body).await.is_err() { + return; + } + + // Minimal successful handshake: AuthenticationOk, a few + // ParameterStatus messages the client reads during connect, + // BackendKeyData, then ReadyForQuery(idle). + let mut reply: Vec = Vec::new(); + reply.extend([b'R', 0, 0, 0, 8, 0, 0, 0, 0]); + for (key, value) in [ + ("server_version", "14.0"), + ("client_encoding", "UTF8"), + ("DateStyle", "ISO, MDY"), + ] { + let payload_len = 4 + key.len() + 1 + value.len() + 1; + reply.push(b'S'); + reply.extend((payload_len as u32).to_be_bytes()); + reply.extend(key.as_bytes()); + reply.push(0); + reply.extend(value.as_bytes()); + reply.push(0); + } + reply.extend([b'K', 0, 0, 0, 12]); + reply.extend(1234u32.to_be_bytes()); + reply.extend(5678u32.to_be_bytes()); + reply.extend([b'Z', 0, 0, 0, 5, b'I']); + if socket.write_all(&reply).await.is_err() { + return; + } + + // Handshake complete. Play dead: drain client bytes forever + // (so writes keep succeeding) but never respond. + let mut buf = [0u8; 4096]; + while socket.read(&mut buf).await.map(|n| n > 0).unwrap_or(false) {} + }); + } + }); + + (addr, connections) +} + +/// Regression test: a `PoolConnection` whose peer has silently gone away must not +/// strand its permit when dropped. +/// +/// On drop the pool spawns a task that `ping`s the connection before returning it. +/// Against the silent server that ping is sent but never answered. If the ping is +/// unbounded, the spawned task parks forever holding the connection's permit, so a +/// `max_connections(1)` pool can never hand out another connection and every later +/// `acquire()` fails with `PoolTimedOut`. With the on-release ping bounded, the +/// permit is released and a fresh connection can be opened. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn pool_recovers_permit_when_connection_unresponsive_on_release() { + // Overall guard so a regression hangs this one test instead of wedging CI. + tokio::time::timeout(Duration::from_secs(60), async { + let (addr, connections) = spawn_silent_postgres().await; + let url = format!( + "postgres://user@{}:{}/db?sslmode=disable", + addr.ip(), + addr.port() + ); + + // acquire_timeout must exceed the on-release ping bound (5s) so that, + // with the fix, the freed permit is observable before this deadline; + // without the fix, acquire genuinely times out here. + let pool = PgPoolOptions::new() + .max_connections(1) + .acquire_timeout(Duration::from_secs(20)) + .connect_lazy(&url) + .expect("build pool"); + + // Opens physical connection #1 (fresh connections are not ping-tested). + let conn = pool + .acquire() + .await + .expect("first acquire opens a connection"); + // Drop triggers the spawned on-release ping, which the silent server + // never answers. + drop(conn); + + // With the bounded ping the permit is released after the connection is + // found unresponsive, so this opens a fresh physical connection. + let conn2 = pool + .acquire() + .await + .expect("permit must be released after the bounded on-release ping"); + drop(conn2); + + assert_eq!( + connections.load(Ordering::SeqCst), + 2, + "the unresponsive connection must be discarded and a fresh one opened" + ); + }) + .await + .expect("test hung: on-release connection test is not bounded"); +}