Dropping a PoolConnection whose connection is silently dead leaks the pool permit forever (return_to_pool cleanup never completes)

[rslowinski]

Bug Description

When a PoolConnection<Postgres> is dropped while its underlying TCP connection is silently dead — the peer stopped responding but never sent RST/FIN, e.g. a NAT/firewall expired the flow, or the common advice of wrapping a slow query in tokio::time::timeout fired — the pool permit is leaked permanently.

PoolConnection::drop spawns return_to_pool() (sqlx-core/src/pool/connection.rs, Drop impl). That cleanup path performs I/O against the connection with no time bound. Against a silently-dead socket the write side succeeds (bytes are ACKed) and the read side waits forever, so the spawned task never finishes and the DecrementSizeGuard it holds never releases the semaphore permit or decrements the pool size. Each such drop permanently costs one pool permit, one tokio task, and one socket. Once it has happened max_connections times, every subsequent acquire() fails with PoolTimedOut and the pool never recovers (only a process restart helps).

This composes badly with the usual recommendation to bound queries with tokio::time::timeout(...) (#3060): the natural cleanup after a timed-out query is dropping the connection, which is exactly the leaking path. Connection::close() cannot help either — it awaits the same dead socket. It is also the failure mode that TCP keepalives (#3540) would normally mitigate at the OS level, but sqlx currently exposes no keepalive configuration.

Minimal Reproduction

Self-contained — no real PostgreSQL needed. A fake server completes the startup handshake, then plays dead while still reading (ACKing) client bytes, which is exactly the observable state of a connection whose flow a middlebox silently dropped:

[dependencies]
sqlx = { version = "0.8.6", default-features = false, features = ["postgres", "runtime-tokio"] }
tokio = { version = "1", features = ["full"] }

//! Reproduction: dropping a `PoolConnection` whose underlying connection is
//! silently dead (peer stopped responding without RST/FIN, e.g. a NAT dropped
//! the flow) permanently leaks the pool permit: the spawned return-to-pool
//! cleanup blocks forever on the dead socket. Enough such drops exhaust the
//! pool; every later `acquire()` fails with `PoolTimedOut`.

use std::net::SocketAddr;
use std::sync::Arc;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::time::Duration;
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::TcpListener;

/// Minimal fake PostgreSQL server: completes the startup handshake, then goes
/// silent while still reading (ACKing) client bytes — the state a real
/// connection is in after a middlebox silently drops the flow.
async fn spawn_silent_postgres() -> (SocketAddr, Arc<AtomicUsize>) {
    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;
            };
            let n = accepted.fetch_add(1, Ordering::SeqCst) + 1;
            println!("    [server] accepted physical connection #{n}");

            tokio::spawn(async move {
                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;
                }

                let mut reply: Vec<u8> = Vec::new();
                reply.extend([b'R', 0, 0, 0, 8, 0, 0, 0, 0]); // AuthenticationOk
                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]); // BackendKeyData
                reply.extend(1234u32.to_be_bytes());
                reply.extend(5678u32.to_be_bytes());
                reply.extend([b'Z', 0, 0, 0, 5, b'I']); // ReadyForQuery
                if socket.write_all(&reply).await.is_err() {
                    return;
                }

                // Play dead: drain client bytes forever, never respond.
                let mut buf = [0u8; 4096];
                while socket.read(&mut buf).await.map(|n| n > 0).unwrap_or(false) {}
                println!("    [server] a connection was closed by the client");
            });
        }
    });

    (addr, connections)
}

#[tokio::main]
async fn main() {
    let (addr, connections) = spawn_silent_postgres().await;
    let url = format!(
        "postgres://user@{}:{}/db?sslmode=disable",
        addr.ip(),
        addr.port()
    );
    let pool = sqlx::postgres::PgPoolOptions::new()
        .max_connections(2)
        .acquire_timeout(Duration::from_secs(3))
        .connect_lazy(&url)
        .expect("lazy pool");

    for attempt in 1..=3u32 {
        println!(
            "attempt {attempt}: pool size={} idle={}",
            pool.size(),
            pool.num_idle()
        );
        let t0 = std::time::Instant::now();
        match pool.acquire().await {
            Ok(mut conn) => {
                println!("  acquired after {:?}", t0.elapsed());
                match tokio::time::timeout(
                    Duration::from_millis(500),
                    sqlx::query("SELECT 1").execute(&mut *conn),
                )
                .await
                {
                    Ok(res) => println!("  query completed: {res:?}"),
                    Err(_) => {
                        println!("  query timed out after 500ms; dropping the PoolConnection");
                        drop(conn); // the 'natural' cleanup — this is what leaks
                    }
                }
            }
            Err(e) => println!("  acquire FAILED after {:?}: {e}", t0.elapsed()),
        }
        tokio::time::sleep(Duration::from_millis(500)).await;
    }

    println!(
        "final: pool size={} idle={} | physical connections opened: {}",
        pool.size(),
        pool.num_idle(),
        connections.load(Ordering::SeqCst)
    );
    println!("(with detach() instead of drop, attempt 3 acquires a fresh connection instead)");
}

Output on 0.8.6:

attempt 1: pool size=0 idle=0
    [server] accepted physical connection #1
  acquired after 688.042µs
  query timed out after 500ms; dropping the PoolConnection
attempt 2: pool size=1 idle=0
    [server] accepted physical connection #2
  acquired after 525.375µs
  query timed out after 500ms; dropping the PoolConnection
attempt 3: pool size=2 idle=0
  acquire FAILED after 3.002197958s: pool timed out while waiting for an open connection
final: pool size=2 idle=0 | physical connections opened: 2

Note the fake server logs zero client disconnects: both sockets are still open, parked inside the spawned return-to-pool tasks, and pool.size() stays at 2 with idle=0 forever.

Expected Behavior

Either:

• (a) the return-to-pool cleanup is bounded (e.g. by acquire_timeout or a dedicated release timeout), closing the connection hard and releasing the permit when the health-check I/O doesn't complete in time; or
• (b) documentation states that a connection suspected dead must be discarded via PoolConnection::detach() — which releases the permit synchronously and closes the socket without any awaits. I verified detach() + drop works reliably as a workaround (each subsequent acquire opens a fresh physical connection and the pool stays healthy).

Happy to submit a PR bounding the release path if maintainers agree on the approach.

Workaround

// instead of: drop(conn)
let raw = conn.detach();
drop(raw); // synchronous socket close, permit already released

Info

• SQLx version: 0.8.6
• SQLx features enabled: postgres, runtime-tokio
• Database server and version: reproduces against the mock above; originally observed against PostgreSQL 15/16 behind NAT
• Operating system: macOS 15 (arm64) and Linux (x86_64/arm64) both reproduce
• Rust version: rustc 1.96.0 (ac68faa20 2026-05-25)