diff --git a/src/client_live_tests.rs b/src/client_live_tests.rs index 09085be..6c645d1 100644 --- a/src/client_live_tests.rs +++ b/src/client_live_tests.rs @@ -1716,3 +1716,627 @@ async fn prepare_error_in_open_transaction_on_direct_connection_surfaces_real_sq let key = connection_key(¶ms); POOLS.lock().unwrap().remove(&key); } + +/// Regression test for #181 — the 25P02 cascade from a `query_typed` *runtime* +/// error inside an open transaction (the path #170 deliberately left +/// unchanged; the same cascade class from a different source). +/// +/// #170 closed the cascade for the **prepare-failure** path: a prepare +/// Parse-error inside an open transaction aborts it, and #170's short-circuit +/// surfaces the real error + ROLLBACKs. But a **runtime** error at execute +/// time (division by zero, a constraint violation surfacing at fetch, a 42P01 +/// on a relation dropped between prepare and fetch) also aborts the +/// transaction — and #170's `txn_aborted: false` on the `query_typed` error +/// arm means the caller keeps `in_transaction=true` and re-pins the aborted +/// connection. The next batch's first statement 25P02s. +/// +/// This test uses `SELECT 1/0` — syntactically valid, so the pre-resolve +/// `prepare` *succeeds* (warm cache, `prepare_err=None`), and the 22012 +/// division-by-zero fires at `query_typed` execute. On current code (pre-#181) +/// the follow-up `SELECT 1` 25P02-cascades; with #181 it succeeds. +#[tokio::test] +#[ignore = "needs a live PostgreSQL (PG_PLUGIN_TEST_HOST=127.0.0.1 ... podman 54320)"] +async fn query_typed_runtime_error_in_open_transaction_does_not_cascade() { + use crate::handlers::query::{execute_query_batch, release_session}; + use serde_json::json; + + let _guard = POOLS_TEST_LOCK.lock().await; + + let params = live_params(&correct_password()) + .expect("PG_PLUGIN_TEST_HOST not set — set PG_PLUGIN_TEST_HOST=127.0.0.1 (and optionally PG_PLUGIN_TEST_PORT/USER/DB) to run live tests"); + let session_id = "test-181-runtime-cascade"; + + // Batch 1: BEGIN (opens a transaction, pins the connection, sets + // in_transaction=true for the next batch). + let begin_params = json!({ + "driver": "postgresql", + "host": params.host, + "port": params.port, + "database": params.database, + "username": params.username, + "password": params.password, + "pooler_mode": "transaction", + "queries": ["BEGIN"], + "session_id": session_id, + "limit": 100, + "page": 1, + }); + let begin_resp = execute_query_batch(json!(1), &begin_params).await; + assert!( + begin_resp.get("error").is_none(), + "BEGIN batch should succeed; got: {begin_resp}" + ); + + // Batch 2: SELECT 1/0. The prepare succeeds (syntactically valid → + // prepare_err=None → the `else` branch of the `if prepare_err.is_some()` + // block in `exec_query_on_client`); the 22012 + // division-by-zero is a runtime error at execute. Inside the open + // transaction, this aborts it server-side. On current code the error + // surfaces (the real 22012 — no 25P02 masks it on the *failing* + // statement), but `txn_aborted: false` means the connection is re-pinned + // in the aborted state. With #181, `txn_aborted: true` forces + // in_transaction=false and the connection is released + ROLLBACK'd. + let select_params = json!({ + "driver": "postgresql", + "host": params.host, + "port": params.port, + "database": params.database, + "username": params.username, + "password": params.password, + "pooler_mode": "transaction", + "queries": ["SELECT 1/0 AS zero"], + "session_id": session_id, + "limit": 50, + "page": 1, + }); + let select_resp = tokio::time::timeout( + std::time::Duration::from_secs(30), + execute_query_batch(json!(2), &select_params), + ) + .await + .expect("the division-by-zero batch hung past 30s — unexpected"); + assert!( + select_resp.get("error").is_none(), + "batch should not top-level error; got: {select_resp}" + ); + let results = select_resp + .get("result") + .and_then(|r| r.get("results")) + .and_then(|r| r.as_array()) + .expect("session-aware batch returns result.results"); + assert_eq!( + results.len(), + 1, + "one query in the batch; got: {select_resp}" + ); + let first = &results[0]; + let err_msg = first + .get("error") + .and_then(|e| e.as_str()) + .expect("the SELECT 1/0 should error (division by zero); got: {first}"); + + // The failing statement surfaces the real 22012 (no 25P02 masks it on + // this statement — the cascade is only on the *next* statement). + assert!( + err_msg.contains("division by zero"), + "the error should be the real 22012 (\"division by zero\"); got: {err_msg:?}" + ); + + // Batch 3: a plain `SELECT 1` on the same session. On current code the + // aborted connection was re-pinned (in_transaction stayed true), so this + // 25P02-cascades — the user sees "current transaction is aborted" instead + // of `1`. With #181, the transaction was closed and the connection + // released, so this runs cleanly. + let followup_params = json!({ + "driver": "postgresql", + "host": params.host, + "port": params.port, + "database": params.database, + "username": params.username, + "password": params.password, + "pooler_mode": "transaction", + "queries": ["SELECT 1 AS one"], + "session_id": session_id, + "limit": 50, + "page": 1, + }); + let followup_resp = tokio::time::timeout( + std::time::Duration::from_secs(30), + execute_query_batch(json!(3), &followup_params), + ) + .await + .expect("the follow-up SELECT hung past 30s — the connection was likely re-pinned in aborted state (#181 cascade)"); + assert!( + followup_resp.get("error").is_none(), + "follow-up batch should not top-level error; got: {followup_resp}" + ); + let followup_results = followup_resp + .get("result") + .and_then(|r| r.get("results")) + .and_then(|r| r.as_array()) + .expect("follow-up batch returns result.results"); + let followup_first = &followup_results[0]; + let followup_err = followup_first.get("error").and_then(|e| e.as_str()); + assert!( + followup_err.is_none(), + "the follow-up SELECT must succeed (the connection was NOT re-pinned in \ + aborted state — the #181 cascade); got: {followup_err:?}" + ); + let followup_rows = followup_first + .get("result") + .and_then(|r| r.get("rows")) + .and_then(|r| r.as_array()) + .expect("follow-up result.rows is an array"); + assert_eq!( + followup_rows[0].as_array().unwrap()[0], + json!(1), + "follow-up SELECT 1 returns 1; got: {followup_first}" + ); + + // Release the session (frees the pinned connection). #181 already + // ROLLBACK'd the aborted transaction. + let _ = release_session(json!(4), &json!({"session_id": session_id})).await; + + // Teardown: evict the pool entry so no state leaks to other tests. + let key = connection_key(¶ms); + POOLS.lock().unwrap().remove(&key); +} + +/// Regression test for #181 on the `execute_typed` (non-result-set) path — +/// the second of the two error arms #170 left as `txn_aborted: false`. A +/// failed DDL/write inside an open transaction aborts it server-side; #181 +/// ROLLBACKs and surfaces the real error so the follow-up statement doesn't +/// 25P02-cascade. +/// +/// Uses `INSERT INTO ...nonexistent_table` — a non-result-set statement that +/// 42P01s at execute (the `execute_typed` arm — the `if !returns_result_set` +/// early-return in `exec_query_on_client`), distinct +/// from the headline test's `SELECT 1/0` which is a result-set statement +/// (the `query_typed` arm). Together the two tests cover both #181 arms. +#[tokio::test] +#[ignore = "needs a live PostgreSQL (PG_PLUGIN_TEST_HOST=127.0.0.1 ... podman 54320)"] +async fn execute_typed_failure_in_open_transaction_does_not_cascade() { + use crate::handlers::query::{execute_query_batch, release_session}; + use serde_json::json; + + let _guard = POOLS_TEST_LOCK.lock().await; + + let params = live_params(&correct_password()) + .expect("PG_PLUGIN_TEST_HOST not set — set PG_PLUGIN_TEST_HOST=127.0.0.1 (and optionally PG_PLUGIN_TEST_PORT/USER/DB) to run live tests"); + let session_id = "test-181-execute-typed-cascade"; + + // Batch 1: BEGIN. + let begin_params = json!({ + "driver": "postgresql", + "host": params.host, + "port": params.port, + "database": params.database, + "username": params.username, + "password": params.password, + "pooler_mode": "transaction", + "queries": ["BEGIN"], + "session_id": session_id, + "limit": 100, + "page": 1, + }); + let begin_resp = execute_query_batch(json!(1), &begin_params).await; + assert!( + begin_resp.get("error").is_none(), + "BEGIN batch should succeed; got: {begin_resp}" + ); + + // Batch 2: INSERT into a nonexistent table — a non-result-set statement + // that 42P01s at execute (the `execute_typed` arm, not `query_typed`). + // Inside the open transaction this aborts it; #181 ROLLBACKs + surfaces + // the real 42P01. + let insert_params = json!({ + "driver": "postgresql", + "host": params.host, + "port": params.port, + "database": params.database, + "username": params.username, + "password": params.password, + "pooler_mode": "transaction", + "queries": ["INSERT INTO plugin_test_181_nonexistent_table VALUES (1)"], + "session_id": session_id, + "limit": 50, + "page": 1, + }); + let insert_resp = tokio::time::timeout( + std::time::Duration::from_secs(30), + execute_query_batch(json!(2), &insert_params), + ) + .await + .expect("the INSERT batch hung past 30s — unexpected"); + assert!( + insert_resp.get("error").is_none(), + "batch should not top-level error; got: {insert_resp}" + ); + let results = insert_resp + .get("result") + .and_then(|r| r.get("results")) + .and_then(|r| r.as_array()) + .expect("session-aware batch returns result.results"); + assert_eq!( + results.len(), + 1, + "one query in the batch; got: {insert_resp}" + ); + let first = &results[0]; + let err_msg = first + .get("error") + .and_then(|e| e.as_str()) + .expect("the INSERT should error (nonexistent table); got: {first}"); + assert!( + err_msg.contains("does not exist"), + "the error should be the real 42P01 (\"does not exist\"); got: {err_msg:?}" + ); + assert!( + !err_msg.contains("current transaction is aborted"), + "the error must not be the 25P02 cascade; got: {err_msg:?}" + ); + assert!( + err_msg.contains("rolled back"), + "the error should tell the user the transaction was rolled back; got: {err_msg:?}" + ); + + // Batch 3: SELECT 1 — must succeed (no 25P02 cascade). + let followup_params = json!({ + "driver": "postgresql", + "host": params.host, + "port": params.port, + "database": params.database, + "username": params.username, + "password": params.password, + "pooler_mode": "transaction", + "queries": ["SELECT 1 AS one"], + "session_id": session_id, + "limit": 50, + "page": 1, + }); + let followup_resp = tokio::time::timeout( + std::time::Duration::from_secs(30), + execute_query_batch(json!(3), &followup_params), + ) + .await + .expect("the follow-up SELECT hung past 30s — the #181 cascade on the execute_typed path"); + assert!( + followup_resp.get("error").is_none(), + "follow-up batch should not top-level error; got: {followup_resp}" + ); + let followup_results = followup_resp + .get("result") + .and_then(|r| r.get("results")) + .and_then(|r| r.as_array()) + .expect("follow-up batch returns result.results"); + let followup_first = &followup_results[0]; + assert!( + followup_first + .get("error") + .and_then(|e| e.as_str()) + .is_none(), + "the follow-up SELECT must succeed (the connection was NOT re-pinned in \ + aborted state — the #181 cascade); got: {followup_first}" + ); + + let _ = release_session(json!(4), &json!({"session_id": session_id})).await; + let key = connection_key(¶ms); + POOLS.lock().unwrap().remove(&key); +} + +/// Regression test for #181 on the **direct-connection** batch path inside an +/// open transaction (mirroring #170's direct-connection variant). The +/// headline test uses `pooler_mode: "transaction"`; this one has no +/// `pooler_mode` (direct connection) but still opens a transaction via a +/// bare-batch `BEGIN`, exercising the `else` arm of `run_batch_in_session` +/// with `in_transaction=true` and `pre_resolve_types=true`. A runtime error +/// (`SELECT 1/0`) must short-circuit the cascade here too. +#[tokio::test] +#[ignore = "needs a live PostgreSQL (PG_PLUGIN_TEST_HOST=127.0.0.1 ... podman 54320)"] +async fn query_typed_runtime_error_on_direct_connection_in_txn_does_not_cascade() { + use crate::handlers::query::execute_query_batch; + use serde_json::json; + + let _guard = POOLS_TEST_LOCK.lock().await; + + let params = live_params(&correct_password()) + .expect("PG_PLUGIN_TEST_HOST not set — set PG_PLUGIN_TEST_HOST=127.0.0.1 (and optionally PG_PLUGIN_TEST_PORT/USER/DB) to run live tests"); + + // Bare batch (no `session_id`, no `pooler_mode` → direct connection) with + // BEGIN then SELECT 1/0. The BEGIN opens a transaction on the shared + // direct client, so the SELECT routes to the `else` arm with + // `in_transaction=true` and `pre_resolve_types=true`. The 22012 + // division-by-zero fires at execute; #181 ROLLBACKs + forces + // `in_transaction=false`, so the connection is released (not re-pinned + // aborted) and a follow-up `SELECT 1` succeeds. + let batch_params = json!({ + "driver": "postgresql", + "host": params.host, + "port": params.port, + "database": params.database, + "username": params.username, + "password": params.password, + // No `pooler_mode` → direct connection. + "queries": ["BEGIN", "SELECT 1/0 AS zero"], + "limit": 50, + "page": 1, + }); + let resp = tokio::time::timeout( + std::time::Duration::from_secs(30), + execute_query_batch(json!(1), &batch_params), + ) + .await + .expect("the bare direct-connection batch hung past 30s — unexpected"); + assert!( + resp.get("error").is_none(), + "batch should not top-level error; got: {resp}" + ); + let results = resp + .get("result") + .and_then(|r| r.as_array()) + .expect("bare batch returns a result array"); + assert_eq!(results.len(), 2, "two queries in the batch; got: {resp}"); + + // Statement 1: BEGIN succeeds. + let begin_outcome = &results[0]; + assert!( + begin_outcome + .get("error") + .and_then(|e| e.as_str()) + .is_none(), + "BEGIN should succeed; got: {begin_outcome}" + ); + + // Statement 2: SELECT 1/0 — the real 22012 surfaces, the transaction is + // rolled back. + let select_outcome = &results[1]; + let err_msg = select_outcome + .get("error") + .and_then(|e| e.as_str()) + .expect("the SELECT 1/0 should error (division by zero); got: {select_outcome}"); + assert!( + err_msg.contains("division by zero"), + "the error should be the real 22012 (\"division by zero\"); got: {err_msg:?}" + ); + assert!( + err_msg.contains("rolled back"), + "the error should tell the user the transaction was rolled back; got: {err_msg:?}" + ); + + // Follow-up batch: SELECT 1 on the same (now-released) connection. Must + // succeed — the aborted transaction was ROLLBACK'd, not re-pinned. + let followup_params = json!({ + "driver": "postgresql", + "host": params.host, + "port": params.port, + "database": params.database, + "username": params.username, + "password": params.password, + "queries": ["SELECT 1 AS one"], + "limit": 50, + "page": 1, + }); + let followup_resp = tokio::time::timeout( + std::time::Duration::from_secs(30), + execute_query_batch(json!(2), &followup_params), + ) + .await + .expect("the follow-up SELECT hung past 30s — the #181 cascade on the direct path"); + let followup_results = followup_resp + .get("result") + .and_then(|r| r.as_array()) + .expect("bare batch returns a result array"); + let followup_first = &followup_results[0]; + assert!( + followup_first + .get("error") + .and_then(|e| e.as_str()) + .is_none(), + "the follow-up SELECT must succeed (no 25P02 cascade); got: {followup_first}" + ); + + let key = connection_key(¶ms); + POOLS.lock().unwrap().remove(&key); +} + +/// Regression guard for the `Closes`/`Chains` gate on #181's two error arms +/// (`execute_typed` and `query_typed` in `exec_query_on_client`). The gate's +/// *negative* behavior — a failed `COMMIT`/`ROLLBACK`/`END`/`COMMIT AND CHAIN` +/// issued via the **batch** path while `in_transaction=true` must NOT receive +/// the spurious ROLLBACK + "rolled back" notice — is uncovered by any other +/// test. The existing `commit_that_fails_on_a_deferred_constraint_releases_ +/// the_session` test in `tests/live_db.rs` uses the single-statement +/// `execute_query` RPC, which passes `in_transaction=false` to +/// `exec_query_on_client` — so the `Closes`/`Chains` gate is never evaluated +/// there, and its assertion (`commit["error"].is_string()` only) would still +/// pass if #181 wrongly appended "rolled back" to a failed COMMIT's error. +/// +/// This test exercises the gate via the batch path: `["BEGIN", "INSERT +/// (violating)", "COMMIT"]` as a session-pinned batch on a transaction +/// pooler, so the COMMIT fails with the deferred-FK violation while +/// `in_transaction=true` (the `BEGIN` set it). The gate must keep +/// `txn_aborted=false` for the failed `COMMIT` (a `Closes` verb — already +/// excluded), so: +/// - the COMMIT's error carries the real SQLSTATE (foreign key violation), +/// NOT "current transaction is aborted" (no 25P02 cascade — the `Closes` +/// path in `in_transaction_after_outcome` already forces +/// `in_transaction=false`), AND +/// - the COMMIT's error does NOT contain "rolled back" (the gate excluded +/// `Closes`, so no spurious ROLLBACK + notice fires for a transaction +/// the server already rolled back), AND +/// - the follow-up `SELECT 1` succeeds (the connection was released, not +/// re-pinned aborted). +/// +/// If someone deleted the `&& !matches!(transaction_effect(query), Closes | +/// Chains)` conjunct from either arm, the COMMIT's error here would gain a +/// spurious "rolled back" suffix and the `!contains("rolled back")` assertion +/// would fail — catching the regression. +#[tokio::test] +#[ignore = "needs a live PostgreSQL (PG_PLUGIN_TEST_HOST=127.0.0.1 ... podman 54320)"] +async fn failed_commit_in_batch_does_not_get_spurious_rolled_back_notice() { + use crate::client; + use crate::handlers::query::{execute_query_batch, release_session}; + use serde_json::json; + + let _guard = POOLS_TEST_LOCK.lock().await; + + let params = live_params(&correct_password()) + .expect("PG_PLUGIN_TEST_HOST not set — set PG_PLUGIN_TEST_HOST=127.0.0.1 (and optionally PG_PLUGIN_TEST_PORT/USER/DB) to run live tests"); + + // Setup: a table with a DEFERRABLE INITIALLY DEFERRED self-FK so the + // INSERT succeeds but COMMIT fails (the FK check fires at commit). + let table = "plugin_test_181_gate_deferred_fk"; + for stmt in [ + format!("DROP TABLE IF EXISTS {table}"), + format!( + "CREATE TABLE {table} (id INT PRIMARY KEY, other_id INT \ + REFERENCES {table}(id) DEFERRABLE INITIALLY DEFERRED)" + ), + ] { + client::execute_typed(¶ms, &stmt, &[]) + .await + .unwrap_or_else(|e| panic!("setup statement failed ({stmt}): {e}")); + } + + let session_id = "test-181-closes-gate"; + + // The batch: BEGIN opens a transaction (sets in_transaction=true for the + // subsequent statements in this session-pinned batch), the INSERT + // succeeds (FK is deferred), then COMMIT fails with the FK violation. + // The COMMIT is a `Closes`-effect statement, so the Closes/Chains gate + // must keep `txn_aborted=false` for it — no spurious ROLLBACK, no + // "rolled back" suffix. The `Closes` path in `in_transaction_after_ + // outcome` already forces `in_transaction=false` (the server ended the + // transaction when COMMIT failed), so there's no 25P02 cascade either. + let batch_params = json!({ + "driver": "postgresql", + "host": params.host, + "port": params.port, + "database": params.database, + "username": params.username, + "password": params.password, + "pooler_mode": "transaction", + "queries": [ + "BEGIN", + format!("INSERT INTO {table} VALUES (1, 999)"), + "COMMIT", + ], + "session_id": session_id, + "limit": 100, + "page": 1, + }); + let resp = tokio::time::timeout( + std::time::Duration::from_secs(30), + execute_query_batch(json!(1), &batch_params), + ) + .await + .expect("the BEGIN/INSERT/COMMIT batch hung past 30s — unexpected"); + assert!( + resp.get("error").is_none(), + "batch should not top-level error; got: {resp}" + ); + let results = resp + .get("result") + .and_then(|r| r.get("results")) + .and_then(|r| r.as_array()) + .expect("session-aware batch returns result.results"); + assert_eq!(results.len(), 3, "three queries in the batch; got: {resp}"); + + // Statement 1 (BEGIN) and 2 (INSERT) succeed. + assert!( + results[0].get("error").and_then(|e| e.as_str()).is_none(), + "BEGIN should succeed; got: {}", + results[0] + ); + assert!( + results[1].get("error").and_then(|e| e.as_str()).is_none(), + "the deferred-FK INSERT should succeed (FK check fires at COMMIT); got: {}", + results[1] + ); + + // Statement 3 (COMMIT) fails with the real FK violation — the gate's + // negative behavior is what we're guarding. + let commit_outcome = &results[2]; + let err_msg = commit_outcome + .get("error") + .and_then(|e| e.as_str()) + .expect("the COMMIT should error (deferred FK violation); got: {commit_outcome}"); + + // The COMMIT's error surfaces the real SQLSTATE (foreign key violation), + // NOT a 25P02 cascade ("current transaction is aborted"). The `Closes` + // path in `in_transaction_after_outcome` already forces + // `in_transaction=false`, so the failed COMMIT's reply reports + // `in_transaction: false` and the connection is released — no re-pin of + // an aborted connection. + assert!( + err_msg.contains("foreign key") || err_msg.contains("violates"), + "the error should be the real FK violation (\"foreign key\" / \"violates\"), \ + not a 25P02 cascade; got: {err_msg:?}" + ); + assert!( + !err_msg.contains("current transaction is aborted"), + "the error must not be the 25P02 cascade; got: {err_msg:?}" + ); + + // The load-bearing negative assertion: the failed COMMIT's error must + // NOT contain "rolled back". The Closes/Chains gate excludes `COMMIT` + // (a `Closes` verb), so #181's spurious ROLLBACK + "rolled back" notice + // must NOT fire — the server already rolled back the transaction when + // COMMIT failed; our ROLLBACK would be a no-op and our notice would be + // factually wrong (the failing statement WAS the commit, not "a statement + // in it"). If the gate's `Closes` conjunct is removed, this assertion + // fails. + assert!( + !err_msg.contains("rolled back"), + "the failed COMMIT must NOT get the 'rolled back' notice (the Closes/Chains \ + gate excludes it — the server already rolled back); got: {err_msg:?}" + ); + + // Follow-up batch on the same session: SELECT 1 must succeed (the + // connection was released, not re-pinned aborted). The `Closes` path set + // `in_transaction=false`, so the session slot is empty and a fresh + // connection runs the follow-up. + let followup_params = json!({ + "driver": "postgresql", + "host": params.host, + "port": params.port, + "database": params.database, + "username": params.username, + "password": params.password, + "pooler_mode": "transaction", + "queries": ["SELECT 1 AS one"], + "session_id": session_id, + "limit": 50, + "page": 1, + }); + let followup_resp = tokio::time::timeout( + std::time::Duration::from_secs(30), + execute_query_batch(json!(2), &followup_params), + ) + .await + .expect("the follow-up SELECT hung past 30s — the connection was re-pinned in aborted state"); + assert!( + followup_resp.get("error").is_none(), + "follow-up batch should not top-level error; got: {followup_resp}" + ); + let followup_results = followup_resp + .get("result") + .and_then(|r| r.get("results")) + .and_then(|r| r.as_array()) + .expect("follow-up batch returns result.results"); + let followup_first = &followup_results[0]; + assert!( + followup_first + .get("error") + .and_then(|e| e.as_str()) + .is_none(), + "the follow-up SELECT must succeed (no 25P02 cascade — the Closes path \ + released the connection); got: {followup_first}" + ); + + // Release the session and teardown. + let _ = release_session(json!(3), &json!({"session_id": session_id})).await; + let _ = client::execute_typed(¶ms, &format!("DROP TABLE IF EXISTS {table}"), &[]).await; + let key = connection_key(¶ms); + POOLS.lock().unwrap().remove(&key); +} diff --git a/src/handlers/query.rs b/src/handlers/query.rs index 2312a1e..2838afa 100644 --- a/src/handlers/query.rs +++ b/src/handlers/query.rs @@ -522,20 +522,48 @@ async fn exec_query( const POOLED_TXN_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(90); /// The outcome of `exec_query_on_client`, carrying a flag the batch caller needs -/// to decide whether the transaction it was tracking is still open (#170). +/// to decide whether the transaction it was tracking is still open (#170 + #181). /// -/// `txn_aborted` is `true` only when the pre-resolve `prepare` failed *inside an -/// open transaction* and the handler short-circuited — ROLLBACK'd the aborted -/// transaction server-side and returned the prepare's real error instead of -/// running `query_typed` into a guaranteed 25P02 ("current transaction is -/// aborted"). The caller (`run_batch_in_session`) treats `txn_aborted=true` as -/// `Closes`-equivalent: force `in_transaction=false` so the (now-healthy) -/// connection is not re-pinned as if its transaction were still open. +/// `txn_aborted` is `true` in two cases (both inside an open transaction, both +/// after the handler ROLLBACK'd the aborted transaction server-side): /// -/// It is `false` on every other path: success, a prepare failure *outside* a -/// transaction (the recoverable swallow — a 42P05 on a real PgBouncer bare -/// batch where `query_typed` may still succeed), a `query_typed`/extraction -/// error, or a timeout fired outside a transaction. +/// 1. **#170 — prepare-failure short-circuit:** the pre-resolve `prepare` failed +/// inside an open transaction (a Parse-time error: 42P01, 42501, syntax +/// error, …). The handler short-circuits — ROLLBACK + surface the prepare's +/// real error instead of running `query_typed` into a guaranteed 25P02 +/// ("current transaction is aborted"). +/// 2. **#181 — runtime-error classification:** `query_typed`/`execute_typed` +/// failed at execute time inside an open transaction (a runtime error: +/// division by zero, a constraint violation at fetch, a failed DDL/write). +/// A server `DbError` (any ERROR-class response — WARNING/NOTICE never +/// arrive as a query `Error`, they're out-of-band `AsyncMessage::Notice`) +/// always aborts an open transaction; non-`DbError` errors (transport/ +/// timeout) are conservatively assumed to have aborted it. The handler +/// ROLLBACKs + surfaces the real error + signals `txn_aborted=true`. +/// +/// In both cases the caller (`run_batch_in_session`) treats `txn_aborted=true` +/// as `Closes`-equivalent: force `in_transaction=false` so the (now-healthy) +/// connection is not re-pinned as if its transaction were still open (which +/// would 25P02-cascade the next batch's first statement). Gated on +/// `transaction_effect(query)` being neither `Closes` nor `Chains` so a failed +/// `COMMIT`/`ROLLBACK`/`END`/`COMMIT AND CHAIN` — which already ended the +/// transaction server-side and is handled by the existing `Closes`/`Chains` +/// path in `in_transaction_after_outcome` — does NOT get a spurious ROLLBACK + +/// "rolled back" notice. +/// +/// `txn_aborted` is `false` on every other path: success; a prepare failure +/// *outside* a transaction (the recoverable swallow — a 42P05 on a real +/// PgBouncer bare batch where `query_typed` may still succeed); a +/// `query_typed`/`execute_typed` error *outside* a transaction (nothing to +/// abort); or a timeout fired outside a transaction. +/// +/// **Note on the one-shot path:** `run_pinned_body` calls +/// `exec_query_on_client` with `in_transaction=true` (it runs inside +/// `pin_pooled_txn`'s own BEGIN/COMMIT), so a runtime error there CAN set +/// `txn_aborted=true`. But `run_pinned_body` intentionally ignores the flag +/// (returns only `outcome.result`) — `pin_pooled_txn` owns the one-shot +/// client's transaction lifecycle and ROLLBACKs on `Err` regardless. So +/// `txn_aborted=true` on that path does not flow to a batch caller. struct QueryOutcome { result: Result, txn_aborted: bool, @@ -558,8 +586,73 @@ fn should_short_circuit_on_prepare_err(in_transaction: bool) -> bool { in_transaction } +/// Pure inner: classify whether an error aborted the open transaction, given +/// whether the error is a server `DbError` and whether a transaction is open. +/// Extracted so the classification is unit-testable without a live +/// `tokio_postgres::Error` (whose `DbError` can't be constructed without a +/// real server response — same constraint that drove `pooler_error_lead`'s +/// extraction in `client.rs`). +/// +/// Companion to `should_short_circuit_on_prepare_err` (#170): that helper +/// decides whether to short-circuit *before* `query_typed` runs (a prepare +/// failure); this one classifies *after* a `query_typed`/`execute_typed` +/// runtime error (the #181 path). Together they are the two faces of the 25P02 +/// cascade — #170 (prepare) + #181 (runtime). +fn error_aborts_open_txn_inner(is_db_error: bool, in_transaction: bool) -> bool { + if !in_transaction { + return false; + } + // A `DbError` is a server ERROR-class response — every one puts an open + // transaction block into the 'E'/failed state (WARNING/NOTICE never arrive + // as a query `Error`; they're delivered out-of-band as `AsyncMessage::Notice`, + // so every `Error` carrying a `DbError` is ERROR-class). A non-`DbError` + // error is transport/timeout/client-side — the server's verdict is unknown, + // so conservatively assume aborted (the rare `FromSql` over-fire, where + // the txn is actually untouched, is documented in `error_aborted_open_txn`). + // + // `is_db_error || true` collapses to `true` (so `is_db_error` is a no-op + // here): both the DbError and the conservative non-DbError cases return + // `true`. The `is_db_error` parameter is kept *deliberately* so the + // `error_aborts_open_txn_inner_truth_table` unit test can assert the + // intent for all 4 `(is_db_error, in_transaction)` cases — documenting + // that both error classes abort when `in_transaction`, not just one. + // This reads as a tautology (clippy's `overly_complex_bool_expr` notes + // it under `-W clippy::pedantic`, but the default `-D warnings` is clean); + // it is NOT a typo for `&&`/`!`. The outer `error_aborted_open_txn` still + // threads `e.as_db_error().is_some()` so the discriminator is exercised + // end-to-end, even though the inner collapses it. + is_db_error || true +} + +/// Whether a `query_typed`/`execute_typed` error inside an open transaction +/// aborted the transaction server-side (#181). See +/// `error_aborts_open_txn_inner` for the pure logic. +/// +/// A server `DbError` (any ERROR-class response) always aborts an open +/// transaction. Non-`DbError` errors (transport/timeout) are conservatively +/// assumed to have aborted it — the server's verdict is unknown and leaving a +/// possibly-aborted transaction on a pooled connection would poison the pool +/// for the next borrower. +/// +/// Residual: `Kind::FromSql` (a client-side row-decode failure) fires *after* +/// the server returned rows successfully, so the transaction is NOT aborted +/// by it — but `Kind` is private in tokio-postgres (only `is_closed()` leaks +/// one variant), so `FromSql` is indistinguishable from `Timeout`/`Closed`/ +/// `Io` via the public API. The conservative `true` over-fires `FromSql`, +/// issuing a spurious ROLLBACK on a still-valid transaction. This is rare (a +/// bug in our type conversion, not normal operation) and recoverable (the +/// user re-issues `BEGIN`); it's the accepted cost of not poisoning the pool +/// on the more common `Timeout`/`Closed`/`Io` cases. The plugin's query/execute +/// paths pass `&[]` (no params), so `Kind::ToSql` never fires here at all. +/// +/// Returns `false` when not `in_transaction` (nothing to abort). +fn error_aborted_open_txn(e: &tokio_postgres::Error, in_transaction: bool) -> bool { + error_aborts_open_txn_inner(e.as_db_error().is_some(), in_transaction) +} + /// Recompute `in_transaction` after a batch statement, honoring the -/// `txn_aborted` signal from `exec_query_on_client` (#170). Extracted as a +/// `txn_aborted` signal from `exec_query_on_client` (#170 prepare-failure +/// short-circuit + #181 runtime-error classification). Extracted as a /// pure helper so the recompute logic is unit-testable without a live /// database. /// @@ -741,11 +834,27 @@ async fn run_pinned_body( // Decoding + extract_value unchanged. `pre_resolve_types = false` because // `run_pinned_body` already pre-resolved via its own `prepare` above. // `in_transaction = true` because this runs inside `pin_pooled_txn`'s own - // BEGIN/COMMIT wrapper — but since `pre_resolve_types = false`, no prepare - // runs in `exec_query_on_client` here, so the #170 short-circuit can't - // fire (`prepare_err` is `None`); `txn_aborted` is always `false` on this - // path. (`run_pinned_body`'s own `prepare` above already surfaced its - // error directly at line 734-736.) + // BEGIN/COMMIT wrapper. `pre_resolve_types = false` means no pre-resolve + // `prepare` runs in `exec_query_on_client` here, so the #170 prepare-failure + // short-circuit can't fire (`prepare_err` is `None`). + // + // #181's runtime-error classification CAN fire here: `run_pinned_body`'s own + // `prepare` above (line ~788) succeeds for a syntactically-valid query like + // `SELECT 1/0`, then `query_typed` 22012s at execute → `error_aborted_open_txn` + // returns `true` → `txn_aborted = true`, and `exec_query_on_client` issues a + // best-effort ROLLBACK. This is *redundant but harmless*: `pin_pooled_txn` + // (line ~744-748) already ROLLBACKs on `run_pinned_body`'s `Err` and the + // one-shot client drops (closing the connection) regardless. The + // `txn_aborted` signal itself is intentionally ignored on this path — + // `run_pinned_body` returns only `outcome.result`, and `pin_pinned_txn` + // owns the transaction lifecycle (BEGIN/COMMIT/ROLLBACK) for the one-shot + // client. So `txn_aborted=true` here does NOT flow to a batch caller's + // `in_transaction_after_outcome` (there is no batch caller — this is the + // single-query `exec_query_pooled_txn` path). + // + // (`run_pinned_body`'s own `prepare` above already surfaced a *prepare-time* + // error directly at line ~788-790, before `exec_query_on_client` runs; only + // a *runtime* error at `query_typed` reaches #181's classification here.) let outcome = exec_query_on_client( pg_client, query, @@ -767,10 +876,10 @@ async fn run_pinned_body( ); } - debug_assert!( - !outcome.txn_aborted, - "run_pinned_body passes pre_resolve_types=false; the #170 short-circuit can't fire" - ); + // `txn_aborted` may be `true` here (a #181 runtime-error classification on + // the `query_typed` path), but it's intentionally ignored — `pin_pooled_txn` + // owns the one-shot client's transaction lifecycle and ROLLBACKs on this + // function's `Err` regardless. Only `outcome.result` propagates. outcome.result } @@ -911,9 +1020,12 @@ async fn apply_search_path(pg_client: &PgClient, schema: &str) -> Result<(), Str /// `prepare` before the `query_typed` data fetch (#166/#168). `in_transaction` /// is whether this query is running inside an *open* transaction (a /// session-pinned batch reusing a prior `BEGIN`, or a bare batch whose earlier -/// statement was `BEGIN`) — it gates the #170 25P02 short-circuit: a failed -/// `prepare` inside an open transaction aborts it server-side, so -/// `query_typed` is guaranteed to 25P02 rather than surface the real error. +/// statement was `BEGIN`) — it gates both the #170 prepare-failure short- +/// circuit AND the #181 runtime-error classification: a failed `prepare` +/// (#170) or a `query_typed`/`execute_typed` runtime error (#181) inside an +/// open transaction aborts it server-side, so the handler ROLLBACKs + surfaces +/// the real error + signals `txn_aborted=true` (see `QueryOutcome` and +/// `error_aborted_open_txn`). /// The two flags are independent: a session-pinned pooler connection passes /// `is_transaction_pooler = true` (for correct error phrasing) AND /// `pre_resolve_types = true` (because the backend is pinned by the user's @@ -932,22 +1044,70 @@ async fn exec_query_on_client( // Unnamed-statement protocol (`execute_typed`) — pooler-safe under // transaction-mode poolers (PgBouncer); see `client`'s module note // (tabularis#1097). No params for this branch. - let affected = pg_client - .execute_typed(query, &[]) - .await - .map_err(|e| client::format_pg_error(&e)); - let result = affected.map(|affected| { - json!({ - "columns": [], - "rows": [], - "affected_rows": affected, - "truncated": false, - "pagination": null, + let affected = pg_client.execute_typed(query, &[]).await; + // #181: a runtime error (a failed DDL/write, a constraint violation) + // inside an open transaction aborts it server-side. Classify on the + // raw `tokio_postgres::Error` before formatting, so the caller can + // force `in_transaction=false` and the aborted connection isn't + // re-pinned (which would 25P02-cascade the next batch's first + // statement). Gate on `transaction_effect(query)` being neither `Closes` + // nor `Chains` so a failed `COMMIT`/`ROLLBACK`/`END` (`Closes`) AND a + // failed `COMMIT AND CHAIN`/`ROLLBACK AND CHAIN`/`ABORT AND CHAIN` + // (`Chains` — a failed `Chains` ends the transaction server-side just + // like a failed `Closes`, per `in_transaction_after`'s `(Chains, false) + // => false` arm) are NOT given a spurious ROLLBACK + "rolled back" + // notice. Those verbs already ended the transaction server-side and + // are handled by the existing `Closes`/`Chains` path in + // `in_transaction_after_outcome`. + let txn_aborted = match &affected { + Ok(_) => false, + Err(e) => { + error_aborted_open_txn(e, in_transaction) + && !matches!( + transaction_effect(query), + TransactionEffect::Closes | TransactionEffect::Chains + ) + } + }; + if txn_aborted { + // Best-effort ROLLBACK; the transaction is aborted, so ROLLBACK is + // the one command PostgreSQL still accepts in this state. A + // failure here is logged, not fatal — the connection still drops + // back to the pool and `RecyclingMethod::Fast` will surface a + // stale-state issue on reuse if it ever matters. + if let Err(rollback_err) = pg_client.batch_execute("ROLLBACK").await { + log::warn!( + "ROLLBACK after an execute_typed error aborted a transaction failed: {rollback_err}" + ); + } + } + let result = affected + .map(|affected| { + json!({ + "columns": [], + "rows": [], + "affected_rows": affected, + "truncated": false, + "pagination": null, + }) }) - }); + .map_err(|e| { + let msg = client::format_pg_error(&e); + if txn_aborted { + // Tell the user the transaction was rolled back so the + // state change isn't implicit — a multi-statement batch's + // user needs to know earlier writes were discarded. + format!( + "{msg}\n\nThe transaction was rolled back (a statement in it \ + failed with an error that aborted the transaction)." + ) + } else { + msg + } + }); return QueryOutcome { result, - txn_aborted: false, + txn_aborted, }; } @@ -1133,12 +1293,21 @@ async fn exec_query_on_client( // failed (cold type cache → mid-stream `typeinfo_statement` hang risk, // #160/#163/#170). When the prepare succeeded (cache warm) or was skipped, // `query_typed` is unbounded-safe. - let rows: Result, String> = if prepare_err.is_some() { + // #181: keep the raw `tokio_postgres::Error` here (defer `map_err` to the + // single outer error arm below) so we can classify it with + // `error_aborted_open_txn` before formatting — a runtime error inside an + // open transaction aborted the txn and the caller must not re-pin the + // aborted connection. The timed branch's `Ok(Err(e))` arm is only reached + // when `!in_transaction` (the short-circuit above returned when + // `in_transaction`), so classification there returns `false` (the + // `!in_transaction` guard) and the outer arm's behavior is unchanged for + // that sub-path. + let rows: Result, tokio_postgres::Error> = if prepare_err.is_some() { match tokio::time::timeout(POOLED_TXN_TIMEOUT, pg_client.query_typed(&final_query, &[])) .await { Ok(Ok(rows)) => Ok(rows), - Ok(Err(e)) => Err(client::format_query_error(&e, is_transaction_pooler)), + Ok(Err(e)) => Err(e), Err(_elapsed) => { // The `in_transaction=true` + `prepare_err.is_some()` case was // intercepted by the short-circuit above (which already @@ -1168,17 +1337,49 @@ async fn exec_query_on_client( } } } else { - pg_client - .query_typed(&final_query, &[]) - .await - .map_err(|e| client::format_query_error(&e, is_transaction_pooler)) + pg_client.query_typed(&final_query, &[]).await }; let rows = match rows { Ok(rows) => rows, Err(e) => { + // #181: a `query_typed` runtime error (division by zero, a + // constraint violation at fetch, a 42P01 on a relation dropped + // between prepare and fetch) inside an open transaction aborts + // it server-side. Classify on the raw error before formatting, + // ROLLBACK the now-dead transaction, and signal `txn_aborted` + // so the caller forces `in_transaction=false` and the aborted + // connection isn't re-pinned (which would 25P02-cascade the next + // batch's first statement). Gate on `transaction_effect(query)` + // being neither `Closes` nor `Chains` so a failed + // `COMMIT`/`ROLLBACK`/`END`/`COMMIT AND CHAIN` (already handled by + // the `Closes`/`Chains` path in `in_transaction_after_outcome`) + // doesn't get a spurious ROLLBACK + "rolled back" notice. (Result- + // set queries reaching this arm are never transaction-control + // verbs, so the gate is always-true here — kept for parity with + // the `execute_typed` arm and to document the intent.) + let txn_aborted = error_aborted_open_txn(&e, in_transaction) + && !matches!( + transaction_effect(query), + TransactionEffect::Closes | TransactionEffect::Chains + ); + if txn_aborted { + if let Err(rollback_err) = pg_client.batch_execute("ROLLBACK").await { + log::warn!( + "ROLLBACK after a query_typed error aborted a transaction failed: {rollback_err}" + ); + } + } + let msg = client::format_query_error(&e, is_transaction_pooler); return QueryOutcome { - result: Err(e), - txn_aborted: false, + result: Err(if txn_aborted { + format!( + "{msg}\n\nThe transaction was rolled back (a statement in it \ + failed with an error that aborted the transaction)." + ) + } else { + msg + }), + txn_aborted, }; } }; diff --git a/src/handlers/query_tests.rs b/src/handlers/query_tests.rs index 84f78e8..5057d61 100644 --- a/src/handlers/query_tests.rs +++ b/src/handlers/query_tests.rs @@ -11,8 +11,9 @@ //! pagination is applied to a statement. use super::{ - build_search_path_sql, in_transaction_after_outcome, raw_explain_output, returns_result_set, - sets_session_local_state, should_short_circuit_on_prepare_err, strip_leading_sql_comments, + build_search_path_sql, error_aborts_open_txn_inner, in_transaction_after_outcome, + raw_explain_output, returns_result_set, sets_session_local_state, + should_short_circuit_on_prepare_err, strip_leading_sql_comments, supports_trailing_limit_clause, transaction_effect, TransactionEffect, }; @@ -332,6 +333,56 @@ fn in_transaction_after_outcome_forces_false_when_txn_aborted() { ); } +#[test] +fn error_aborts_open_txn_inner_truth_table() { + // #181: the pure inner classification. A server DbError (any ERROR-class + // response — WARNING/NOTICE never arrive as a query `Error`, they're + // out-of-band `AsyncMessage::Notice`) always aborts an open transaction + // block. A non-DbError (transport/timeout) conservatively assumes aborted + // when `in_transaction` (server verdict unknown — prefer ending the txn + // over poisoning the pool). When not `in_transaction`, nothing to abort. + assert!( + error_aborts_open_txn_inner(/* is_db_error */ true, /* in_transaction */ true), + "a server DbError inside an open txn aborts it" + ); + assert!( + !error_aborts_open_txn_inner(true, false), + "a server DbError outside a transaction has nothing to abort" + ); + assert!( + error_aborts_open_txn_inner(false, true), + "a non-DbError (transport/timeout) inside an open txn conservatively aborts" + ); + assert!( + !error_aborts_open_txn_inner(false, false), + "a non-DbError outside a transaction has nothing to abort" + ); +} + +#[test] +fn error_aborted_open_txn_timeout_error_conservative() { + // #181: exercise the real `error_aborted_open_txn` on a constructible + // non-DbError (`__private_api_timeout` is the one public constructor for a + // tokio_postgres::Error that doesn't carry a DbError). This confirms the + // outer fn delegates to the inner and that the conservative `true` for + // non-DbError fires inside a transaction, `false` outside. + let timeout_err = tokio_postgres::Error::__private_api_timeout(); + assert!( + super::error_aborted_open_txn(&timeout_err, true), + "a timeout error inside an open txn is conservatively treated as aborting it" + ); + assert!( + !super::error_aborted_open_txn(&timeout_err, false), + "a timeout error outside a transaction has nothing to abort" + ); + // The error must NOT carry a DbError (that's the whole point of using + // __private_api_timeout as the test fixture — it's a non-DbError). + assert!( + timeout_err.as_db_error().is_none(), + "the timeout test fixture must be a non-DbError (else the test isn't exercising the conservative path)" + ); +} + #[test] fn transaction_effect_reads_optional_noise_words_and_inner_comments() { for (query, effect) in [