Rust #325: roll back a PostgreSQL transaction cancelled during BEGIN

rust/Cargo.toml

@@ -53,3 +53,4 @@ debug = 0
 [patch.crates-io]
 apache-avro = { path = "vendor/apache-avro" }
 sqlx-mysql = { path = "vendor/sqlx-mysql" }
+sqlx-postgres = { path = "vendor/sqlx-postgres" }

rust/src/runtime/postgres.rs

@@ -348,6 +348,52 @@ mod tests {
         }
     }
 
+    /// A request dropped while its transaction's `BEGIN` is in flight must not
+    /// leave the pooled connection inside a transaction. With published SQLx
+    /// 0.9.0 it did (vendor/sqlx-postgres/DURABLE-PATCH.md): the next borrower's
+    /// probe ran inside it, native's read snapshot `BEGIN` then failed with
+    /// 25001, and every later statement on that connection with 25P02.
+    #[tokio::test]
+    #[ignore = "requires a disposable PostgreSQL; run the qualification_ tests explicitly"]
+    async fn qualification_a_begin_cancelled_in_flight_leaves_no_transaction_behind() {
+        let db = Database::new().await;
+        let pool = PgPoolOptions::new()
+            .max_connections(1)
+            .min_connections(0)
+            .connect_with(db.options.clone())
+            .await
+            .unwrap();
+        let pid: i32 = sqlx::query_scalar("SELECT pg_backend_pid()").fetch_one(&pool).await.unwrap();
+        // The sleep holds the BEGIN's round trip open so the cancellation lands inside it.
+        let cancelled = tokio::time::timeout(Duration::from_millis(300), pool.begin_with("BEGIN; SELECT pg_sleep(2)")).await;
+        assert!(cancelled.is_err(), "the begin must still be in flight when it is cancelled");
+        // The queued ROLLBACK leaves once the abandoned statement has answered.
+        tokio::time::sleep(Duration::from_millis(3500)).await;
+
+        let mut observer = db.connection().await;
+        let state: Option<String> = sqlx::query_scalar("SELECT state FROM pg_stat_activity WHERE pid=$1")
+            .bind(pid)
+            .fetch_optional(&mut observer)
+            .await
+            .unwrap();
+        assert_eq!(state.as_deref(), Some("idle"), "the pooled connection is still inside a transaction");
+
+        // The scheduler's order on the same connection: a pooled probe, then a
+        // read snapshot, then a pooled statement.
+        let same: i32 = sqlx::query_scalar("SELECT pg_backend_pid()").fetch_one(&pool).await.unwrap();
+        assert_eq!(same, pid);
+        let mut read = pool.begin_with(<Postgres as super::super::backend::Backend>::READ_SNAPSHOT_BEGIN).await.unwrap();
+        let one: i32 = sqlx::query_scalar("SELECT 1").fetch_one(&mut *read).await.unwrap();
+        assert_eq!(one, 1);
+        read.rollback().await.unwrap();
+        let one: i32 = sqlx::query_scalar("SELECT 1").fetch_one(&pool).await.unwrap();
+        assert_eq!(one, 1);
+
+        observer.close().await.unwrap();
+        pool.close().await;
+        db.remove().await;
+    }
+
     #[test]
     fn compiled_catalog_preserves_long_object_identities_and_int64_limits() {
         let rows: Catalog =

rust/vendor/sqlx-postgres/.cargo_vcs_info.json

@@ -0,0 +1,6 @@
+{
+  "git": {
+    "sha1": "003b698e99e024f3621b8043a2426fde5b741171"
+  },
+  "path_in_vcs": "sqlx-postgres"
+}