diff --git a/src/client.rs b/src/client.rs index ff2a6c5..be0156d 100644 --- a/src/client.rs +++ b/src/client.rs @@ -71,6 +71,44 @@ pub(crate) fn format_pg_error(e: &tokio_postgres::Error) -> String { } } +/// Format a `tokio_postgres::Error` from a query path, rewriting +/// prepared-statement errors (SQLSTATE 26000 / 42P05) with actionable +/// guidance when the connection is through a transaction-mode pooler. +/// +/// Under PgBouncer `pool_mode = transaction`, tokio-postgres's internal +/// `typeinfo_statement` (fired during `Row` deserialization for non-builtin +/// column types — custom enums, composites, extension types) splits across +/// two Syncs and can be orphaned by pooler reassignment, producing +/// `ERROR: prepared statement "typeinfo_statement" does not exist`. The fix +/// is pooler-side: `max_prepared_statements > 0` in pgbouncer.ini makes +/// PgBouncer track prepared statements across reassignments. This rewrites +/// the cryptic error into that guidance so the user knows what to configure +/// (#157, tabularis#1097). +/// +/// For non-pooler connections or non-prepared-statement errors, falls +/// through to `format_pg_error` unchanged. +pub(crate) fn format_query_error(e: &tokio_postgres::Error, is_transaction_pooler: bool) -> String { + if is_transaction_pooler { + if let Some(code) = e.code() { + if *code == tokio_postgres::error::SqlState::INVALID_SQL_STATEMENT_NAME + || *code == tokio_postgres::error::SqlState::DUPLICATE_PSTATEMENT + { + return format!( + "{}\n\n\ + This error occurs under transaction-mode poolers (e.g. PgBouncer) \ + when a named prepared statement is lost to connection reassignment. \ + For queries returning custom types (enums, composites, extension types), \ + set `max_prepared_statements > 0` in pgbouncer.ini — this enables \ + server-side prepared statement caching so named statements survive \ + reassignment. See: https://www.pgbouncer.org/config.html#max_prepared_statements", + format_pg_error(e) + ); + } + } + } + format_pg_error(e) +} + /// Format a `pool.get()` failure. The pool's own connection-establishment /// handshake (bad database name, bad password, ...) surfaces as a /// `tokio_postgres::Error` wrapped in `PoolError::Backend` — the exact same diff --git a/src/client_live_tests.rs b/src/client_live_tests.rs index dacb1ca..404a9e4 100644 --- a/src/client_live_tests.rs +++ b/src/client_live_tests.rs @@ -48,6 +48,7 @@ fn live_params(password: &str) -> Option { ssl_key: None, connection_string: None, startup_script: None, + pooler_mode: None, }) } diff --git a/src/client_tests.rs b/src/client_tests.rs index 0d8f343..a0c466d 100644 --- a/src/client_tests.rs +++ b/src/client_tests.rs @@ -36,6 +36,7 @@ fn params(host: &str, port: u16, db: &str, user: &str) -> ConnectionParams { ssl_key: None, connection_string: None, startup_script: None, + pooler_mode: None, } } @@ -804,6 +805,7 @@ fn params_with_ssl(ssl_mode: &str) -> ConnectionParams { ssl_key: None, connection_string: None, startup_script: None, + pooler_mode: None, } } diff --git a/src/handlers/query.rs b/src/handlers/query.rs index 9d02758..bff79d0 100644 --- a/src/handlers/query.rs +++ b/src/handlers/query.rs @@ -158,7 +158,14 @@ async fn run_batch_in_session( let mut results: Vec = Vec::new(); for query in queries { let start = Instant::now(); - let outcome = exec_query_on_client(&pg_client, query, limit, page).await; + let outcome = exec_query_on_client( + &pg_client, + query, + limit, + page, + conn_params.is_transaction_pooler(), + ) + .await; let elapsed_ms = start.elapsed().as_secs_f64() * 1000.0; in_transaction = @@ -296,7 +303,14 @@ async fn exec_query( apply_search_path(&pg_client, s).await?; } - exec_query_on_client(&pg_client, query, limit, page).await + exec_query_on_client( + &pg_client, + query, + limit, + page, + conn_params.is_transaction_pooler(), + ) + .await } /// Build a `SET search_path TO` statement that prepends `schema` onto the @@ -386,6 +400,7 @@ async fn exec_query_on_client( query: &str, limit: Option, page: u32, + is_transaction_pooler: bool, ) -> Result { // Check if the statement returns a result set if !returns_result_set(query) { @@ -427,10 +442,15 @@ async fn exec_query_on_client( // Unnamed-statement protocol (`query_typed`) — pooler-safe under // transaction-mode poolers (PgBouncer); see `client`'s module note // (tabularis#1097). No params for user-issued queries. + // + // On a transaction-mode pooler, a prepared-statement error (26000/42P05) + // from tokio-postgres's internal `typeinfo_statement` (fired by + // non-builtin column types) is rewritten with actionable guidance via + // `format_query_error` (#157). let rows = pg_client .query_typed(&final_query, &[]) .await - .map_err(|e| client::format_pg_error(&e))?; + .map_err(|e| client::format_query_error(&e, is_transaction_pooler))?; if rows.is_empty() { // Get columns from the statement if possible. `prepare`'s own @@ -445,14 +465,27 @@ async fn exec_query_on_client( // statement prepared and executed across separate Syncs, the same // pooler-unsafe pattern this PR eliminates elsewhere. That path is // a tokio-postgres 0.7 limitation (see the module note in - // `client.rs`); this `prepare` is the only remaining named- - // statement call, used only here for column discovery on empty - // result sets. - let columns: Vec = if let Ok(stmt) = pg_client.prepare(&final_query).await { - stmt.columns() - .iter() - .map(|c| c.name().to_string()) - .collect() + // `client.rs`). + // + // On a transaction-mode pooler connection, skip `prepare()` + // entirely and return `columns: []` — the host handles empty + // columns gracefully, and avoiding `prepare()` eliminates the + // plugin's own named-statement trigger (#157, tabularis#1097). + // The remaining trigger (tokio-postgres's internal + // `typeinfo_statement` during `Row` deserialization on the + // non-empty-rows path) is handled by the error-message rewrite + // above + `max_prepared_statements` on the pooler side. On direct + // connections, `prepare()` is still used (it's safe there and + // gives better column discovery for empty result sets). + let columns: Vec = if !is_transaction_pooler { + if let Ok(stmt) = pg_client.prepare(&final_query).await { + stmt.columns() + .iter() + .map(|c| c.name().to_string()) + .collect() + } else { + vec![] + } } else { vec![] }; diff --git a/src/models.rs b/src/models.rs index 2f3ab82..a2a9937 100644 --- a/src/models.rs +++ b/src/models.rs @@ -20,6 +20,16 @@ pub struct ConnectionParams { pub ssl_key: Option, pub connection_string: Option, pub startup_script: Option, + /// Connection pooler mode, set by the host when the user declares the + /// connection goes through a transaction-mode pooler (e.g. PgBouncer + /// `pool_mode = transaction`). When `"transaction"`, the plugin avoids + /// its own named-statement `prepare()` calls (which split across two + /// Syncs and can be orphaned by pooler reassignment) and surfaces + /// actionable error messages for tokio-postgres's internal + /// `typeinfo_statement` (which requires `max_prepared_statements > 0` + /// on the pooler side for non-builtin column types — see #157, + /// tabularis#1097). `None` for direct connections — no behavior change. + pub pooler_mode: Option, } impl ConnectionParams { @@ -48,8 +58,20 @@ impl ConnectionParams { ssl_key: get_str("ssl_key"), connection_string: get_str("connection_string"), startup_script: get_str("startup_script"), + pooler_mode: get_str("pooler_mode"), } } + + /// True when the connection goes through a transaction-mode pooler + /// (e.g. PgBouncer `pool_mode = transaction`). Drives pooler-safe + /// behavior: the plugin avoids its own named-statement `prepare()` calls + /// and surfaces actionable error messages for tokio-postgres's internal + /// `typeinfo_statement` (#157, tabularis#1097). + pub fn is_transaction_pooler(&self) -> bool { + self.pooler_mode + .as_deref() + .is_some_and(|m| m.eq_ignore_ascii_case("transaction")) + } } /// Extract the nested `params` object every RPC method receives. @@ -82,3 +104,7 @@ pub struct RoutineCallArg { #[serde(default)] pub is_raw: bool, } + +#[cfg(test)] +#[path = "models_tests.rs"] +mod tests; diff --git a/src/models_tests.rs b/src/models_tests.rs new file mode 100644 index 0000000..b76d0a9 --- /dev/null +++ b/src/models_tests.rs @@ -0,0 +1,74 @@ +//! Unit tests for `ConnectionParams::is_transaction_pooler` and the +//! `pooler_mode` field parsing (#157). + +use crate::models::ConnectionParams; +use serde_json::json; + +fn params_with_pooler_mode(mode: Option<&str>) -> ConnectionParams { + ConnectionParams::from_value(&json!({ + "driver": "postgres", + "host": "localhost", + "port": 5432, + "database": "db", + "username": "user", + "pooler_mode": mode, + })) +} + +#[test] +fn is_transaction_pooler_true_for_transaction() { + assert!(params_with_pooler_mode(Some("transaction")).is_transaction_pooler()); +} + +#[test] +fn is_transaction_pooler_false_for_none() { + let params = ConnectionParams::from_value(&json!({ + "driver": "postgres", + "host": "localhost", + "port": 5432, + "database": "db", + "username": "user", + })); + assert!(!params.is_transaction_pooler()); +} + +#[test] +fn is_transaction_pooler_false_for_session() { + assert!(!params_with_pooler_mode(Some("session")).is_transaction_pooler()); +} + +#[test] +fn is_transaction_pooler_false_for_empty_string() { + assert!(!params_with_pooler_mode(Some("")).is_transaction_pooler()); +} + +#[test] +fn is_transaction_pooler_false_for_typo() { + assert!(!params_with_pooler_mode(Some("transation")).is_transaction_pooler()); +} + +/// Case-insensitive: the host might send "Transaction" or "TRANSACTION" +/// from a differently-cased dropdown label or config file. The comparison +/// must not silently disable pooler-safe behavior on a valid-but-cased +/// value (boundary-and-edge-cases finding from the #158 ensemble review). +#[test] +fn is_transaction_pooler_case_insensitive() { + assert!(params_with_pooler_mode(Some("Transaction")).is_transaction_pooler()); + assert!(params_with_pooler_mode(Some("TRANSACTION")).is_transaction_pooler()); + assert!(params_with_pooler_mode(Some("Transaction")).is_transaction_pooler()); +} + +/// Non-string JSON for `pooler_mode` (number, bool, null) is silently +/// treated as None — not a transaction pooler. +#[test] +fn is_transaction_pooler_false_for_non_string_json() { + let params = ConnectionParams::from_value(&json!({ + "driver": "postgres", + "host": "localhost", + "port": 5432, + "database": "db", + "username": "user", + "pooler_mode": 42, + })); + assert!(!params.is_transaction_pooler()); +}