fix(postgres): end a transaction block left open on a pooled connection

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(

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;

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.

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::<Postgres>().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(())
+}