From 3904a00c5a2fdc02787e01078451398873107789 Mon Sep 17 00:00:00 2001 From: John Mortlock <10041761+jmortlock@users.noreply.github.com> Date: Wed, 16 Sep 2026 12:37:33 +0930 Subject: [PATCH] fix(postgres): end a transaction block left open on a pooled connection `Floating::return_to_pool` validates a connection on release with `Connection::ping`, and the Postgres impl was a bare `wait_until_ready`: it drains the `ReadyForQuery` but never looks at its transaction-status byte. A session sitting inside a transaction block is therefore reported healthy and handed to the next borrower, whose statements run inside it and hold its locks. Two shapes reach that point with a client-side `transaction_depth` of zero, so neither drop guard has queued a `ROLLBACK`: - A future cancelled while `BEGIN`'s round trip is in flight (#4393). #4394 fixed this one at the source by claiming the depth before the await. - A statement that fails inside a block opened by a multi-statement `raw_sql`, where the block is opened and aborted within a single simple-query message and the depth is never raised at all. This one does not self-heal: the next borrower's `BEGIN` fails because the block is already aborted, leaving the depth at zero again, so every subsequent checkout fails with 25P02 until `max_lifetime` recycles the connection. Check the server-reported status in `ping` and roll back when it disagrees. The status byte is already decoded into `PgConnection::transaction_status` on every `ReadyForQuery`, so the check itself is free, and the `ROLLBACK` is only sent on a connection that is actually dirty. The check is gated on `transaction_depth == 0`. `ping` is public API and a caller may be holding a transaction deliberately; a non-zero depth means the client knows about the block and owns its lifetime, so it is left alone. Only a block the client has no record of is ended here -- which also means there is never a savepoint to restore to, hence a plain `ROLLBACK`. Also exposes `PgConnection::transaction_status`, so pool users can make the same distinction themselves. `Connection::is_in_transaction` reports the client-side depth, which is precisely the value that is wrong in both shapes above, and the server's view was not reachable from outside the crate. Tests: `it_does_not_return_a_connection_inside_a_transaction_to_the_pool` covers both shapes plus the leaked `BEGIN READ ONLY` case, where the damage surfaces as a failing write rather than 25P02, and `it_does_not_roll_back_a_transaction_the_caller_owns` pins the depth guard. The first fails on main with `left: Error, right: Idle`. --- sqlx-postgres/src/connection/mod.rs | 49 ++++++++- sqlx-postgres/src/lib.rs | 2 +- sqlx-postgres/src/message/ready_for_query.rs | 2 +- tests/postgres/postgres.rs | 101 ++++++++++++++++++- 4 files changed, 150 insertions(+), 4 deletions(-) diff --git a/sqlx-postgres/src/connection/mod.rs b/sqlx-postgres/src/connection/mod.rs index 76637a8a93..b441261aaa 100644 --- a/sqlx-postgres/src/connection/mod.rs +++ b/sqlx-postgres/src/connection/mod.rs @@ -147,6 +147,33 @@ impl PgConnection { } } + /// The transaction status the server reported in the most recent + /// [`ReadyForQuery`][crate::message::ReadyForQuery] message. + /// + /// This is the *server's* view of the session, which is not always the + /// client's: [`Connection::is_in_transaction`] reports the client-side + /// transaction depth, and that depth can be zero while the server is + /// inside a block. A future cancelled while `BEGIN` is in flight, or a + /// statement that fails inside a block opened by a multi-statement + /// [`raw_sql`][crate::raw_sql] query, both leave the session in + /// [`TransactionStatus::Transaction`] or [`TransactionStatus::Error`] + /// with a depth of zero. + /// + /// Reading this costs no round trip: the status byte rides along on every + /// `ReadyForQuery`, so it is already in memory. + /// + /// ```rust,no_run + /// # use sqlx_postgres::{PgConnection, TransactionStatus}; + /// # fn example(conn: &PgConnection) { + /// if conn.transaction_status() != TransactionStatus::Idle { + /// // the session is inside a transaction block, open or failed + /// } + /// # } + /// ``` + pub fn transaction_status(&self) -> TransactionStatus { + self.inner.transaction_status + } + pub(crate) async fn invalidate_cached_statement( &mut self, sql: &str, @@ -209,7 +236,27 @@ impl Connection for PgConnection { // The simplest call-and-response that's possible. self.write_sync(); - self.wait_until_ready().await + self.wait_until_ready().await?; + + // `wait_until_ready` has just refreshed `transaction_status` from the server's own + // `ReadyForQuery`, so this is free -- and it is the only view that catches a session + // left inside a block with a client-side depth of zero. `Pool` pings on release, which + // makes this the point where such a connection is cleaned up instead of being handed + // to the next borrower. See `transaction_status` for how the depth comes to disagree. + if self.inner.transaction_depth == 0 + && self.inner.transaction_status != TransactionStatus::Idle + { + // The depth guard above matters: `ping` is public and a caller may well use it + // inside a transaction they are deliberately holding. A non-zero depth means the + // client knows about the block and owns its lifetime, so it is left alone; only a + // block the client has no record of is ended here. That also means there is never + // a savepoint to restore to, hence a plain `ROLLBACK`. + self.queue_simple_query("ROLLBACK")?; + self.inner.stream.flush().await?; + self.wait_until_ready().await?; + } + + Ok(()) } fn begin( diff --git a/sqlx-postgres/src/lib.rs b/sqlx-postgres/src/lib.rs index f68c928881..e083f6f551 100644 --- a/sqlx-postgres/src/lib.rs +++ b/sqlx-postgres/src/lib.rs @@ -55,7 +55,7 @@ pub use copy::{PgCopyIn, PgPoolCopyExt}; pub use database::Postgres; pub use error::{PgDatabaseError, PgErrorPosition}; pub use listener::{PgListener, PgNotification}; -pub use message::PgSeverity; +pub use message::{PgSeverity, TransactionStatus}; pub use options::{PgConnectOptions, PgSslMode}; pub use query_result::PgQueryResult; pub use row::PgRow; diff --git a/sqlx-postgres/src/message/ready_for_query.rs b/sqlx-postgres/src/message/ready_for_query.rs index a1f6761b89..2946f6fc45 100644 --- a/sqlx-postgres/src/message/ready_for_query.rs +++ b/sqlx-postgres/src/message/ready_for_query.rs @@ -3,7 +3,7 @@ use sqlx_core::bytes::Bytes; use crate::error::Error; use crate::message::{BackendMessage, BackendMessageFormat}; -#[derive(Debug)] +#[derive(Debug, Clone, Copy, PartialEq, Eq)] #[repr(u8)] pub enum TransactionStatus { /// Not in a transaction block. diff --git a/tests/postgres/postgres.rs b/tests/postgres/postgres.rs index 0e5f2829c9..2bbddbfdb7 100644 --- a/tests/postgres/postgres.rs +++ b/tests/postgres/postgres.rs @@ -3,7 +3,7 @@ use futures_util::{Stream, StreamExt, TryStreamExt}; use sqlx::postgres::types::Oid; use sqlx::postgres::{ PgAdvisoryLock, PgConnectOptions, PgConnection, PgDatabaseError, PgErrorPosition, PgListener, - PgPoolOptions, PgRow, PgSeverity, Postgres, PG_COPY_MAX_DATA_LEN, + PgPoolOptions, PgRow, PgSeverity, Postgres, TransactionStatus, PG_COPY_MAX_DATA_LEN, }; use sqlx::{Column, Connection, Executor, Row, SqlSafeStr, Statement, TypeInfo}; use sqlx_core::sql_str::AssertSqlSafe; @@ -2295,3 +2295,102 @@ async fn it_rolls_back_a_transaction_cancelled_during_begin() -> anyhow::Result< Ok(()) } + +// Regression: a connection must not be returned to the pool inside a transaction block the +// client has no record of. `PgConnection::ping` -- which `Floating::return_to_pool` uses to +// validate a connection on release -- used to be a bare `wait_until_ready`, draining the +// `ReadyForQuery` without ever inspecting its transaction-status byte. Two shapes reach it +// with a client-side `transaction_depth` of zero, so neither drop guard queues a `ROLLBACK`: +// a future cancelled while `BEGIN` is in flight (#4393), and a statement that fails inside a +// block opened by a multi-statement `raw_sql`. The second never self-heals: the next +// borrower's `BEGIN` fails because the block is already aborted, which leaves the depth at +// zero again, so every checkout fails with 25P02 until `max_lifetime` recycles it. +#[sqlx_macros::test] +async fn it_does_not_return_a_connection_inside_a_transaction_to_the_pool() -> anyhow::Result<()> { + let pool = PgPoolOptions::new() + .max_connections(1) + .min_connections(0) + .connect(&dotenvy::var("DATABASE_URL")?) + .await?; + + // Shape 2: a block opened and then aborted, entirely within one simple-query message, so + // the client's transaction depth is never raised. Deterministic -- no timing involved. + { + let mut conn = pool.acquire().await?; + let err = sqlx::raw_sql("BEGIN; SELECT 1/0;") + .execute(&mut *conn) + .await + .expect_err("division by zero should fail"); + assert_eq!( + err.as_database_error().and_then(|e| e.code()).as_deref(), + Some("22012") + ); + assert!( + !conn.is_in_transaction(), + "precondition: the client-side depth stays zero, which is what defeats the guards" + ); + // released here: `ping` must end the block + } + + // The same connection comes back (the pool holds exactly one). + let mut conn = pool.acquire().await?; + assert_eq!( + conn.transaction_status(), + TransactionStatus::Idle, + "connection was returned to the pool still inside a transaction block" + ); + let one: i32 = sqlx::query_scalar("SELECT 1").fetch_one(&mut *conn).await?; + assert_eq!(one, 1, "a recycled connection must still answer queries"); + drop(conn); + + // Shape 1's lasting damage: a leaked `BEGIN READ ONLY` makes the next borrower's WRITE + // fail, which is how this usually surfaces in production rather than as 25P02. + { + let mut conn = pool.acquire().await?; + sqlx::raw_sql("BEGIN READ ONLY").execute(&mut *conn).await?; + assert!(!conn.is_in_transaction()); + } + + let mut conn = pool.acquire().await?; + let read_only: String = sqlx::query_scalar("SELECT current_setting('transaction_read_only')") + .fetch_one(&mut *conn) + .await?; + assert_eq!( + read_only, "off", + "connection was returned to the pool inside a READ ONLY block" + ); + sqlx::query("CREATE TEMP TABLE ping_rollback_probe (x int)") + .execute(&mut *conn) + .await?; + + Ok(()) +} + +// The other side of the guard above: `ping` must leave a transaction the caller is +// deliberately holding completely alone. `ping` is public API, and rolling back a live +// transaction underneath a caller would be far worse than the bug being fixed. +#[sqlx_macros::test] +async fn it_does_not_roll_back_a_transaction_the_caller_owns() -> anyhow::Result<()> { + let mut conn = new::().await?; + + let mut tx = conn.begin().await?; + sqlx::query("CREATE TEMP TABLE ping_keeps_my_transaction (x int)") + .execute(&mut *tx) + .await?; + + tx.ping().await?; + + // Still inside the same transaction, and its work survived. + let count: i64 = sqlx::query_scalar("SELECT count(*) FROM ping_keeps_my_transaction") + .fetch_one(&mut *tx) + .await?; + assert_eq!(count, 0, "the temp table must still exist after a ping"); + assert!( + tx.is_in_transaction(), + "ping rolled back the caller's transaction" + ); + + tx.commit().await?; + + Ok(()) +}