From ba5f47bc6980136d9fe83a261f200ae5cbe559b4 Mon Sep 17 00:00:00 2001 From: Will Hopkins Date: Sun, 26 Jul 2026 11:16:48 -0700 Subject: [PATCH 01/21] fix(2pc): require GID prefix when parsing transaction IDs --- .../client/query_engine/two_pc/transaction.rs | 15 +++++++++++++++ 1 file changed, 15 insertions(+) diff --git a/pgdog/src/frontend/client/query_engine/two_pc/transaction.rs b/pgdog/src/frontend/client/query_engine/two_pc/transaction.rs index 4466fa650..4cd76fc1d 100644 --- a/pgdog/src/frontend/client/query_engine/two_pc/transaction.rs +++ b/pgdog/src/frontend/client/query_engine/two_pc/transaction.rs @@ -40,7 +40,15 @@ impl Display for TwoPcTransaction { impl FromStr for TwoPcTransaction { type Err = (); + /// Parse a transaction from a GID generated by any PgDog process. + /// The GID must carry the [`PREFIX`] marker; only the trailing + /// numeric ID is significant because the rest of the prefix embeds + /// process identity, which differs between instances and restarts. fn from_str(s: &str) -> Result { + if !s.starts_with(PREFIX) { + return Err(()); + } + let id = s.rsplit("_").next().map(|id| id.parse()); if let Some(Ok(id)) = id { @@ -65,6 +73,13 @@ mod test { assert_eq!(reverse.0, transaction.0); } + #[test] + fn test_reject_gid_without_prefix() { + assert!(TwoPcTransaction::from_str("123").is_err()); + assert!(TwoPcTransaction::from_str("app_txn_123").is_err()); + assert!(TwoPcTransaction::from_str("__pgdog_123").is_err()); + } + #[test] fn test_instance_id() { for id in [1024, 11111111, usize::MAX, usize::MIN] { From 398c824b43b2f0670f78c1764c76701d3fedaf0c Mon Sep 17 00:00:00 2001 From: Will Hopkins Date: Sun, 26 Jul 2026 11:30:38 -0700 Subject: [PATCH 02/21] fix(2pc): clean up prepared transactions by durable ID after restart --- pgdog/src/backend/pool/connection/binding.rs | 61 ++++++++++++++++- .../client/query_engine/two_pc/manager.rs | 7 +- .../client/query_engine/two_pc/statement.rs | 61 ++++++++++++++++- .../client/query_engine/two_pc/test.rs | 66 +++++++++++++++++++ 4 files changed, 189 insertions(+), 6 deletions(-) diff --git a/pgdog/src/backend/pool/connection/binding.rs b/pgdog/src/backend/pool/connection/binding.rs index a762b248d..302bdd57c 100644 --- a/pgdog/src/backend/pool/connection/binding.rs +++ b/pgdog/src/backend/pool/connection/binding.rs @@ -5,10 +5,13 @@ use crate::{ ClientRequest, client::query_engine::{ TwoPcPhase, - two_pc::{TwoPcTransaction, statement::phase_control}, + two_pc::{ + TwoPcTransaction, TwoPcTransactionOnShard, + statement::{phase_control, phase_control_gid}, + }, }, }, - net::{FrontendPid, ProtocolMessage, Query, parameter::Parameters}, + net::{DataRow, FrontendPid, ProtocolMessage, Query, parameter::Parameters}, state::State, }; @@ -411,6 +414,60 @@ impl Binding { } } + /// Resolve a two-phase transaction whose exact GIDs may have been + /// created by another PgDog process, e.g. during crash recovery. + /// + /// GID prefixes embed the identity of the process that created the + /// transaction, which changes across restarts, so each shard is + /// scanned for prepared transactions matching the durable numeric + /// transaction ID and shard index, and the phase statement runs + /// against the exact GIDs found. A shard with no matching GID is + /// already resolved and is skipped. + pub(crate) async fn two_pc_cleanup( + &mut self, + transaction: TwoPcTransaction, + phase: TwoPcPhase, + ) -> Result<(), Error> { + match self { + Binding::MultiShard(servers, _) => { + let futures = servers + .iter_mut() + .enumerate() + .map(|(shard, server)| async move { + let target = TwoPcTransactionOnShard::new(transaction, shard); + let rows: Vec = server + .fetch_all( + "SELECT gid FROM pg_prepared_xacts WHERE database = current_database()", + ) + .await?; + + for row in rows { + let Some(gid) = row.get_text(0) else { + continue; + }; + if !target.matches_gid(&gid) { + continue; + } + server.execute(phase_control_gid(&gid, phase)).await?; + if phase == TwoPcPhase::Phase2 { + server.stats_mut().transaction_2pc(); + } + } + + Ok::<(), Error>(()) + }); + + for result in join_all(futures).await { + result?; + } + + Ok(()) + } + + _ => Err(Error::TwoPcMultiShardOnly), + } + } + /// Link client to server. pub async fn link_client( &mut self, diff --git a/pgdog/src/frontend/client/query_engine/two_pc/manager.rs b/pgdog/src/frontend/client/query_engine/two_pc/manager.rs index ff22ffb23..3339fc70c 100644 --- a/pgdog/src/frontend/client/query_engine/two_pc/manager.rs +++ b/pgdog/src/frontend/client/query_engine/two_pc/manager.rs @@ -345,6 +345,11 @@ impl Manager { } /// Reconnect to cluster if available and rollback the two-phase transaction. + /// + /// The transaction may have been created by an earlier PgDog + /// process, whose GID prefix can't be reconstructed, so cleanup + /// scans each shard for prepared transactions matching the durable + /// numeric transaction ID and acts on the exact GIDs found. async fn cleanup_phase(&self, transaction: TwoPcTransaction) -> Result<(), Error> { let state = match self.inner.lock().transactions.get(&transaction).cloned() { Some(state) => state, @@ -379,7 +384,7 @@ impl Manager { &Route::write(ShardWithPriority::new_override_transaction(Shard::All)), ) .await?; - connection.two_pc(transaction, phase).await?; + connection.two_pc_cleanup(transaction, phase).await?; connection.disconnect(); Ok(()) diff --git a/pgdog/src/frontend/client/query_engine/two_pc/statement.rs b/pgdog/src/frontend/client/query_engine/two_pc/statement.rs index 9c59824e4..0017fe4e4 100644 --- a/pgdog/src/frontend/client/query_engine/two_pc/statement.rs +++ b/pgdog/src/frontend/client/query_engine/two_pc/statement.rs @@ -24,6 +24,18 @@ impl TwoPcTransactionOnShard { pub(crate) fn transaction(&self) -> TwoPcTransaction { self.transaction } + + /// Whether `gid`, as listed in `pg_prepared_xacts`, refers to this + /// transaction on this shard. GID prefixes embed the identity of the + /// PgDog process that created the transaction, which changes across + /// restarts, so matching uses the durable numeric transaction ID and + /// the shard index. + pub(crate) fn matches_gid(&self, gid: &str) -> bool { + match gid.parse::() { + Ok(parsed) => parsed.transaction == self.transaction && parsed.shard == self.shard, + Err(()) => false, + } + } } impl Display for TwoPcTransactionOnShard { @@ -54,10 +66,19 @@ pub(crate) fn phase_control( ) -> String { let txn = TwoPcTransactionOnShard::new(transaction, shard); + phase_control_gid(&txn.to_string(), phase) +} + +/// Build `PREPARE TRANSACTION`, `COMMIT PREPARED`, or `ROLLBACK PREPARED` +/// for a prepared transaction identified by its exact GID. The GID is +/// embedded as a quoted literal with any single quotes doubled. +pub(crate) fn phase_control_gid(gid: &str, phase: TwoPcPhase) -> String { + let gid = gid.replace('\'', "''"); + match phase { - TwoPcPhase::Phase1 => format!("PREPARE TRANSACTION '{txn}'"), - TwoPcPhase::Phase2 => format!("COMMIT PREPARED '{txn}'"), - TwoPcPhase::Rollback => format!("ROLLBACK PREPARED '{txn}'"), + TwoPcPhase::Phase1 => format!("PREPARE TRANSACTION '{gid}'"), + TwoPcPhase::Phase2 => format!("COMMIT PREPARED '{gid}'"), + TwoPcPhase::Rollback => format!("ROLLBACK PREPARED '{gid}'"), } } @@ -101,6 +122,40 @@ mod test { ); } + #[test] + fn matches_gid_ignores_prefix() { + let transaction: TwoPcTransaction = "__pgdog_2pc_123".parse().unwrap(); + let target = TwoPcTransactionOnShard::new(transaction, 1); + + // Rendered by this process. + assert!(target.matches_gid(&target.to_string())); + // Rendered by a process with a different instance ID. + assert!(target.matches_gid("__pgdog_2pc_deadbeef_123_1")); + // Rendered by a process with a deployment ID. + assert!(target.matches_gid("__pgdog_2pc_prod_deadbeef_123_1")); + + // Wrong shard. + assert!(!target.matches_gid("__pgdog_2pc_deadbeef_123_0")); + // Wrong transaction ID. + assert!(!target.matches_gid("__pgdog_2pc_deadbeef_9123_1")); + // Prepared transactions from other applications. + assert!(!target.matches_gid("app_txn_123_1")); + assert!(!target.matches_gid("123_1")); + assert!(!target.matches_gid("")); + } + + #[test] + fn phase_control_gid_escapes_quotes() { + assert_eq!( + phase_control_gid("it's", TwoPcPhase::Rollback), + "ROLLBACK PREPARED 'it''s'" + ); + assert_eq!( + phase_control_gid("__pgdog_2pc_a_1_0", TwoPcPhase::Phase2), + "COMMIT PREPARED '__pgdog_2pc_a_1_0'" + ); + } + #[test] fn phase_control_statements() { let transaction = TwoPcTransaction::new(); diff --git a/pgdog/src/frontend/client/query_engine/two_pc/test.rs b/pgdog/src/frontend/client/query_engine/two_pc/test.rs index 0ad10f27d..8cef69d6d 100644 --- a/pgdog/src/frontend/client/query_engine/two_pc/test.rs +++ b/pgdog/src/frontend/client/query_engine/two_pc/test.rs @@ -83,6 +83,72 @@ async fn test_cleanup_transaction_phase_one() { ); } +#[tokio::test] +async fn test_cleanup_transaction_foreign_prefix() { + config::load_test(); + logger(); + let cluster = databases().all().iter().next().unwrap().1.clone(); + + // A transaction restored from the WAL after a restart. The GIDs on + // the shards carry the previous process's prefix, which the current + // process can't reconstruct. + let transaction: TwoPcTransaction = "__pgdog_2pc_314159265358979".parse().unwrap(); + + let mut conn = Connection::new(cluster.user(), cluster.name(), false).unwrap(); + conn.connect( + &Request::default(), + &Route::write(ShardWithPriority::new_default_unset(Shard::All)), + ) + .await + .unwrap(); + conn.execute("BEGIN").await.unwrap(); + conn.execute("CREATE TABLE test_cleanup_foreign_prefix(id BIGINT)") + .await + .unwrap(); + conn.execute("PREPARE TRANSACTION '__pgdog_2pc_previousinstance_314159265358979_0'") + .await + .unwrap(); + conn.disconnect(); + + let manager = Manager::get(); + manager.restore_transaction( + transaction, + cluster.user().to_string(), + cluster.name().to_string(), + TwoPcPhase::Phase1, + ); + + manager.wait_until_cleaned_up(transaction).await; + assert!(manager.transaction(&transaction).is_none()); + + conn.connect( + &Request::default(), + &Route::write(ShardWithPriority::new_default_unset(Shard::All)), + ) + .await + .unwrap(); + + let leftover = conn + .execute( + "SELECT gid FROM pg_prepared_xacts \ + WHERE gid LIKE '__pgdog_2pc_previousinstance_%'", + ) + .await + .unwrap(); + assert!( + leftover.iter().find(|p| p.code() == 'D').is_none(), + "prepared transactions with the previous process's prefix were not cleaned up" + ); + + // Phase 1 transactions are rolled back: the table doesn't exist. + let table = conn + .execute("SELECT * FROM test_cleanup_foreign_prefix") + .await + .err() + .unwrap(); + assert!(table.to_string().contains("does not exist")); +} + #[tokio::test] async fn test_cleanup_transaction_phase_two() { config::load_test(); From 56f7469155a4b1dcab14375a4dd8bb61b38a34a5 Mon Sep 17 00:00:00 2001 From: Will Hopkins Date: Sun, 26 Jul 2026 11:50:31 -0700 Subject: [PATCH 03/21] fix(integration): build 2pc wal_helper with pgdog default features --- integration/two_pc_crash_safety/wal_helper/Cargo.toml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/integration/two_pc_crash_safety/wal_helper/Cargo.toml b/integration/two_pc_crash_safety/wal_helper/Cargo.toml index 9ec8096b4..57c73732a 100644 --- a/integration/two_pc_crash_safety/wal_helper/Cargo.toml +++ b/integration/two_pc_crash_safety/wal_helper/Cargo.toml @@ -10,5 +10,5 @@ path = "src/main.rs" [dependencies] bytes = "1" -pgdog = { path = "../../../pgdog", default-features = false } +pgdog = { path = "../../../pgdog" } tokio = { version = "1", features = ["macros", "rt-multi-thread"] } From c676bbe664e63db3352afe1ab025b32328037299 Mon Sep 17 00:00:00 2001 From: Will Hopkins Date: Mon, 27 Jul 2026 12:33:08 -0700 Subject: [PATCH 04/21] feat: validate NODE_ID and DEPLOYMENT_ID at startup --- pgdog/src/main.rs | 3 ++ pgdog/src/util.rs | 91 ++++++++++++++++++++++++++++++++++++++++++++--- 2 files changed, 89 insertions(+), 5 deletions(-) diff --git a/pgdog/src/main.rs b/pgdog/src/main.rs index 684c87e55..980c05a32 100644 --- a/pgdog/src/main.rs +++ b/pgdog/src/main.rs @@ -105,6 +105,9 @@ fn main() -> Result<(), Box> { } async fn pgdog(command: Option) -> Result<(), Box> { + // Fail fast on identifiers that would produce malformed 2pc GIDs. + pgdog::util::validate_instance_identity()?; + // Run atexit handlers on SIGTERM (e.g. llvm-cov profile flushing). #[cfg(unix)] install_sigterm_handler(); diff --git a/pgdog/src/util.rs b/pgdog/src/util.rs index 8ea220448..6d8003682 100644 --- a/pgdog/src/util.rs +++ b/pgdog/src/util.rs @@ -116,10 +116,9 @@ pub fn random_string(n: usize) -> String { } // Generate a unique 8-character hex instance ID on first access -static INSTANCE_ID: Lazy = Lazy::new(|| { - if let Ok(node_id) = env::var("NODE_ID") { - node_id - } else { +static INSTANCE_ID: Lazy = Lazy::new(|| match env::var("NODE_ID") { + Ok(node_id) if !node_id.is_empty() => node_id, + _ => { let mut rng = rand::rng(); (0..8) .map(|_| { @@ -132,10 +131,52 @@ static INSTANCE_ID: Lazy = Lazy::new(|| { /// Get the instance ID for this pgdog instance. /// This is generated once at startup and persists for the lifetime of the process. +/// An unset or empty `NODE_ID` auto-generates a random ID; otherwise the value +/// is restricted to [`safe_identifier`] characters, which +/// [`validate_instance_identity`] enforces at startup. pub fn instance_id() -> &'static str { &INSTANCE_ID } +/// True if `s` contains only ASCII alphanumerics, underscores and hyphens: +/// the alphabet PgDog allows in instance identifiers and, by extension, in +/// the two-phase commit GIDs they are embedded in. +pub fn safe_identifier(s: &str) -> bool { + s.bytes() + .all(|b| b.is_ascii_alphanumeric() || b == b'_' || b == b'-') +} + +/// Environment-provided instance identifier rejected by +/// [`validate_instance_identity`]. +#[derive(Debug, thiserror::Error)] +#[error( + "{var} may only contain ASCII letters, digits, '_' and '-' \ + (it is embedded in two-phase commit transaction identifiers), got: {value:?}" +)] +pub struct IdentityError { + pub var: &'static str, + pub value: String, +} + +/// Validate the environment-provided identifiers that end up embedded in +/// two-phase commit GIDs. Called at startup so a misconfigured `NODE_ID` +/// or `DEPLOYMENT_ID` fails fast; GIDs built from validated identifiers +/// contain no characters that need quoting, which cleanup relies on when +/// it embeds recovered GIDs into transaction control statements. +/// +/// Unset and empty values are valid: both mean the identifier is absent. +pub fn validate_instance_identity() -> Result<(), IdentityError> { + for var in ["NODE_ID", "DEPLOYMENT_ID"] { + if let Ok(value) = env::var(var) + && !safe_identifier(&value) + { + return Err(IdentityError { var, value }); + } + } + + Ok(()) +} + /// Get an externally assigned, unique, node identifier /// for this instance of PgDog. /// @@ -148,12 +189,16 @@ pub fn node_id() -> Result { instance_id().split("-").last().unwrap().parse() } -static DEPLOYMENT_ID: Lazy> = Lazy::new(|| env::var("DEPLOYMENT_ID").ok()); +static DEPLOYMENT_ID: Lazy> = + Lazy::new(|| env::var("DEPLOYMENT_ID").ok().filter(|id| !id.is_empty())); /// Get the ID of this PgDog deployment. /// /// This should be _globally_ unique /// and is used to differentiate 2pc transactions. +/// An unset or empty `DEPLOYMENT_ID` means no deployment ID; otherwise the +/// value is restricted to [`safe_identifier`] characters, which +/// [`validate_instance_identity`] enforces at startup. /// pub(crate) fn deployment_id() -> Option<&'static str> { DEPLOYMENT_ID.as_deref() @@ -341,6 +386,42 @@ mod test { use super::*; use crate::test_utils::*; + #[test] + fn test_safe_identifier() { + assert!(safe_identifier("pgdog-node-1")); + assert!(safe_identifier("prod_us_east_2")); + assert!(safe_identifier("deadbeef")); + assert!(!safe_identifier("it's")); + assert!(!safe_identifier("node\\1")); + assert!(!safe_identifier("node 1")); + assert!(!safe_identifier("nöde")); + } + + #[test] + fn test_validate_instance_identity() { + let _node = set_env_var("NODE_ID", "pgdog-node-1"); + let _deployment = set_env_var("DEPLOYMENT_ID", "prod"); + assert!(validate_instance_identity().is_ok()); + + { + let _bad = set_env_var("NODE_ID", "it's"); + assert!(matches!( + validate_instance_identity(), + Err(IdentityError { var: "NODE_ID", .. }) + )); + } + + // Empty means absent, like unset. + let _empty = set_env_var("NODE_ID", ""); + assert!(validate_instance_identity().is_ok()); + } + + #[test] + fn test_empty_deployment_id_is_absent() { + let _empty = set_env_var("DEPLOYMENT_ID", ""); + assert_eq!(deployment_id(), None); + } + #[test] fn test_human_duration() { assert_eq!(human_duration(Duration::from_millis(500)), "500ms"); From 09049bc520e49efa1a7eb642da57a85d566241ef Mon Sep 17 00:00:00 2001 From: Will Hopkins Date: Mon, 27 Jul 2026 12:33:59 -0700 Subject: [PATCH 05/21] fix(2pc): refuse to resolve prepared transactions outside the GID alphabet --- .../client/query_engine/two_pc/statement.rs | 28 ++++++++++++++++++- 1 file changed, 27 insertions(+), 1 deletion(-) diff --git a/pgdog/src/frontend/client/query_engine/two_pc/statement.rs b/pgdog/src/frontend/client/query_engine/two_pc/statement.rs index 0017fe4e4..66dc182ab 100644 --- a/pgdog/src/frontend/client/query_engine/two_pc/statement.rs +++ b/pgdog/src/frontend/client/query_engine/two_pc/statement.rs @@ -2,7 +2,10 @@ use std::{fmt::Display, str::FromStr}; +use tracing::warn; + use crate::frontend::client::query_engine::two_pc::TwoPcTransaction; +use crate::util::safe_identifier; use super::TwoPcPhase; @@ -30,11 +33,29 @@ impl TwoPcTransactionOnShard { /// PgDog process that created the transaction, which changes across /// restarts, so matching uses the durable numeric transaction ID and /// the shard index. + /// + /// A matched GID is guaranteed to contain only [`safe_identifier`] + /// characters, so it can be embedded verbatim in a quoted SQL literal. + /// PgDog only generates such GIDs (`NODE_ID` and `DEPLOYMENT_ID` are + /// validated at startup); a GID that matches the numeric ID but not + /// the alphabet was not created by PgDog and is refused with a + /// warning. pub(crate) fn matches_gid(&self, gid: &str) -> bool { - match gid.parse::() { + let matches = match gid.parse::() { Ok(parsed) => parsed.transaction == self.transaction && parsed.shard == self.shard, Err(()) => false, + }; + + if matches && !safe_identifier(gid) { + warn!( + "[2pc] prepared transaction {:?} matches transaction {} on shard {} \ + but contains characters PgDog never generates; refusing to resolve it", + gid, self.transaction, self.shard + ); + return false; } + + matches } } @@ -142,6 +163,11 @@ mod test { assert!(!target.matches_gid("app_txn_123_1")); assert!(!target.matches_gid("123_1")); assert!(!target.matches_gid("")); + // Matching ID but characters PgDog never generates: such GIDs + // are refused so they can never reach a quoted SQL literal. + assert!(!target.matches_gid("__pgdog_2pc_it's_123_1")); + assert!(!target.matches_gid("__pgdog_2pc_a\\'b_123_1")); + assert!(!target.matches_gid("__pgdog_2pc_a b_123_1")); } #[test] From a883ee25c87ab953c89804f9924b9a8ad7826b11 Mon Sep 17 00:00:00 2001 From: Will Hopkins Date: Mon, 27 Jul 2026 12:35:11 -0700 Subject: [PATCH 06/21] refactor(2pc): render cleanup statements where the scanned GID is matched --- pgdog/src/backend/pool/connection/binding.rs | 16 +++++++---- .../client/query_engine/two_pc/statement.rs | 27 +++---------------- 2 files changed, 14 insertions(+), 29 deletions(-) diff --git a/pgdog/src/backend/pool/connection/binding.rs b/pgdog/src/backend/pool/connection/binding.rs index 302bdd57c..573636368 100644 --- a/pgdog/src/backend/pool/connection/binding.rs +++ b/pgdog/src/backend/pool/connection/binding.rs @@ -5,10 +5,7 @@ use crate::{ ClientRequest, client::query_engine::{ TwoPcPhase, - two_pc::{ - TwoPcTransaction, TwoPcTransactionOnShard, - statement::{phase_control, phase_control_gid}, - }, + two_pc::{TwoPcTransaction, TwoPcTransactionOnShard, statement::phase_control}, }, }, net::{DataRow, FrontendPid, ProtocolMessage, Query, parameter::Parameters}, @@ -448,7 +445,16 @@ impl Binding { if !target.matches_gid(&gid) { continue; } - server.execute(phase_control_gid(&gid, phase)).await?; + // matches_gid guarantees the gid contains no + // characters that need quoting. + let statement = match phase { + TwoPcPhase::Phase2 => format!("COMMIT PREPARED '{gid}'"), + TwoPcPhase::Rollback => format!("ROLLBACK PREPARED '{gid}'"), + TwoPcPhase::Phase1 => { + unreachable!("cleanup resolves transactions; it never prepares") + } + }; + server.execute(statement).await?; if phase == TwoPcPhase::Phase2 { server.stats_mut().transaction_2pc(); } diff --git a/pgdog/src/frontend/client/query_engine/two_pc/statement.rs b/pgdog/src/frontend/client/query_engine/two_pc/statement.rs index 66dc182ab..872baa547 100644 --- a/pgdog/src/frontend/client/query_engine/two_pc/statement.rs +++ b/pgdog/src/frontend/client/query_engine/two_pc/statement.rs @@ -87,19 +87,10 @@ pub(crate) fn phase_control( ) -> String { let txn = TwoPcTransactionOnShard::new(transaction, shard); - phase_control_gid(&txn.to_string(), phase) -} - -/// Build `PREPARE TRANSACTION`, `COMMIT PREPARED`, or `ROLLBACK PREPARED` -/// for a prepared transaction identified by its exact GID. The GID is -/// embedded as a quoted literal with any single quotes doubled. -pub(crate) fn phase_control_gid(gid: &str, phase: TwoPcPhase) -> String { - let gid = gid.replace('\'', "''"); - match phase { - TwoPcPhase::Phase1 => format!("PREPARE TRANSACTION '{gid}'"), - TwoPcPhase::Phase2 => format!("COMMIT PREPARED '{gid}'"), - TwoPcPhase::Rollback => format!("ROLLBACK PREPARED '{gid}'"), + TwoPcPhase::Phase1 => format!("PREPARE TRANSACTION '{txn}'"), + TwoPcPhase::Phase2 => format!("COMMIT PREPARED '{txn}'"), + TwoPcPhase::Rollback => format!("ROLLBACK PREPARED '{txn}'"), } } @@ -170,18 +161,6 @@ mod test { assert!(!target.matches_gid("__pgdog_2pc_a b_123_1")); } - #[test] - fn phase_control_gid_escapes_quotes() { - assert_eq!( - phase_control_gid("it's", TwoPcPhase::Rollback), - "ROLLBACK PREPARED 'it''s'" - ); - assert_eq!( - phase_control_gid("__pgdog_2pc_a_1_0", TwoPcPhase::Phase2), - "COMMIT PREPARED '__pgdog_2pc_a_1_0'" - ); - } - #[test] fn phase_control_statements() { let transaction = TwoPcTransaction::new(); From 74703c83b5ea64ee5596aeeb3d7be3d6fcca2479 Mon Sep 17 00:00:00 2001 From: Will Hopkins Date: Mon, 27 Jul 2026 12:36:13 -0700 Subject: [PATCH 07/21] fix(2pc): leave prepared transactions the cleanup user cannot resolve to the operator --- pgdog/src/backend/pool/connection/binding.rs | 23 +++++++++++++++++--- 1 file changed, 20 insertions(+), 3 deletions(-) diff --git a/pgdog/src/backend/pool/connection/binding.rs b/pgdog/src/backend/pool/connection/binding.rs index 573636368..de72f08d4 100644 --- a/pgdog/src/backend/pool/connection/binding.rs +++ b/pgdog/src/backend/pool/connection/binding.rs @@ -13,6 +13,7 @@ use crate::{ }; use futures::future::join_all; +use tracing::warn; use super::*; @@ -454,9 +455,25 @@ impl Binding { unreachable!("cleanup resolves transactions; it never prepares") } }; - server.execute(statement).await?; - if phase == TwoPcPhase::Phase2 { - server.stats_mut().transaction_2pc(); + match server.execute(&statement[..]).await { + Ok(_) => { + if phase == TwoPcPhase::Phase2 { + server.stats_mut().transaction_2pc(); + } + } + // Insufficient privilege: the prepared transaction + // is owned by a role the configured user can't act + // for. Retrying can't succeed until an operator + // intervenes, so leave the transaction to them + // rather than retrying forever. + Err(Error::ExecutionError(err)) if err.code == "42501" => { + warn!( + "[2pc] insufficient privilege to run {}; \ + resolve the prepared transaction manually", + statement + ); + } + Err(err) => return Err(err), } } From a05ef1b5dd8c938b7409d198fa87faf419ba03dd Mon Sep 17 00:00:00 2001 From: Will Hopkins Date: Mon, 27 Jul 2026 12:37:01 -0700 Subject: [PATCH 08/21] test(2pc): use a random GID in the foreign prefix cleanup test --- .../client/query_engine/two_pc/test.rs | 27 ++++++++++--------- 1 file changed, 15 insertions(+), 12 deletions(-) diff --git a/pgdog/src/frontend/client/query_engine/two_pc/test.rs b/pgdog/src/frontend/client/query_engine/two_pc/test.rs index 8cef69d6d..372f4769c 100644 --- a/pgdog/src/frontend/client/query_engine/two_pc/test.rs +++ b/pgdog/src/frontend/client/query_engine/two_pc/test.rs @@ -91,8 +91,12 @@ async fn test_cleanup_transaction_foreign_prefix() { // A transaction restored from the WAL after a restart. The GIDs on // the shards carry the previous process's prefix, which the current - // process can't reconstruct. - let transaction: TwoPcTransaction = "__pgdog_2pc_314159265358979".parse().unwrap(); + // process can't reconstruct. The numeric ID is random so a failed + // run's leftover prepared transaction can't block reruns. + let number: u64 = rand::random(); + let transaction: TwoPcTransaction = format!("__pgdog_2pc_{number}").parse().unwrap(); + let foreign_gid = format!("__pgdog_2pc_previousinstance_{number}_0"); + let table = format!("test_cleanup_foreign_prefix_{number}"); let mut conn = Connection::new(cluster.user(), cluster.name(), false).unwrap(); conn.connect( @@ -102,10 +106,10 @@ async fn test_cleanup_transaction_foreign_prefix() { .await .unwrap(); conn.execute("BEGIN").await.unwrap(); - conn.execute("CREATE TABLE test_cleanup_foreign_prefix(id BIGINT)") + conn.execute(format!("CREATE TABLE {table}(id BIGINT)")) .await .unwrap(); - conn.execute("PREPARE TRANSACTION '__pgdog_2pc_previousinstance_314159265358979_0'") + conn.execute(format!("PREPARE TRANSACTION '{foreign_gid}'")) .await .unwrap(); conn.disconnect(); @@ -129,24 +133,23 @@ async fn test_cleanup_transaction_foreign_prefix() { .unwrap(); let leftover = conn - .execute( - "SELECT gid FROM pg_prepared_xacts \ - WHERE gid LIKE '__pgdog_2pc_previousinstance_%'", - ) + .execute(format!( + "SELECT gid FROM pg_prepared_xacts WHERE gid = '{foreign_gid}'" + )) .await .unwrap(); assert!( leftover.iter().find(|p| p.code() == 'D').is_none(), - "prepared transactions with the previous process's prefix were not cleaned up" + "prepared transaction with the previous process's prefix was not cleaned up" ); // Phase 1 transactions are rolled back: the table doesn't exist. - let table = conn - .execute("SELECT * FROM test_cleanup_foreign_prefix") + let missing = conn + .execute(format!("SELECT * FROM {table}")) .await .err() .unwrap(); - assert!(table.to_string().contains("does not exist")); + assert!(missing.to_string().contains("does not exist")); } #[tokio::test] From 35b29d1406641e169934caa65edeba2a84f181a1 Mon Sep 17 00:00:00 2001 From: Will Hopkins Date: Mon, 27 Jul 2026 13:16:34 -0700 Subject: [PATCH 09/21] feat(2pc): record the coordinator GID prefix in the WAL --- .../wal_helper/src/main.rs | 4 ++ .../client/query_engine/two_pc/manager.rs | 19 ++++++- .../client/query_engine/two_pc/test.rs | 1 + .../client/query_engine/two_pc/transaction.rs | 2 +- .../client/query_engine/two_pc/wal/record.rs | 55 +++++++++++++++++++ .../query_engine/two_pc/wal/recovery.rs | 17 +++++- .../client/query_engine/two_pc/wal/segment.rs | 4 ++ .../client/query_engine/two_pc/wal/writer.rs | 7 ++- 8 files changed, 102 insertions(+), 7 deletions(-) diff --git a/integration/two_pc_crash_safety/wal_helper/src/main.rs b/integration/two_pc_crash_safety/wal_helper/src/main.rs index 2481fe219..792db9724 100644 --- a/integration/two_pc_crash_safety/wal_helper/src/main.rs +++ b/integration/two_pc_crash_safety/wal_helper/src/main.rs @@ -33,10 +33,14 @@ async fn main() { let mut segment = Segment::create(&dir, 0).await.expect("create segment"); let mut buf = BytesMut::new(); + // The prefix is left empty to produce the record shape written by + // versions that did not store it, so recovery of such WALs (via the + // numeric-ID fallback) stays covered end to end. Record::Begin(BeginPayload { txn, user, database, + prefix: String::new(), }) .encode(&mut buf) .expect("encode begin"); diff --git a/pgdog/src/frontend/client/query_engine/two_pc/manager.rs b/pgdog/src/frontend/client/query_engine/two_pc/manager.rs index 3339fc70c..f9964faa5 100644 --- a/pgdog/src/frontend/client/query_engine/two_pc/manager.rs +++ b/pgdog/src/frontend/client/query_engine/two_pc/manager.rs @@ -205,12 +205,14 @@ impl Manager { identifier: &Arc, phase: TwoPcPhase, ) -> Result { + let prefix = TwoPcTransaction::global_prefix(); let prior = { let mut guard = self.inner.lock(); let prior = guard.transactions.get(&transaction).cloned(); let entry = guard.transactions.entry(transaction).or_default(); entry.identifier = identifier.clone(); entry.phase = phase; + entry.prefix = prefix.clone(); prior }; @@ -221,6 +223,7 @@ impl Manager { transaction, identifier.user.clone(), identifier.database.clone(), + prefix, ) .await } @@ -262,14 +265,20 @@ impl Manager { transaction: TwoPcTransaction, user: String, database: String, + prefix: String, phase: TwoPcPhase, ) { let identifier = Arc::new(User { user, database }); { let mut guard = self.inner.lock(); - guard - .transactions - .insert(transaction, TransactionInfo { phase, identifier }); + guard.transactions.insert( + transaction, + TransactionInfo { + phase, + identifier, + prefix, + }, + ); guard.queue.push_back(transaction); } self.stats.incr_recovered(); @@ -417,6 +426,10 @@ impl Manager { pub struct TransactionInfo { pub phase: TwoPcPhase, pub identifier: Arc, + /// Coordinator GID prefix the transaction's prepared names were + /// rendered with. Empty when restored from a WAL record written by + /// a version that did not store it. + pub prefix: String, } #[derive(Default, Debug)] diff --git a/pgdog/src/frontend/client/query_engine/two_pc/test.rs b/pgdog/src/frontend/client/query_engine/two_pc/test.rs index 372f4769c..228139939 100644 --- a/pgdog/src/frontend/client/query_engine/two_pc/test.rs +++ b/pgdog/src/frontend/client/query_engine/two_pc/test.rs @@ -119,6 +119,7 @@ async fn test_cleanup_transaction_foreign_prefix() { transaction, cluster.user().to_string(), cluster.name().to_string(), + "__pgdog_2pc_previousinstance_".to_string(), TwoPcPhase::Phase1, ); diff --git a/pgdog/src/frontend/client/query_engine/two_pc/transaction.rs b/pgdog/src/frontend/client/query_engine/two_pc/transaction.rs index 4cd76fc1d..24c557881 100644 --- a/pgdog/src/frontend/client/query_engine/two_pc/transaction.rs +++ b/pgdog/src/frontend/client/query_engine/two_pc/transaction.rs @@ -18,7 +18,7 @@ impl TwoPcTransaction { /// A prefix to identify two-phase commit transactions generated /// by this PgDog process. - fn global_prefix() -> String { + pub(super) fn global_prefix() -> String { format!( "{PREFIX}{}{}_", if let Some(cluster_id) = deployment_id() { diff --git a/pgdog/src/frontend/client/query_engine/two_pc/wal/record.rs b/pgdog/src/frontend/client/query_engine/two_pc/wal/record.rs index ead80e1c9..1eb51c135 100644 --- a/pgdog/src/frontend/client/query_engine/two_pc/wal/record.rs +++ b/pgdog/src/frontend/client/query_engine/two_pc/wal/record.rs @@ -96,6 +96,12 @@ pub struct BeginPayload { pub txn: TwoPcTransaction, pub user: String, pub database: String, + /// Coordinator GID prefix the transaction's prepared names are + /// rendered with. Recovery combines it with `txn` to reconstruct + /// the exact GIDs on the shards. Empty when the record was written + /// by a version that did not store it. + #[serde(default)] + pub prefix: String, } /// Payload for records that carry only a transaction id @@ -118,6 +124,11 @@ pub struct CheckpointEntry { pub database: String, /// `true` iff a [`Record::Committing`] had been fsynced for this txn. pub decided: bool, + /// Coordinator GID prefix, carried over from the transaction's + /// [`BeginPayload`]. Empty when the originating record did not + /// store it. + #[serde(default)] + pub prefix: String, } /// A successfully decoded record and the number of bytes it consumed. @@ -227,9 +238,50 @@ mod tests { txn: TwoPcTransaction::new(), user: "alice".into(), database: "shop".into(), + prefix: "__pgdog_2pc_prod_deadbeef_".into(), })); } + #[test] + fn begin_without_prefix_decodes_with_empty_prefix() { + // The on-disk shape written by versions that predate the + // `prefix` field: same tag, payload lacking the key. + #[derive(Serialize)] + struct OldBeginPayload { + txn: TwoPcTransaction, + user: String, + database: String, + } + + let txn = TwoPcTransaction::new(); + let payload = rmp_serde::to_vec_named(&OldBeginPayload { + txn, + user: "alice".into(), + database: "shop".into(), + }) + .unwrap(); + + let mut buf = Vec::new(); + let tag = Tag::Begin as u8; + let mut crc = crc32c::crc32c(&[tag]); + crc = crc32c::crc32c_append(crc, &payload); + buf.put_u32_le(1 + payload.len() as u32); + buf.put_u32_le(crc); + buf.put_u8(tag); + buf.put_slice(&payload); + + let decoded = Record::decode(&buf).unwrap().unwrap(); + assert_eq!( + decoded.record, + Record::Begin(BeginPayload { + txn, + user: "alice".into(), + database: "shop".into(), + prefix: String::new(), + }) + ); + } + #[test] fn round_trip_committing() { round_trip(Record::Committing(TxnPayload { @@ -253,12 +305,14 @@ mod tests { user: "u1".into(), database: "d1".into(), decided: false, + prefix: String::new(), }, CheckpointEntry { txn: TwoPcTransaction::new(), user: "u2".into(), database: "d2".into(), decided: true, + prefix: "__pgdog_2pc_prod_a_".into(), }, ], })); @@ -313,6 +367,7 @@ mod tests { txn: TwoPcTransaction::new(), user: "u".into(), database: "d".into(), + prefix: "__pgdog_2pc_".into(), }); let b = Record::Committing(TxnPayload { txn: TwoPcTransaction::new(), diff --git a/pgdog/src/frontend/client/query_engine/two_pc/wal/recovery.rs b/pgdog/src/frontend/client/query_engine/two_pc/wal/recovery.rs index 5610fca02..1967dd465 100644 --- a/pgdog/src/frontend/client/query_engine/two_pc/wal/recovery.rs +++ b/pgdog/src/frontend/client/query_engine/two_pc/wal/recovery.rs @@ -26,6 +26,9 @@ struct Entry { user: String, database: String, decided: bool, + /// Coordinator GID prefix from the Begin record; empty for records + /// written by versions that did not store it. + prefix: String, } /// What [`recover_transactions`] hands back to the WAL setup path. @@ -129,9 +132,10 @@ pub(super) async fn recover_transactions( user: entry.user.clone(), database: entry.database.clone(), decided: entry.decided, + prefix: entry.prefix.clone(), }, ); - manager.restore_transaction(txn, entry.user, entry.database, phase); + manager.restore_transaction(txn, entry.user, entry.database, entry.prefix, phase); } if corruption { tracing::warn!( @@ -184,6 +188,7 @@ fn apply(working: &mut HashMap, record: Record) { user: p.user, database: p.database, decided: false, + prefix: p.prefix, }, ); } @@ -202,6 +207,7 @@ fn apply(working: &mut HashMap, record: Record) { user, database, decided, + prefix, .. } in p.active { @@ -211,6 +217,7 @@ fn apply(working: &mut HashMap, record: Record) { user, database, decided, + prefix, }, ); } @@ -238,11 +245,13 @@ mod tests { txn: txn(1), user: "alice".into(), database: "shop".into(), + prefix: "__pgdog_2pc_a_".into(), }), ); let entry = w.get(&txn(1)).unwrap(); assert_eq!(entry.user, "alice"); assert_eq!(entry.database, "shop"); + assert_eq!(entry.prefix, "__pgdog_2pc_a_"); assert!(!entry.decided); } @@ -255,6 +264,7 @@ mod tests { txn: txn(1), user: "u".into(), database: "d".into(), + prefix: "__pgdog_2pc_a_".into(), }), ); apply(&mut w, Record::Committing(TxnPayload { txn: txn(1) })); @@ -277,6 +287,7 @@ mod tests { txn: txn(1), user: "u".into(), database: "d".into(), + prefix: "__pgdog_2pc_a_".into(), }), ); apply(&mut w, Record::End(TxnPayload { txn: txn(1) })); @@ -292,6 +303,7 @@ mod tests { txn: txn(1), user: "u1".into(), database: "d1".into(), + prefix: "__pgdog_2pc_a_".into(), }), ); apply( @@ -300,6 +312,7 @@ mod tests { txn: txn(2), user: "u2".into(), database: "d2".into(), + prefix: "__pgdog_2pc_a_".into(), }), ); apply( @@ -310,12 +323,14 @@ mod tests { user: "u99".into(), database: "d99".into(), decided: true, + prefix: "__pgdog_2pc_b_".into(), }], }), ); assert_eq!(w.len(), 1); let entry = w.get(&txn(99)).unwrap(); assert_eq!(entry.user, "u99"); + assert_eq!(entry.prefix, "__pgdog_2pc_b_"); assert!(entry.decided); } } diff --git a/pgdog/src/frontend/client/query_engine/two_pc/wal/segment.rs b/pgdog/src/frontend/client/query_engine/two_pc/wal/segment.rs index d799380b3..096202347 100644 --- a/pgdog/src/frontend/client/query_engine/two_pc/wal/segment.rs +++ b/pgdog/src/frontend/client/query_engine/two_pc/wal/segment.rs @@ -410,6 +410,7 @@ mod tests { txn: TwoPcTransaction::new(), user: "u".into(), database: "d".into(), + prefix: "__pgdog_2pc_".into(), }); let r2 = Record::End(TxnPayload { txn: TwoPcTransaction::new(), @@ -442,6 +443,7 @@ mod tests { txn: TwoPcTransaction::new(), user: "u".into(), database: "d".into(), + prefix: "__pgdog_2pc_".into(), }) .encode(&mut buf) .unwrap(); @@ -489,6 +491,7 @@ mod tests { txn: TwoPcTransaction::new(), user: "u".into(), database: "d".into(), + prefix: "__pgdog_2pc_".into(), }); let mut buf = BytesMut::new(); r1.encode(&mut buf).unwrap(); @@ -556,6 +559,7 @@ mod tests { txn: TwoPcTransaction::new(), user: payload.clone(), database: "d".into(), + prefix: "__pgdog_2pc_".into(), }); r.encode(&mut buf).unwrap(); expected.push(r); diff --git a/pgdog/src/frontend/client/query_engine/two_pc/wal/writer.rs b/pgdog/src/frontend/client/query_engine/two_pc/wal/writer.rs index ed80e6c0f..1b1e0c7e3 100644 --- a/pgdog/src/frontend/client/query_engine/two_pc/wal/writer.rs +++ b/pgdog/src/frontend/client/query_engine/two_pc/wal/writer.rs @@ -181,18 +181,20 @@ impl Wal { } /// Log that `txn` is about to issue PREPARE TRANSACTION on its - /// participants. Must complete before any PREPARE leaves the - /// coordinator. + /// participants under GIDs rendered with `prefix`. Must complete + /// before any PREPARE leaves the coordinator. pub async fn append_begin( &self, txn: TwoPcTransaction, user: String, database: String, + prefix: String, ) -> Result> { self.append(Record::Begin(BeginPayload { txn, user, database, + prefix, })) .await } @@ -455,6 +457,7 @@ fn apply_to_snapshot( user: p.user.clone(), database: p.database.clone(), decided: false, + prefix: p.prefix.clone(), }, ); undo.push(Undo { txn: p.txn, prior }); From 6f2d57536dce8d860edf0ac1e9c9dd332cb62d43 Mon Sep 17 00:00:00 2001 From: Will Hopkins Date: Mon, 27 Jul 2026 13:19:38 -0700 Subject: [PATCH 10/21] fix(2pc): resolve recovered transactions by their recorded GID --- pgdog/src/backend/pool/connection/binding.rs | 50 +++++-- .../client/query_engine/two_pc/manager.rs | 12 +- .../client/query_engine/two_pc/statement.rs | 26 +++- .../client/query_engine/two_pc/test.rs | 127 +++++++++++++++++- .../client/query_engine/two_pc/transaction.rs | 5 + 5 files changed, 201 insertions(+), 19 deletions(-) diff --git a/pgdog/src/backend/pool/connection/binding.rs b/pgdog/src/backend/pool/connection/binding.rs index de72f08d4..ac266b4c8 100644 --- a/pgdog/src/backend/pool/connection/binding.rs +++ b/pgdog/src/backend/pool/connection/binding.rs @@ -10,6 +10,7 @@ use crate::{ }, net::{DataRow, FrontendPid, ProtocolMessage, Query, parameter::Parameters}, state::State, + util::safe_identifier, }; use futures::future::join_all; @@ -412,18 +413,21 @@ impl Binding { } } - /// Resolve a two-phase transaction whose exact GIDs may have been - /// created by another PgDog process, e.g. during crash recovery. + /// Resolve a two-phase transaction on every shard during cleanup + /// and crash recovery. /// - /// GID prefixes embed the identity of the process that created the - /// transaction, which changes across restarts, so each shard is - /// scanned for prepared transactions matching the durable numeric - /// transaction ID and shard index, and the phase statement runs - /// against the exact GIDs found. A shard with no matching GID is - /// already resolved and is skipped. + /// `prefix` is the coordinator GID prefix recorded when the + /// transaction was created; combined with the transaction's numeric + /// ID it names the exact prepared transaction on each shard, and the + /// phase statement runs against the GIDs found in + /// `pg_prepared_xacts`. An empty `prefix` means the transaction was + /// restored from a WAL record that did not store it; matching then + /// falls back to the durable numeric transaction ID and shard index. + /// A shard with no matching GID is already resolved and is skipped. pub(crate) async fn two_pc_cleanup( &mut self, transaction: TwoPcTransaction, + prefix: &str, phase: TwoPcPhase, ) -> Result<(), Error> { match self { @@ -433,6 +437,26 @@ impl Binding { .enumerate() .map(|(shard, server)| async move { let target = TwoPcTransactionOnShard::new(transaction, shard); + // The recorded GID is trusted to be a quotable + // literal because identifier alphabets are + // validated at startup; anything else on disk + // gets the alphabet-checked numeric fallback. + let expected = match prefix.is_empty() { + true => None, + false => { + let gid = target.gid(prefix); + if safe_identifier(&gid) { + Some(gid) + } else { + warn!( + "[2pc] recorded gid {:?} contains characters \ + PgDog never generates; matching by numeric ID", + gid + ); + None + } + } + }; let rows: Vec = server .fetch_all( "SELECT gid FROM pg_prepared_xacts WHERE database = current_database()", @@ -443,11 +467,15 @@ impl Binding { let Some(gid) = row.get_text(0) else { continue; }; - if !target.matches_gid(&gid) { + let matched = match &expected { + Some(expected) => &gid == expected, + None => target.matches_gid(&gid), + }; + if !matched { continue; } - // matches_gid guarantees the gid contains no - // characters that need quoting. + // Both match paths guarantee the gid contains + // no characters that need quoting. let statement = match phase { TwoPcPhase::Phase2 => format!("COMMIT PREPARED '{gid}'"), TwoPcPhase::Rollback => format!("ROLLBACK PREPARED '{gid}'"), diff --git a/pgdog/src/frontend/client/query_engine/two_pc/manager.rs b/pgdog/src/frontend/client/query_engine/two_pc/manager.rs index f9964faa5..914d7b5f8 100644 --- a/pgdog/src/frontend/client/query_engine/two_pc/manager.rs +++ b/pgdog/src/frontend/client/query_engine/two_pc/manager.rs @@ -356,9 +356,11 @@ impl Manager { /// Reconnect to cluster if available and rollback the two-phase transaction. /// /// The transaction may have been created by an earlier PgDog - /// process, whose GID prefix can't be reconstructed, so cleanup - /// scans each shard for prepared transactions matching the durable - /// numeric transaction ID and acts on the exact GIDs found. + /// process; the GID prefix recorded when it was created names the + /// exact prepared transactions on the shards. Cleanup scans each + /// shard's prepared transactions and acts on the GIDs that match, + /// falling back to the durable numeric transaction ID for WAL + /// records that did not store the prefix. async fn cleanup_phase(&self, transaction: TwoPcTransaction) -> Result<(), Error> { let state = match self.inner.lock().transactions.get(&transaction).cloned() { Some(state) => state, @@ -393,7 +395,9 @@ impl Manager { &Route::write(ShardWithPriority::new_override_transaction(Shard::All)), ) .await?; - connection.two_pc_cleanup(transaction, phase).await?; + connection + .two_pc_cleanup(transaction, &state.prefix, phase) + .await?; connection.disconnect(); Ok(()) diff --git a/pgdog/src/frontend/client/query_engine/two_pc/statement.rs b/pgdog/src/frontend/client/query_engine/two_pc/statement.rs index 872baa547..d673c621e 100644 --- a/pgdog/src/frontend/client/query_engine/two_pc/statement.rs +++ b/pgdog/src/frontend/client/query_engine/two_pc/statement.rs @@ -28,11 +28,21 @@ impl TwoPcTransactionOnShard { self.transaction } + /// The exact GID this transaction was prepared under on this shard, + /// rendered from the coordinator GID prefix recorded when the + /// transaction was created. + pub(crate) fn gid(&self, prefix: &str) -> String { + format!("{}{}_{}", prefix, self.transaction.number(), self.shard) + } + /// Whether `gid`, as listed in `pg_prepared_xacts`, refers to this /// transaction on this shard. GID prefixes embed the identity of the /// PgDog process that created the transaction, which changes across /// restarts, so matching uses the durable numeric transaction ID and - /// the shard index. + /// the shard index. This is the fallback for transactions restored + /// from WAL records that did not store the coordinator GID prefix; + /// with a recorded prefix, cleanup matches the exact GID from + /// [`Self::gid`] instead. /// /// A matched GID is guaranteed to contain only [`safe_identifier`] /// characters, so it can be embedded verbatim in a quoted SQL literal. @@ -161,6 +171,20 @@ mod test { assert!(!target.matches_gid("__pgdog_2pc_a b_123_1")); } + #[test] + fn gid_renders_recorded_prefix() { + let transaction: TwoPcTransaction = "__pgdog_2pc_123".parse().unwrap(); + let target = TwoPcTransactionOnShard::new(transaction, 1); + + assert_eq!( + target.gid("__pgdog_2pc_prod_deadbeef_"), + "__pgdog_2pc_prod_deadbeef_123_1" + ); + // The recorded prefix names one exact GID; the same number + // under a different prefix is a different transaction. + assert_ne!(target.gid("__pgdog_2pc_a_"), target.gid("__pgdog_2pc_b_")); + } + #[test] fn phase_control_statements() { let transaction = TwoPcTransaction::new(); diff --git a/pgdog/src/frontend/client/query_engine/two_pc/test.rs b/pgdog/src/frontend/client/query_engine/two_pc/test.rs index 228139939..3d32b0664 100644 --- a/pgdog/src/frontend/client/query_engine/two_pc/test.rs +++ b/pgdog/src/frontend/client/query_engine/two_pc/test.rs @@ -90,9 +90,9 @@ async fn test_cleanup_transaction_foreign_prefix() { let cluster = databases().all().iter().next().unwrap().1.clone(); // A transaction restored from the WAL after a restart. The GIDs on - // the shards carry the previous process's prefix, which the current - // process can't reconstruct. The numeric ID is random so a failed - // run's leftover prepared transaction can't block reruns. + // the shards carry the previous process's prefix, known here from + // the recorded WAL Begin record. The numeric ID is random so a + // failed run's leftover prepared transaction can't block reruns. let number: u64 = rand::random(); let transaction: TwoPcTransaction = format!("__pgdog_2pc_{number}").parse().unwrap(); let foreign_gid = format!("__pgdog_2pc_previousinstance_{number}_0"); @@ -153,6 +153,127 @@ async fn test_cleanup_transaction_foreign_prefix() { assert!(missing.to_string().contains("does not exist")); } +#[tokio::test] +async fn test_cleanup_transaction_legacy_record() { + config::load_test(); + logger(); + let cluster = databases().all().iter().next().unwrap().1.clone(); + + // A transaction restored from a WAL record written by a version + // that did not store the coordinator GID prefix: cleanup falls + // back to matching by the durable numeric ID. + let number: u64 = rand::random(); + let transaction: TwoPcTransaction = format!("__pgdog_2pc_{number}").parse().unwrap(); + let foreign_gid = format!("__pgdog_2pc_legacyinstance_{number}_0"); + + let mut conn = Connection::new(cluster.user(), cluster.name(), false).unwrap(); + conn.connect( + &Request::default(), + &Route::write(ShardWithPriority::new_default_unset(Shard::All)), + ) + .await + .unwrap(); + conn.execute("BEGIN").await.unwrap(); + conn.execute("SELECT 1").await.unwrap(); + conn.execute(format!("PREPARE TRANSACTION '{foreign_gid}'")) + .await + .unwrap(); + conn.disconnect(); + + let manager = Manager::get(); + manager.restore_transaction( + transaction, + cluster.user().to_string(), + cluster.name().to_string(), + String::new(), + TwoPcPhase::Phase1, + ); + + manager.wait_until_cleaned_up(transaction).await; + assert!(manager.transaction(&transaction).is_none()); + + conn.connect( + &Request::default(), + &Route::write(ShardWithPriority::new_default_unset(Shard::All)), + ) + .await + .unwrap(); + + let leftover = conn + .execute(format!( + "SELECT gid FROM pg_prepared_xacts WHERE gid = '{foreign_gid}'" + )) + .await + .unwrap(); + assert!( + leftover.iter().find(|p| p.code() == 'D').is_none(), + "prepared transaction from a legacy WAL record was not cleaned up" + ); +} + +#[tokio::test] +async fn test_cleanup_leaves_other_prefixes_alone() { + config::load_test(); + logger(); + let cluster = databases().all().iter().next().unwrap().1.clone(); + + // A prepared transaction that shares the numeric ID but carries a + // different coordinator prefix belongs to someone else: cleanup + // with a recorded prefix must not touch it. + let number: u64 = rand::random(); + let transaction: TwoPcTransaction = format!("__pgdog_2pc_{number}").parse().unwrap(); + let other_gid = format!("__pgdog_2pc_otherinstance_{number}_0"); + + let mut conn = Connection::new(cluster.user(), cluster.name(), false).unwrap(); + conn.connect( + &Request::default(), + &Route::write(ShardWithPriority::new_default_unset(Shard::All)), + ) + .await + .unwrap(); + conn.execute("BEGIN").await.unwrap(); + conn.execute("SELECT 1").await.unwrap(); + conn.execute(format!("PREPARE TRANSACTION '{other_gid}'")) + .await + .unwrap(); + conn.disconnect(); + + let manager = Manager::get(); + manager.restore_transaction( + transaction, + cluster.user().to_string(), + cluster.name().to_string(), + "__pgdog_2pc_recordedinstance_".to_string(), + TwoPcPhase::Phase1, + ); + + manager.wait_until_cleaned_up(transaction).await; + // The recorded GID exists nowhere, so the transaction resolves. + assert!(manager.transaction(&transaction).is_none()); + + conn.connect( + &Request::default(), + &Route::write(ShardWithPriority::new_default_unset(Shard::All)), + ) + .await + .unwrap(); + + let survivor = conn + .execute(format!( + "SELECT gid FROM pg_prepared_xacts WHERE gid = '{other_gid}'" + )) + .await + .unwrap(); + assert!( + survivor.iter().find(|p| p.code() == 'D').is_some(), + "prepared transaction under a different prefix was resolved by cleanup" + ); + + conn.execute(format!("ROLLBACK PREPARED '{other_gid}'")) + .await + .unwrap(); +} + #[tokio::test] async fn test_cleanup_transaction_phase_two() { config::load_test(); diff --git a/pgdog/src/frontend/client/query_engine/two_pc/transaction.rs b/pgdog/src/frontend/client/query_engine/two_pc/transaction.rs index 24c557881..63a269632 100644 --- a/pgdog/src/frontend/client/query_engine/two_pc/transaction.rs +++ b/pgdog/src/frontend/client/query_engine/two_pc/transaction.rs @@ -16,6 +16,11 @@ impl TwoPcTransaction { Self(rng().random_range(0..usize::MAX)) } + /// The durable numeric ID, unique per transaction. + pub(super) fn number(&self) -> usize { + self.0 + } + /// A prefix to identify two-phase commit transactions generated /// by this PgDog process. pub(super) fn global_prefix() -> String { From ddc6f1af034e1ceee5960bee93789f1b737ee968 Mon Sep 17 00:00:00 2001 From: Will Hopkins Date: Mon, 27 Jul 2026 13:45:13 -0700 Subject: [PATCH 11/21] fix(2pc): render per-shard GIDs from actual shard numbers --- pgdog/src/backend/pool/connection/binding.rs | 15 ++++++++++----- 1 file changed, 10 insertions(+), 5 deletions(-) diff --git a/pgdog/src/backend/pool/connection/binding.rs b/pgdog/src/backend/pool/connection/binding.rs index ac266b4c8..d13c0b7a9 100644 --- a/pgdog/src/backend/pool/connection/binding.rs +++ b/pgdog/src/backend/pool/connection/binding.rs @@ -366,20 +366,25 @@ impl Binding { pub(crate) async fn two_pc_on_guards( servers: &mut [Guard], + state: &MultiShard, transaction: TwoPcTransaction, phase: TwoPcPhase, ) -> Result<(), Error> { let skip_missing = matches!(phase, TwoPcPhase::Phase2 | TwoPcPhase::Rollback); let mut futures = Vec::new(); - for (shard, server) in servers.iter_mut().enumerate() { + for (position, server) in servers.iter_mut().enumerate() { + // Map positional index to actual shard number. + // When only a subset of shards is connected (Shard::Multi binding), + // positional indices don't match actual shard numbers. + let shard = state.shard_index(position); let query = phase_control(transaction, shard, phase); futures.push(server.execute(query)); } let results = join_all(futures).await; - for (shard, result) in results.into_iter().enumerate() { + for (position, result) in results.into_iter().enumerate() { match result { Err(Error::ExecutionError(err)) => { if !(skip_missing && err.code == "42704") { @@ -389,7 +394,7 @@ impl Binding { Err(err) => return Err(err), Ok(_) => { if phase == TwoPcPhase::Phase2 { - servers[shard].stats_mut().transaction_2pc(); + servers[position].stats_mut().transaction_2pc(); } } } @@ -405,8 +410,8 @@ impl Binding { phase: TwoPcPhase, ) -> Result<(), Error> { match self { - Binding::MultiShard(servers, _) => { - Self::two_pc_on_guards(servers, transaction, phase).await + Binding::MultiShard(servers, state) => { + Self::two_pc_on_guards(servers, state, transaction, phase).await } _ => Err(Error::TwoPcMultiShardOnly), From bcfa3f52cef607cb39d3301a6a2e6f94b39b7c12 Mon Sep 17 00:00:00 2001 From: Will Hopkins Date: Mon, 27 Jul 2026 13:48:53 -0700 Subject: [PATCH 12/21] docs(2pc): describe the alphabet check on recorded GIDs --- pgdog/src/backend/pool/connection/binding.rs | 9 +++++---- 1 file changed, 5 insertions(+), 4 deletions(-) diff --git a/pgdog/src/backend/pool/connection/binding.rs b/pgdog/src/backend/pool/connection/binding.rs index d13c0b7a9..9c9951237 100644 --- a/pgdog/src/backend/pool/connection/binding.rs +++ b/pgdog/src/backend/pool/connection/binding.rs @@ -442,10 +442,11 @@ impl Binding { .enumerate() .map(|(shard, server)| async move { let target = TwoPcTransactionOnShard::new(transaction, shard); - // The recorded GID is trusted to be a quotable - // literal because identifier alphabets are - // validated at startup; anything else on disk - // gets the alphabet-checked numeric fallback. + // A GID rendered from the recorded prefix is + // checked against the identifier alphabet before + // it is embedded in a quoted literal; a prefix + // outside the alphabet falls back to numeric-ID + // matching, which performs the same check. let expected = match prefix.is_empty() { true => None, false => { From 12ead9f11d5a66b40e22fa8797d71859e14cc5bb Mon Sep 17 00:00:00 2001 From: Will Hopkins Date: Mon, 27 Jul 2026 13:49:11 -0700 Subject: [PATCH 13/21] style(2pc): branch on prefix presence with if instead of match --- pgdog/src/backend/pool/connection/binding.rs | 27 ++++++++++---------- 1 file changed, 13 insertions(+), 14 deletions(-) diff --git a/pgdog/src/backend/pool/connection/binding.rs b/pgdog/src/backend/pool/connection/binding.rs index 9c9951237..e82b27fe3 100644 --- a/pgdog/src/backend/pool/connection/binding.rs +++ b/pgdog/src/backend/pool/connection/binding.rs @@ -447,20 +447,19 @@ impl Binding { // it is embedded in a quoted literal; a prefix // outside the alphabet falls back to numeric-ID // matching, which performs the same check. - let expected = match prefix.is_empty() { - true => None, - false => { - let gid = target.gid(prefix); - if safe_identifier(&gid) { - Some(gid) - } else { - warn!( - "[2pc] recorded gid {:?} contains characters \ - PgDog never generates; matching by numeric ID", - gid - ); - None - } + let expected = if prefix.is_empty() { + None + } else { + let gid = target.gid(prefix); + if safe_identifier(&gid) { + Some(gid) + } else { + warn!( + "[2pc] recorded gid {:?} contains characters \ + PgDog never generates; matching by numeric ID", + gid + ); + None } }; let rows: Vec = server From 8941d1aeb27169dd0a677bf30b2c11899623b13b Mon Sep 17 00:00:00 2001 From: Will Hopkins Date: Mon, 27 Jul 2026 13:49:48 -0700 Subject: [PATCH 14/21] style(2pc): indent the prepared transactions scan chain --- pgdog/src/backend/pool/connection/binding.rs | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/pgdog/src/backend/pool/connection/binding.rs b/pgdog/src/backend/pool/connection/binding.rs index e82b27fe3..fd2ce9a62 100644 --- a/pgdog/src/backend/pool/connection/binding.rs +++ b/pgdog/src/backend/pool/connection/binding.rs @@ -463,10 +463,10 @@ impl Binding { } }; let rows: Vec = server - .fetch_all( - "SELECT gid FROM pg_prepared_xacts WHERE database = current_database()", - ) - .await?; + .fetch_all( + "SELECT gid FROM pg_prepared_xacts WHERE database = current_database()", + ) + .await?; for row in rows { let Some(gid) = row.get_text(0) else { From b02c011368b564d6b3c68a21af9516d45e8cdc01 Mon Sep 17 00:00:00 2001 From: Will Hopkins Date: Mon, 27 Jul 2026 13:50:07 -0700 Subject: [PATCH 15/21] style(2pc): pass the cleanup statement by reference --- pgdog/src/backend/pool/connection/binding.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pgdog/src/backend/pool/connection/binding.rs b/pgdog/src/backend/pool/connection/binding.rs index fd2ce9a62..b9afca63a 100644 --- a/pgdog/src/backend/pool/connection/binding.rs +++ b/pgdog/src/backend/pool/connection/binding.rs @@ -488,7 +488,7 @@ impl Binding { unreachable!("cleanup resolves transactions; it never prepares") } }; - match server.execute(&statement[..]).await { + match server.execute(&statement).await { Ok(_) => { if phase == TwoPcPhase::Phase2 { server.stats_mut().transaction_2pc(); From 209ce52f5ddb53ac49569dc1ef409ac193c7fc21 Mon Sep 17 00:00:00 2001 From: Will Hopkins Date: Mon, 27 Jul 2026 13:50:21 -0700 Subject: [PATCH 16/21] docs(integration): correct wal_helper comment about the empty prefix --- integration/two_pc_crash_safety/wal_helper/src/main.rs | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) diff --git a/integration/two_pc_crash_safety/wal_helper/src/main.rs b/integration/two_pc_crash_safety/wal_helper/src/main.rs index 792db9724..1b3396bc3 100644 --- a/integration/two_pc_crash_safety/wal_helper/src/main.rs +++ b/integration/two_pc_crash_safety/wal_helper/src/main.rs @@ -33,9 +33,8 @@ async fn main() { let mut segment = Segment::create(&dir, 0).await.expect("create segment"); let mut buf = BytesMut::new(); - // The prefix is left empty to produce the record shape written by - // versions that did not store it, so recovery of such WALs (via the - // numeric-ID fallback) stays covered end to end. + // The prefix is left empty so recovery takes the numeric-ID + // fallback path, keeping that path covered end to end. Record::Begin(BeginPayload { txn, user, From e0ecd83fcd7345062476b054254d9611993524bd Mon Sep 17 00:00:00 2001 From: Will Hopkins Date: Mon, 27 Jul 2026 13:50:48 -0700 Subject: [PATCH 17/21] docs(2pc): state prefix volatility and fallback triggers precisely --- .../client/query_engine/two_pc/statement.rs | 13 +++++++------ .../client/query_engine/two_pc/transaction.rs | 3 ++- 2 files changed, 9 insertions(+), 7 deletions(-) diff --git a/pgdog/src/frontend/client/query_engine/two_pc/statement.rs b/pgdog/src/frontend/client/query_engine/two_pc/statement.rs index d673c621e..42a3b0551 100644 --- a/pgdog/src/frontend/client/query_engine/two_pc/statement.rs +++ b/pgdog/src/frontend/client/query_engine/two_pc/statement.rs @@ -37,12 +37,13 @@ impl TwoPcTransactionOnShard { /// Whether `gid`, as listed in `pg_prepared_xacts`, refers to this /// transaction on this shard. GID prefixes embed the identity of the - /// PgDog process that created the transaction, which changes across - /// restarts, so matching uses the durable numeric transaction ID and - /// the shard index. This is the fallback for transactions restored - /// from WAL records that did not store the coordinator GID prefix; - /// with a recorded prefix, cleanup matches the exact GID from - /// [`Self::gid`] instead. + /// PgDog process that created the transaction, which can change + /// across restarts, so matching uses the durable numeric transaction + /// ID and the shard index. This is the fallback for transactions + /// restored from WAL records that did not store the coordinator GID + /// prefix, and for recorded prefixes that fail the identifier + /// alphabet check; with a recorded prefix, cleanup matches the exact + /// GID from [`Self::gid`] instead. /// /// A matched GID is guaranteed to contain only [`safe_identifier`] /// characters, so it can be embedded verbatim in a quoted SQL literal. diff --git a/pgdog/src/frontend/client/query_engine/two_pc/transaction.rs b/pgdog/src/frontend/client/query_engine/two_pc/transaction.rs index 63a269632..10bf62a43 100644 --- a/pgdog/src/frontend/client/query_engine/two_pc/transaction.rs +++ b/pgdog/src/frontend/client/query_engine/two_pc/transaction.rs @@ -48,7 +48,8 @@ impl FromStr for TwoPcTransaction { /// Parse a transaction from a GID generated by any PgDog process. /// The GID must carry the [`PREFIX`] marker; only the trailing /// numeric ID is significant because the rest of the prefix embeds - /// process identity, which differs between instances and restarts. + /// process identity, which can differ between instances and + /// restarts. fn from_str(s: &str) -> Result { if !s.starts_with(PREFIX) { return Err(()); From 5a02ac6d1b165c165ab1dfa1a3285450dc8d72b3 Mon Sep 17 00:00:00 2001 From: Will Hopkins Date: Mon, 27 Jul 2026 13:51:01 -0700 Subject: [PATCH 18/21] docs(2pc): document gid as rendering the given prefix --- pgdog/src/frontend/client/query_engine/two_pc/statement.rs | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/pgdog/src/frontend/client/query_engine/two_pc/statement.rs b/pgdog/src/frontend/client/query_engine/two_pc/statement.rs index 42a3b0551..41f89ff14 100644 --- a/pgdog/src/frontend/client/query_engine/two_pc/statement.rs +++ b/pgdog/src/frontend/client/query_engine/two_pc/statement.rs @@ -28,9 +28,9 @@ impl TwoPcTransactionOnShard { self.transaction } - /// The exact GID this transaction was prepared under on this shard, - /// rendered from the coordinator GID prefix recorded when the - /// transaction was created. + /// The GID this transaction is prepared under on this shard when + /// rendered with `prefix`, the coordinator GID prefix recorded at + /// transaction creation. pub(crate) fn gid(&self, prefix: &str) -> String { format!("{}{}_{}", prefix, self.transaction.number(), self.shard) } From faea7dc643f3e3dadc4e752facb173b055e42576 Mon Sep 17 00:00:00 2001 From: Will Hopkins Date: Mon, 27 Jul 2026 13:51:19 -0700 Subject: [PATCH 19/21] style(2pc): bind every checkpoint entry field during recovery --- pgdog/src/frontend/client/query_engine/two_pc/wal/recovery.rs | 1 - 1 file changed, 1 deletion(-) diff --git a/pgdog/src/frontend/client/query_engine/two_pc/wal/recovery.rs b/pgdog/src/frontend/client/query_engine/two_pc/wal/recovery.rs index 1967dd465..84a52e105 100644 --- a/pgdog/src/frontend/client/query_engine/two_pc/wal/recovery.rs +++ b/pgdog/src/frontend/client/query_engine/two_pc/wal/recovery.rs @@ -208,7 +208,6 @@ fn apply(working: &mut HashMap, record: Record) { database, decided, prefix, - .. } in p.active { working.insert( From 1f5c66b8725843a5b4d34a805b2046e18072a122 Mon Sep 17 00:00:00 2001 From: Will Hopkins Date: Mon, 27 Jul 2026 13:51:43 -0700 Subject: [PATCH 20/21] refactor(util): rename safe_identifier to is_safe_identifier --- pgdog/src/backend/pool/connection/binding.rs | 4 ++-- .../client/query_engine/two_pc/statement.rs | 6 ++--- pgdog/src/util.rs | 24 +++++++++---------- 3 files changed, 17 insertions(+), 17 deletions(-) diff --git a/pgdog/src/backend/pool/connection/binding.rs b/pgdog/src/backend/pool/connection/binding.rs index b9afca63a..d8deb68f2 100644 --- a/pgdog/src/backend/pool/connection/binding.rs +++ b/pgdog/src/backend/pool/connection/binding.rs @@ -10,7 +10,7 @@ use crate::{ }, net::{DataRow, FrontendPid, ProtocolMessage, Query, parameter::Parameters}, state::State, - util::safe_identifier, + util::is_safe_identifier, }; use futures::future::join_all; @@ -451,7 +451,7 @@ impl Binding { None } else { let gid = target.gid(prefix); - if safe_identifier(&gid) { + if is_safe_identifier(&gid) { Some(gid) } else { warn!( diff --git a/pgdog/src/frontend/client/query_engine/two_pc/statement.rs b/pgdog/src/frontend/client/query_engine/two_pc/statement.rs index 41f89ff14..6ca310561 100644 --- a/pgdog/src/frontend/client/query_engine/two_pc/statement.rs +++ b/pgdog/src/frontend/client/query_engine/two_pc/statement.rs @@ -5,7 +5,7 @@ use std::{fmt::Display, str::FromStr}; use tracing::warn; use crate::frontend::client::query_engine::two_pc::TwoPcTransaction; -use crate::util::safe_identifier; +use crate::util::is_safe_identifier; use super::TwoPcPhase; @@ -45,7 +45,7 @@ impl TwoPcTransactionOnShard { /// alphabet check; with a recorded prefix, cleanup matches the exact /// GID from [`Self::gid`] instead. /// - /// A matched GID is guaranteed to contain only [`safe_identifier`] + /// A matched GID is guaranteed to contain only [`is_safe_identifier`] /// characters, so it can be embedded verbatim in a quoted SQL literal. /// PgDog only generates such GIDs (`NODE_ID` and `DEPLOYMENT_ID` are /// validated at startup); a GID that matches the numeric ID but not @@ -57,7 +57,7 @@ impl TwoPcTransactionOnShard { Err(()) => false, }; - if matches && !safe_identifier(gid) { + if matches && !is_safe_identifier(gid) { warn!( "[2pc] prepared transaction {:?} matches transaction {} on shard {} \ but contains characters PgDog never generates; refusing to resolve it", diff --git a/pgdog/src/util.rs b/pgdog/src/util.rs index 6d8003682..535676b81 100644 --- a/pgdog/src/util.rs +++ b/pgdog/src/util.rs @@ -132,7 +132,7 @@ static INSTANCE_ID: Lazy = Lazy::new(|| match env::var("NODE_ID") { /// Get the instance ID for this pgdog instance. /// This is generated once at startup and persists for the lifetime of the process. /// An unset or empty `NODE_ID` auto-generates a random ID; otherwise the value -/// is restricted to [`safe_identifier`] characters, which +/// is restricted to [`is_safe_identifier`] characters, which /// [`validate_instance_identity`] enforces at startup. pub fn instance_id() -> &'static str { &INSTANCE_ID @@ -141,7 +141,7 @@ pub fn instance_id() -> &'static str { /// True if `s` contains only ASCII alphanumerics, underscores and hyphens: /// the alphabet PgDog allows in instance identifiers and, by extension, in /// the two-phase commit GIDs they are embedded in. -pub fn safe_identifier(s: &str) -> bool { +pub fn is_safe_identifier(s: &str) -> bool { s.bytes() .all(|b| b.is_ascii_alphanumeric() || b == b'_' || b == b'-') } @@ -168,7 +168,7 @@ pub struct IdentityError { pub fn validate_instance_identity() -> Result<(), IdentityError> { for var in ["NODE_ID", "DEPLOYMENT_ID"] { if let Ok(value) = env::var(var) - && !safe_identifier(&value) + && !is_safe_identifier(&value) { return Err(IdentityError { var, value }); } @@ -197,7 +197,7 @@ static DEPLOYMENT_ID: Lazy> = /// This should be _globally_ unique /// and is used to differentiate 2pc transactions. /// An unset or empty `DEPLOYMENT_ID` means no deployment ID; otherwise the -/// value is restricted to [`safe_identifier`] characters, which +/// value is restricted to [`is_safe_identifier`] characters, which /// [`validate_instance_identity`] enforces at startup. /// pub(crate) fn deployment_id() -> Option<&'static str> { @@ -387,14 +387,14 @@ mod test { use crate::test_utils::*; #[test] - fn test_safe_identifier() { - assert!(safe_identifier("pgdog-node-1")); - assert!(safe_identifier("prod_us_east_2")); - assert!(safe_identifier("deadbeef")); - assert!(!safe_identifier("it's")); - assert!(!safe_identifier("node\\1")); - assert!(!safe_identifier("node 1")); - assert!(!safe_identifier("nöde")); + fn test_is_safe_identifier() { + assert!(is_safe_identifier("pgdog-node-1")); + assert!(is_safe_identifier("prod_us_east_2")); + assert!(is_safe_identifier("deadbeef")); + assert!(!is_safe_identifier("it's")); + assert!(!is_safe_identifier("node\\1")); + assert!(!is_safe_identifier("node 1")); + assert!(!is_safe_identifier("nöde")); } #[test] From 09beb8a0bafc83a00003baa8a694526ec2351c03 Mon Sep 17 00:00:00 2001 From: Will Hopkins Date: Mon, 27 Jul 2026 13:52:21 -0700 Subject: [PATCH 21/21] test(2pc): cover checkpoint entries written without a prefix --- .../client/query_engine/two_pc/wal/record.rs | 51 +++++++++++++++++++ 1 file changed, 51 insertions(+) diff --git a/pgdog/src/frontend/client/query_engine/two_pc/wal/record.rs b/pgdog/src/frontend/client/query_engine/two_pc/wal/record.rs index 1eb51c135..56ebea495 100644 --- a/pgdog/src/frontend/client/query_engine/two_pc/wal/record.rs +++ b/pgdog/src/frontend/client/query_engine/two_pc/wal/record.rs @@ -296,6 +296,57 @@ mod tests { })); } + #[test] + fn checkpoint_without_prefix_decodes_with_empty_prefix() { + // The on-disk shape written by versions that predate the + // `prefix` field on checkpoint entries. + #[derive(Serialize)] + struct OldCheckpointEntry { + txn: TwoPcTransaction, + user: String, + database: String, + decided: bool, + } + #[derive(Serialize)] + struct OldCheckpointPayload { + active: Vec, + } + + let txn = TwoPcTransaction::new(); + let payload = rmp_serde::to_vec_named(&OldCheckpointPayload { + active: vec![OldCheckpointEntry { + txn, + user: "alice".into(), + database: "shop".into(), + decided: true, + }], + }) + .unwrap(); + + let mut buf = Vec::new(); + let tag = Tag::Checkpoint as u8; + let mut crc = crc32c::crc32c(&[tag]); + crc = crc32c::crc32c_append(crc, &payload); + buf.put_u32_le(1 + payload.len() as u32); + buf.put_u32_le(crc); + buf.put_u8(tag); + buf.put_slice(&payload); + + let decoded = Record::decode(&buf).unwrap().unwrap(); + assert_eq!( + decoded.record, + Record::Checkpoint(CheckpointPayload { + active: vec![CheckpointEntry { + txn, + user: "alice".into(), + database: "shop".into(), + decided: true, + prefix: String::new(), + }], + }) + ); + } + #[test] fn round_trip_checkpoint() { round_trip(Record::Checkpoint(CheckpointPayload {