From f924c43c58c1553eb6a9d76862d2b2fe21ba34d4 Mon Sep 17 00:00:00 2001 From: danielshih Date: Tue, 21 Jul 2026 08:16:16 +0000 Subject: [PATCH] Refactor relation metadata handling to use a structured type for improved clarity and maintainability Update dependencies: bump tokio to 1.53.1 and serde_json to 1.0.151 Add server-version preflight checks for replication slot operations and base backup options Add preflight checks for replication slot and base backup options based on PostgreSQL version --- .github/workflows/ci-checks.yml | 6 + Cargo.lock | 8 +- Cargo.toml | 10 +- integration-tests/version_preflight.rs | 198 +++++++++ src/connection/libpq.rs | 561 +------------------------ src/connection/native/connection.rs | 149 +++---- src/protocol.rs | 189 ++++++--- src/sql_builder.rs | 253 +++++++++++ src/stream.rs | 18 +- 9 files changed, 692 insertions(+), 700 deletions(-) create mode 100644 integration-tests/version_preflight.rs diff --git a/.github/workflows/ci-checks.yml b/.github/workflows/ci-checks.yml index 21c18db..85b9662 100644 --- a/.github/workflows/ci-checks.yml +++ b/.github/workflows/ci-checks.yml @@ -330,6 +330,12 @@ jobs: # BASE_BACKUP option combination: asserts the server accepts every generated option name (PG15+), then aborts the backup without draining. cargo test --test base_backup_options -- --ignored --nocapture --test-threads=1 + PGPASSWORD=postgres psql -h localhost -U postgres -d test_walstream \ + -c "SELECT pg_drop_replication_slot(slot_name) FROM pg_replication_slots WHERE slot_type = 'logical';" || true + + # Server-version preflight: end-to-end core CDC path (create slot -> stream -> decode INSERT) plus PG17+ gated slot ops (FAILOVER create + ALTER) that must pass the client-side version gate and execute; the gated portion skips gracefully on PG<17. + cargo test --test version_preflight -- --ignored --nocapture --test-threads=1 + integration-ssl: name: SSL Integration Tests (PG ${{ matrix.pg_version }}) runs-on: ubuntu-latest diff --git a/Cargo.lock b/Cargo.lock index e431405..90cb521 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1321,9 +1321,9 @@ dependencies = [ [[package]] name = "serde_json" -version = "1.0.150" +version = "1.0.151" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e8014e44b4736ed0538adeecded0fce2a272f22dc9578a7eb6b2d9993c74cfb9" +checksum = "c841b55ecdae098c80dcae9cf767f6f8a0c2cdb3416bbef72181df4d0fe73f14" dependencies = [ "itoa", "memchr", @@ -1491,9 +1491,9 @@ checksum = "1f3ccbac311fea05f86f61904b462b55fb3df8837a366dfc601a0161d0532f20" [[package]] name = "tokio" -version = "1.53.0" +version = "1.53.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d988bcd52dbe076d3d46903332f58c912b87a2c49b1428419a5845154762ffee" +checksum = "202caea871b69668250d242070849eb495be178ed697a3e98aebce5bc81a0bed" dependencies = [ "bytes", "libc", diff --git a/Cargo.toml b/Cargo.toml index f86931e..d70f8b7 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -30,7 +30,7 @@ rust-version = "1.87" pg-walstream-macros = { path = "macros", version = "=0.8.0" } [dependencies] -tokio = { version = "1.53.0", features = ["io-util", "net", "time", "macros", "rt", "rt-multi-thread"], optional = true } +tokio = { version = "1.53.1", features = ["io-util", "net", "time", "macros", "rt", "rt-multi-thread"], optional = true } tokio-util = { version = "0.7.18", optional = true } serde = { version = "1.0.229", default-features = false, features = ["derive", "rc", "alloc"] } chrono = { version = "0.4.45", default-features = false, features = ["serde", "alloc"] } @@ -85,12 +85,12 @@ rustls-tls = [ ] [dev-dependencies] -tokio = { version = "1.53.0", features = ["full"] } +tokio = { version = "1.53.1", features = ["full"] } criterion = { package = "codspeed-criterion-compat", version = "5.0.1", features = ["html_reports"] } futures = "0.3.33" serde_bytes = "0.11.19" proptest = "1.11" -serde_json = "1.0.150" +serde_json = "1.0.151" tracing-subscriber = "0.3.23" [[test]] @@ -157,6 +157,10 @@ path = "integration-tests/slot_matrix.rs" name = "base_backup_options" path = "integration-tests/base_backup_options.rs" +[[test]] +name = "version_preflight" +path = "integration-tests/version_preflight.rs" + [[bench]] name = "wal_pipeline" harness = false diff --git a/integration-tests/version_preflight.rs b/integration-tests/version_preflight.rs new file mode 100644 index 0000000..142e862 --- /dev/null +++ b/integration-tests/version_preflight.rs @@ -0,0 +1,198 @@ +#![cfg(any(feature = "libpq", feature = "rustls-tls"))] + +//! Integration tests for the client-side server-version preflight on replication +//! SQL, plus an end-to-end core-CDC-path smoke test. +//! +//! Requires a live PostgreSQL with `wal_level = logical`. On PG17+ the gated-option +//! test exercises FAILOVER slot create/alter through the library; on older servers +//! that portion skips with a `warn!`. The core-path test runs on any supported +//! version. +//! +//! ## Prerequisites +//! - `DATABASE_URL` — replication connection +//! (`postgresql://user:pass@host:5432/db?sslmode=require&replication=database`) +//! - `DATABASE_URL_REGULAR` — regular connection to the same database +//! +//! ## Running +//! ```bash +//! cargo test --test version_preflight -- --ignored --nocapture --test-threads=1 +//! ``` + +use pg_walstream::{ + EventType, LogicalReplicationStream, PgReplicationConnection, ReplicationSlotOptions, + ReplicationStreamConfig, RetryConfig, SlotType, StreamingMode, +}; +use std::time::Duration; +use tokio_util::sync::CancellationToken; +use tracing::warn; + +fn replication_conn_string() -> String { + std::env::var("DATABASE_URL").unwrap_or_else(|_| { + "postgresql://postgres:postgres@localhost:5432/test_walstream?replication=database" + .to_string() + }) +} + +fn regular_conn_string() -> String { + std::env::var("DATABASE_URL_REGULAR").unwrap_or_else(|_| { + // Strip `replication=database` while keeping a valid query string. + let repl = replication_conn_string(); + repl.replace("?replication=database&", "?") + .replace("&replication=database", "") + .replace("?replication=database", "") + }) +} + +fn init_tracing() { + let _ = tracing_subscriber::fmt() + .with_max_level(tracing::Level::INFO) + .try_init(); +} + +fn server_version_num(conn: &mut PgReplicationConnection) -> i64 { + conn.exec("SHOW server_version_num") + .expect("SHOW server_version_num") + .get_value(0, 0) + .expect("server_version_num present") + .parse() + .expect("server_version_num numeric") +} + +fn drop_slot(slot: &str) { + if let Ok(mut conn) = PgReplicationConnection::connect(&replication_conn_string()) { + let _ = conn.exec(&format!( + "SELECT pg_drop_replication_slot('{slot}') WHERE EXISTS \ + (SELECT 1 FROM pg_replication_slots WHERE slot_name = '{slot}')" + )); + } +} + +/// Drops its slot on `Drop` so a panicking assertion never leaks a slot (an +/// orphaned slot pins WAL and can exhaust server disk). +struct SlotGuard(&'static str); +impl Drop for SlotGuard { + fn drop(&mut self) { + drop_slot(self.0); + } +} + +fn core_cfg(slot: &str) -> ReplicationStreamConfig { + ReplicationStreamConfig::new( + slot.to_string(), + "vp_pub".to_string(), + 2, + StreamingMode::Off, + Duration::from_secs(10), + Duration::from_secs(30), + Duration::from_secs(60), + RetryConfig::default(), + ) +} + +/// End-to-end: the library creates the slot (through the new preflight, a no-op +/// for default options), streams, and decodes a live INSERT. Proves the whole +/// pipeline works on the target server. +#[tokio::test] +#[ignore = "requires live PostgreSQL with wal_level=logical"] +async fn core_cdc_path_streams_and_decodes_insert() { + init_tracing(); + let slot = "vp_it_core_slot"; + drop_slot(slot); + let _guard = SlotGuard(slot); + + let mut regular = + PgReplicationConnection::connect(®ular_conn_string()).expect("regular connection"); + let _ = regular.exec( + "CREATE TABLE IF NOT EXISTS vp_events (id SERIAL PRIMARY KEY, payload TEXT NOT NULL)", + ); + let _ = regular.exec("TRUNCATE vp_events RESTART IDENTITY"); + let _ = regular.exec("DROP PUBLICATION IF EXISTS vp_pub"); + regular + .exec("CREATE PUBLICATION vp_pub FOR TABLE vp_events") + .expect("create publication"); + + // The library creates the slot here (exercising the preflight for default + // options) and enters COPY-both. + let mut stream = LogicalReplicationStream::new(&replication_conn_string(), core_cfg(slot)) + .await + .expect("replication stream"); + stream.start(None).await.expect("start"); + + // Insert AFTER the slot exists so its WAL is captured. + regular + .exec("INSERT INTO vp_events (payload) VALUES ('hello')") + .expect("insert"); + + let cancel = CancellationToken::new(); + let decoded = tokio::time::timeout(Duration::from_secs(60), async { + loop { + let event = stream.next_event(&cancel).await.expect("stream error"); + if let EventType::Insert { table, data, .. } = &event.event_type { + assert_eq!(&**table, "vp_events", "insert on the wrong table"); + assert!( + data.get("payload").map(|v| *v == "hello").unwrap_or(false), + "payload column must decode to 'hello'" + ); + break; + } + } + }) + .await; + assert!(decoded.is_ok(), "timed out waiting for the INSERT event"); + + drop(stream); + tokio::time::sleep(Duration::from_millis(200)).await; + let _ = regular.exec("DROP PUBLICATION IF EXISTS vp_pub"); + let _ = regular.exec("DROP TABLE IF EXISTS vp_events"); +} + +/// On PG17+, version-gated slot ops (FAILOVER create + ALTER) pass the new +/// client-side preflight and execute on the live server. Skips below PG17. +#[test] +#[ignore = "requires live PostgreSQL 17+ with wal_level=logical"] +fn gated_slot_ops_pass_preflight_and_execute_on_pg17plus() { + init_tracing(); + let slot = "vp_it_gated_slot"; + drop_slot(slot); + let _guard = SlotGuard(slot); + + let mut regular = + PgReplicationConnection::connect(®ular_conn_string()).expect("regular connection"); + let version = server_version_num(&mut regular); + if version < 170000 { + warn!("skipping gated-ops test: server_version_num {version} < 170000 (FAILOVER is PG17+)"); + return; + } + + let mut repl = + PgReplicationConnection::connect(&replication_conn_string()).expect("replication conn"); + + // FAILOVER create — preflight (>= 170000) passes, server accepts. + let opts = ReplicationSlotOptions { + failover: true, + snapshot: Some("nothing".to_string()), + ..Default::default() + }; + repl.create_replication_slot_with_options(slot, SlotType::Logical, Some("pgoutput"), &opts) + .expect("FAILOVER slot create must pass preflight and succeed on PG17+"); + + // ALTER turning failover off — preflight (>= 170000, no two_phase) passes. + repl.alter_replication_slot(slot, None, Some(false)) + .expect("ALTER_REPLICATION_SLOT (failover) must pass preflight and succeed on PG17+"); + + // Verify the ALTER took effect, then drop. + let r = regular + .exec(&format!( + "SELECT failover FROM pg_replication_slots WHERE slot_name = '{slot}'" + )) + .expect("query slot"); + assert_eq!(r.ntuples(), 1, "slot must exist"); + assert_eq!( + r.get_value(0, 0).as_deref(), + Some("f"), + "failover must be off after ALTER" + ); + + repl.drop_replication_slot(slot, false) + .expect("drop slot must succeed"); +} diff --git a/src/connection/libpq.rs b/src/connection/libpq.rs index 3ce80d2..d9a1e10 100644 --- a/src/connection/libpq.rs +++ b/src/connection/libpq.rs @@ -297,15 +297,6 @@ impl PgReplicationConnection { Ok(result) } - /// Build the SQL string for `START_REPLICATION` (logical). - fn build_start_replication_sql( - slot_name: &str, - start_lsn: XLogRecPtr, - options: &[(&str, &str)], - ) -> Result { - crate::sql_builder::build_start_replication_sql(slot_name, start_lsn, options) - } - /// Start logical replication pub fn start_replication( &mut self, @@ -313,7 +304,7 @@ impl PgReplicationConnection { start_lsn: XLogRecPtr, options: &[(&str, &str)], ) -> Result<()> { - let sql = Self::build_start_replication_sql(slot_name, start_lsn, options)?; + let sql = crate::sql_builder::build_start_replication_sql(slot_name, start_lsn, options)?; debug!("Starting replication: {}", sql); let _result = self.exec(&sql)?; @@ -683,30 +674,17 @@ impl PgReplicationConnection { output_plugin: Option<&str>, options: &ReplicationSlotOptions, ) -> Result { - let sql = Self::build_create_slot_sql(slot_name, slot_type, output_plugin, options)?; + crate::sql_builder::check_create_slot_version(self.server_version(), slot_type, options)?; + let sql = crate::sql_builder::build_create_slot_sql( + slot_name, + slot_type, + output_plugin, + options, + )?; debug!("Creating replication slot: {}", sql); self.exec(&sql) } - /// Build the SQL string for `CREATE_REPLICATION_SLOT`. - fn build_create_slot_sql( - slot_name: &str, - slot_type: SlotType, - output_plugin: Option<&str>, - options: &ReplicationSlotOptions, - ) -> Result { - crate::sql_builder::build_create_slot_sql(slot_name, slot_type, output_plugin, options) - } - - /// Build the SQL string for `ALTER_REPLICATION_SLOT`. - fn build_alter_slot_sql( - slot_name: &str, - two_phase: Option, - failover: Option, - ) -> Result { - crate::sql_builder::build_alter_slot_sql(slot_name, two_phase, failover) - } - /// Alter a replication slot (logical slots only) pub fn alter_replication_slot( &mut self, @@ -714,7 +692,9 @@ impl PgReplicationConnection { two_phase: Option, failover: Option, ) -> Result { - let alter_slot_sql = Self::build_alter_slot_sql(slot_name, two_phase, failover)?; + crate::sql_builder::check_alter_slot_version(self.server_version(), two_phase)?; + let alter_slot_sql = + crate::sql_builder::build_alter_slot_sql(slot_name, two_phase, failover)?; debug!("Altering replication slot: {}", alter_slot_sql); let result = self.exec(&alter_slot_sql)?; @@ -722,11 +702,6 @@ impl PgReplicationConnection { Ok(result) } - /// Build the SQL string for `DROP_REPLICATION_SLOT`. - fn build_drop_slot_sql(slot_name: &str, wait: bool) -> Result { - crate::sql_builder::build_drop_slot_sql(slot_name, wait) - } - /// Drop a replication slot /// /// Generates: `DROP_REPLICATION_SLOT "slot_name" [WAIT]` @@ -736,7 +711,7 @@ impl PgReplicationConnection { /// * `slot_name` - Name of the replication slot to drop /// * `wait` - If true, the command waits until the slot becomes inactive instead of returning an error when the slot is in use pub fn drop_replication_slot(&mut self, slot_name: &str, wait: bool) -> Result<()> { - let sql = Self::build_drop_slot_sql(slot_name, wait)?; + let sql = crate::sql_builder::build_drop_slot_sql(slot_name, wait)?; debug!("Dropping replication slot: {}", sql); let result = self.exec(&sql)?; @@ -753,11 +728,6 @@ impl PgReplicationConnection { Ok(()) } - /// Build the SQL string for `READ_REPLICATION_SLOT`. - fn build_read_slot_sql(slot_name: &str) -> Result { - crate::sql_builder::build_read_slot_sql(slot_name) - } - /// Read information about a replication slot /// /// Generates: `READ_REPLICATION_SLOT "slot_name"` @@ -768,7 +738,8 @@ impl PgReplicationConnection { &mut self, slot_name: &str, ) -> Result { - let sql = Self::build_read_slot_sql(slot_name)?; + crate::sql_builder::check_read_slot_version(self.server_version())?; + let sql = crate::sql_builder::build_read_slot_sql(slot_name)?; debug!("Reading replication slot: {}", sql); let result = self.exec(&sql)?; @@ -796,15 +767,6 @@ impl PgReplicationConnection { }) } - /// Build the SQL string for `START_REPLICATION` (physical). - fn build_start_physical_replication_sql( - slot_name: Option<&str>, - start_lsn: XLogRecPtr, - timeline_id: Option, - ) -> Result { - crate::sql_builder::build_start_physical_replication_sql(slot_name, start_lsn, timeline_id) - } - /// Start physical replication pub fn start_physical_replication( &mut self, @@ -812,7 +774,11 @@ impl PgReplicationConnection { start_lsn: XLogRecPtr, timeline_id: Option, ) -> Result<()> { - let sql = Self::build_start_physical_replication_sql(slot_name, start_lsn, timeline_id)?; + let sql = crate::sql_builder::build_start_physical_replication_sql( + slot_name, + start_lsn, + timeline_id, + )?; debug!("Starting physical replication: {}", sql); let _result = self.exec(&sql)?; @@ -846,14 +812,10 @@ impl PgReplicationConnection { Ok(()) } - /// Build the SQL string for `BASE_BACKUP`. - fn build_base_backup_sql(options: &BaseBackupOptions) -> Result { - crate::sql_builder::build_base_backup_sql(options) - } - /// Start a base backup with options pub fn base_backup(&mut self, options: &BaseBackupOptions) -> Result { - let base_backup_sql = Self::build_base_backup_sql(options)?; + crate::sql_builder::check_base_backup_version(self.server_version(), options)?; + let base_backup_sql = crate::sql_builder::build_base_backup_sql(options)?; debug!("Starting base backup: {}", base_backup_sql); let result = self.exec(&base_backup_sql)?; @@ -1122,7 +1084,6 @@ fn drain_buffered_messages( mod tests { use super::*; use crate::sql_builder::quote_literal; - use crate::INVALID_XLOG_REC_PTR; fn sanitize_sql_string_value(value: &str) -> String { let quoted = quote_literal(value).unwrap(); @@ -1318,26 +1279,6 @@ mod tests { drop(conn); // should not panic } - #[test] - fn test_slot_sql_delegates_to_builder() { - // Delegates to sql_builder::build_create_slot_sql (exhaustively covered by create_slot_sql_cases there). - let opts = ReplicationSlotOptions { - failover: true, - ..Default::default() - }; - let sql = PgReplicationConnection::build_create_slot_sql( - "slot", - SlotType::Logical, - Some("pgoutput"), - &opts, - ) - .unwrap(); - assert_eq!( - sql, - r#"CREATE_REPLICATION_SLOT "slot" LOGICAL "pgoutput" (FAILOVER);"# - ); - } - // ======================================== // ReadResult and Bytes integration tests // ======================================== @@ -1488,50 +1429,6 @@ mod tests { assert_eq!(ExecStatusType::PGRES_COPY_BOTH as i32, 8); } - // ======================================== - // build_drop_slot_sql tests - // ======================================== - - #[test] - fn test_build_drop_slot_sql_without_wait() { - let sql = PgReplicationConnection::build_drop_slot_sql("my_slot", false).unwrap(); - assert_eq!(sql, r#"DROP_REPLICATION_SLOT "my_slot";"#); - } - - #[test] - fn test_build_drop_slot_sql_with_wait() { - let sql = PgReplicationConnection::build_drop_slot_sql("my_slot", true).unwrap(); - assert_eq!(sql, r#"DROP_REPLICATION_SLOT "my_slot" WAIT;"#); - } - - #[test] - fn test_build_drop_slot_sql_injection() { - let sql = PgReplicationConnection::build_drop_slot_sql(r#"evil"slot"#, false).unwrap(); - assert_eq!(sql, r#"DROP_REPLICATION_SLOT "evil""slot";"#); - } - - #[test] - fn test_build_drop_slot_sql_injection_with_wait() { - let sql = PgReplicationConnection::build_drop_slot_sql(r#"evil"slot"#, true).unwrap(); - assert_eq!(sql, r#"DROP_REPLICATION_SLOT "evil""slot" WAIT;"#); - } - - // ======================================== - // build_read_slot_sql tests - // ======================================== - - #[test] - fn test_build_read_slot_sql_basic() { - let sql = PgReplicationConnection::build_read_slot_sql("my_slot").unwrap(); - assert_eq!(sql, r#"READ_REPLICATION_SLOT "my_slot";"#); - } - - #[test] - fn test_build_read_slot_sql_injection() { - let sql = PgReplicationConnection::build_read_slot_sql(r#"evil"slot"#).unwrap(); - assert_eq!(sql, r#"READ_REPLICATION_SLOT "evil""slot";"#); - } - // ======================================== // DrainResult tests // ======================================== @@ -1621,422 +1518,6 @@ mod tests { ); } - // ======================================== - // build_base_backup_sql tests - // ======================================== - - #[test] - fn test_base_backup_sql_default_options() { - let opts = BaseBackupOptions::default(); - let sql = PgReplicationConnection::build_base_backup_sql(&opts).unwrap(); - assert_eq!(sql, "BASE_BACKUP"); - } - - #[test] - fn test_base_backup_sql_with_label() { - let opts = BaseBackupOptions { - label: Some("my_backup".to_string()), - ..Default::default() - }; - let sql = PgReplicationConnection::build_base_backup_sql(&opts).unwrap(); - assert_eq!(sql, "BASE_BACKUP (LABEL 'my_backup')"); - } - - #[test] - fn test_base_backup_sql_with_target() { - let opts = BaseBackupOptions { - target: Some("client".to_string()), - ..Default::default() - }; - let sql = PgReplicationConnection::build_base_backup_sql(&opts).unwrap(); - assert_eq!(sql, "BASE_BACKUP (TARGET 'client')"); - } - - #[test] - fn test_base_backup_sql_with_target_detail() { - let opts = BaseBackupOptions { - target: Some("server".to_string()), - target_detail: Some("/var/backups".to_string()), - ..Default::default() - }; - let sql = PgReplicationConnection::build_base_backup_sql(&opts).unwrap(); - assert_eq!( - sql, - "BASE_BACKUP (TARGET 'server', TARGET_DETAIL '/var/backups')" - ); - } - - #[test] - fn test_base_backup_sql_with_progress() { - let opts = BaseBackupOptions { - progress: true, - ..Default::default() - }; - let sql = PgReplicationConnection::build_base_backup_sql(&opts).unwrap(); - assert_eq!(sql, "BASE_BACKUP (PROGRESS true)"); - } - - #[test] - fn test_base_backup_sql_with_checkpoint() { - let opts = BaseBackupOptions { - checkpoint: Some("fast".to_string()), - ..Default::default() - }; - let sql = PgReplicationConnection::build_base_backup_sql(&opts).unwrap(); - assert_eq!(sql, "BASE_BACKUP (CHECKPOINT 'fast')"); - } - - #[test] - fn test_base_backup_sql_with_wal() { - let opts = BaseBackupOptions { - wal: true, - ..Default::default() - }; - let sql = PgReplicationConnection::build_base_backup_sql(&opts).unwrap(); - assert_eq!(sql, "BASE_BACKUP (WAL true)"); - } - - #[test] - fn test_base_backup_sql_with_wait() { - let opts = BaseBackupOptions { - wait: true, - ..Default::default() - }; - let sql = PgReplicationConnection::build_base_backup_sql(&opts).unwrap(); - assert_eq!(sql, "BASE_BACKUP (WAIT true)"); - } - - #[test] - fn test_base_backup_sql_with_compression() { - let opts = BaseBackupOptions { - compression: Some("gzip".to_string()), - ..Default::default() - }; - let sql = PgReplicationConnection::build_base_backup_sql(&opts).unwrap(); - assert_eq!(sql, "BASE_BACKUP (COMPRESSION 'gzip')"); - } - - #[test] - fn test_base_backup_sql_with_compression_detail() { - let opts = BaseBackupOptions { - compression: Some("zstd".to_string()), - compression_detail: Some("level=3".to_string()), - ..Default::default() - }; - let sql = PgReplicationConnection::build_base_backup_sql(&opts).unwrap(); - assert_eq!( - sql, - "BASE_BACKUP (COMPRESSION 'zstd', COMPRESSION_DETAIL 'level=3')" - ); - } - - #[test] - fn test_base_backup_sql_with_max_rate() { - let opts = BaseBackupOptions { - max_rate: Some(32768), - ..Default::default() - }; - let sql = PgReplicationConnection::build_base_backup_sql(&opts).unwrap(); - assert_eq!(sql, "BASE_BACKUP (MAX_RATE 32768)"); - } - - #[test] - fn test_base_backup_sql_with_tablespace_map() { - let opts = BaseBackupOptions { - tablespace_map: true, - ..Default::default() - }; - let sql = PgReplicationConnection::build_base_backup_sql(&opts).unwrap(); - assert_eq!(sql, "BASE_BACKUP (TABLESPACE_MAP true)"); - } - - #[test] - fn test_base_backup_sql_with_verify_checksums() { - let opts = BaseBackupOptions { - verify_checksums: true, - ..Default::default() - }; - let sql = PgReplicationConnection::build_base_backup_sql(&opts).unwrap(); - assert_eq!(sql, "BASE_BACKUP (VERIFY_CHECKSUMS true)"); - } - - #[test] - fn test_base_backup_sql_with_manifest() { - let opts = BaseBackupOptions { - manifest: Some("yes".to_string()), - ..Default::default() - }; - let sql = PgReplicationConnection::build_base_backup_sql(&opts).unwrap(); - assert_eq!(sql, "BASE_BACKUP (MANIFEST 'yes')"); - } - - #[test] - fn test_base_backup_sql_with_manifest_checksums() { - let opts = BaseBackupOptions { - manifest: Some("yes".to_string()), - manifest_checksums: Some("SHA256".to_string()), - ..Default::default() - }; - let sql = PgReplicationConnection::build_base_backup_sql(&opts).unwrap(); - assert_eq!( - sql, - "BASE_BACKUP (MANIFEST 'yes', MANIFEST_CHECKSUMS 'SHA256')" - ); - } - - #[test] - fn test_base_backup_sql_with_incremental() { - let opts = BaseBackupOptions { - incremental: true, - ..Default::default() - }; - let sql = PgReplicationConnection::build_base_backup_sql(&opts).unwrap(); - assert_eq!(sql, "BASE_BACKUP (INCREMENTAL)"); - } - - #[test] - fn test_base_backup_sql_with_multiple_options() { - let opts = BaseBackupOptions { - label: Some("full_backup".to_string()), - progress: true, - wal: true, - checkpoint: Some("fast".to_string()), - verify_checksums: true, - ..Default::default() - }; - let sql = PgReplicationConnection::build_base_backup_sql(&opts).unwrap(); - assert_eq!( - sql, - "BASE_BACKUP (LABEL 'full_backup', PROGRESS true, CHECKPOINT 'fast', WAL true, VERIFY_CHECKSUMS true)" - ); - } - - #[test] - fn test_base_backup_sql_label_injection() { - let opts = BaseBackupOptions { - label: Some("evil'label".to_string()), - ..Default::default() - }; - let sql = PgReplicationConnection::build_base_backup_sql(&opts).unwrap(); - assert_eq!(sql, "BASE_BACKUP (LABEL 'evil''label')"); - } - - // ======================================== - // build_start_replication_sql tests - // ======================================== - - #[test] - fn test_start_replication_sql_with_zero_lsn() { - let sql = PgReplicationConnection::build_start_replication_sql( - "my_slot", - INVALID_XLOG_REC_PTR, - &[("proto_version", "1"), ("publication_names", "my_pub")], - ) - .unwrap(); - assert_eq!( - sql, - r#"START_REPLICATION SLOT "my_slot" LOGICAL 0/0 ("proto_version" '1', "publication_names" 'my_pub')"# - ); - } - - #[test] - fn test_start_replication_sql_with_valid_lsn() { - let lsn: XLogRecPtr = 0x0000_0001_0000_0000; // 1/0 - let sql = PgReplicationConnection::build_start_replication_sql( - "test_slot", - lsn, - &[("proto_version", "2")], - ) - .unwrap(); - assert!(sql.contains("START_REPLICATION SLOT \"test_slot\" LOGICAL")); - assert!(sql.contains("(\"proto_version\" '2')")); - // Should NOT contain "0/0" since we provided a valid LSN - assert!(!sql.contains("0/0")); - } - - #[test] - fn test_start_replication_sql_with_multiple_options() { - let sql = PgReplicationConnection::build_start_replication_sql( - "slot1", - INVALID_XLOG_REC_PTR, - &[ - ("proto_version", "1"), - ("publication_names", "pub1"), - ("messages", "true"), - ], - ) - .unwrap(); - assert!( - sql.contains(r#""proto_version" '1', "publication_names" 'pub1', "messages" 'true'"#) - ); - } - - #[test] - fn test_start_replication_sql_empty_options() { - let sql = PgReplicationConnection::build_start_replication_sql( - "slot1", - INVALID_XLOG_REC_PTR, - &[], - ) - .unwrap(); - assert_eq!(sql, r#"START_REPLICATION SLOT "slot1" LOGICAL 0/0"#); - } - - #[test] - fn test_start_replication_sql_option_injection() { - let sql = PgReplicationConnection::build_start_replication_sql( - r#"evil"slot"#, - INVALID_XLOG_REC_PTR, - &[("key", "it's")], - ) - .unwrap(); - // Slot name should be quoted, value should be sanitized - assert!(sql.contains(r#""evil""slot""#)); - assert!(sql.contains("'it''s'")); - } - - #[test] - fn test_start_replication_sql_single_option() { - let sql = PgReplicationConnection::build_start_replication_sql( - "my_slot", - INVALID_XLOG_REC_PTR, - &[("proto_version", "1")], - ) - .unwrap(); - assert_eq!( - sql, - r#"START_REPLICATION SLOT "my_slot" LOGICAL 0/0 ("proto_version" '1')"# - ); - } - - // ======================================== - // build_alter_slot_sql tests - // ======================================== - - #[test] - fn test_alter_slot_sql_two_phase_true() { - let sql = - PgReplicationConnection::build_alter_slot_sql("my_slot", Some(true), None).unwrap(); - assert_eq!(sql, r#"ALTER_REPLICATION_SLOT "my_slot" (TWO_PHASE true);"#); - } - - #[test] - fn test_alter_slot_sql_two_phase_false() { - let sql = - PgReplicationConnection::build_alter_slot_sql("my_slot", Some(false), None).unwrap(); - assert_eq!( - sql, - r#"ALTER_REPLICATION_SLOT "my_slot" (TWO_PHASE false);"# - ); - } - - #[test] - fn test_alter_slot_sql_failover_true() { - let sql = - PgReplicationConnection::build_alter_slot_sql("my_slot", None, Some(true)).unwrap(); - assert_eq!(sql, r#"ALTER_REPLICATION_SLOT "my_slot" (FAILOVER true);"#); - } - - #[test] - fn test_alter_slot_sql_both_options() { - let sql = PgReplicationConnection::build_alter_slot_sql("my_slot", Some(true), Some(false)) - .unwrap(); - assert_eq!( - sql, - r#"ALTER_REPLICATION_SLOT "my_slot" (TWO_PHASE true, FAILOVER false);"# - ); - } - - #[test] - fn test_alter_slot_sql_no_options_error() { - let result = PgReplicationConnection::build_alter_slot_sql("my_slot", None, None); - assert!(result.is_err()); - let err = result.unwrap_err(); - assert!( - err.to_string() - .contains("At least one option must be specified"), - "Expected option error, got: {err}" - ); - } - - #[test] - fn test_alter_slot_sql_injection() { - let sql = PgReplicationConnection::build_alter_slot_sql(r#"evil"slot"#, Some(true), None) - .unwrap(); - assert!(sql.contains(r#""evil""slot""#)); - } - - // ======================================== - // build_start_physical_replication_sql tests - // ======================================== - - #[test] - fn test_physical_replication_sql_no_slot_zero_lsn() { - let sql = PgReplicationConnection::build_start_physical_replication_sql( - None, - INVALID_XLOG_REC_PTR, - None, - ) - .unwrap(); - assert_eq!(sql, "START_REPLICATION PHYSICAL 0/0"); - } - - #[test] - fn test_physical_replication_sql_with_slot() { - let sql = PgReplicationConnection::build_start_physical_replication_sql( - Some("phys_slot"), - INVALID_XLOG_REC_PTR, - None, - ) - .unwrap(); - assert_eq!(sql, r#"START_REPLICATION SLOT "phys_slot" PHYSICAL 0/0"#); - } - - #[test] - fn test_physical_replication_sql_with_timeline() { - let sql = PgReplicationConnection::build_start_physical_replication_sql( - None, - INVALID_XLOG_REC_PTR, - Some(3), - ) - .unwrap(); - assert_eq!(sql, "START_REPLICATION PHYSICAL 0/0 TIMELINE 3"); - } - - #[test] - fn test_physical_replication_sql_with_valid_lsn() { - let lsn: XLogRecPtr = 0x0000_0001_0000_0000; // 1/0 - let sql = - PgReplicationConnection::build_start_physical_replication_sql(None, lsn, None).unwrap(); - assert!(sql.starts_with("START_REPLICATION PHYSICAL ")); - assert!(!sql.contains("0/0")); - } - - #[test] - fn test_physical_replication_sql_all_options() { - let sql = PgReplicationConnection::build_start_physical_replication_sql( - Some("my_slot"), - INVALID_XLOG_REC_PTR, - Some(2), - ) - .unwrap(); - assert_eq!( - sql, - r#"START_REPLICATION SLOT "my_slot" PHYSICAL 0/0 TIMELINE 2"# - ); - } - - #[test] - fn test_physical_replication_sql_slot_injection() { - let sql = PgReplicationConnection::build_start_physical_replication_sql( - Some(r#"evil"slot"#), - INVALID_XLOG_REC_PTR, - None, - ) - .unwrap(); - assert!(sql.contains(r#""evil""slot""#)); - } - // ======================================== // Public method error propagation tests // (exercises the `?` on builder calls in production methods) diff --git a/src/connection/native/connection.rs b/src/connection/native/connection.rs index bf02fc9..deb9f13 100644 --- a/src/connection/native/connection.rs +++ b/src/connection/native/connection.rs @@ -815,20 +815,17 @@ impl NativeConnection { output_plugin: Option<&str>, options: &ReplicationSlotOptions, ) -> Result { - let sql = Self::build_create_slot_sql(slot_name, slot_type, output_plugin, options)?; + crate::sql_builder::check_create_slot_version(self.server_version(), slot_type, options)?; + let sql = crate::sql_builder::build_create_slot_sql( + slot_name, + slot_type, + output_plugin, + options, + )?; debug!("Creating replication slot: {}", sql); self.exec(&sql) } - fn build_create_slot_sql( - slot_name: &str, - slot_type: SlotType, - output_plugin: Option<&str>, - options: &ReplicationSlotOptions, - ) -> Result { - crate::sql_builder::build_create_slot_sql(slot_name, slot_type, output_plugin, options) - } - /// Alter a replication slot (logical slots only). pub fn alter_replication_slot( &mut self, @@ -836,6 +833,7 @@ impl NativeConnection { two_phase: Option, failover: Option, ) -> Result { + crate::sql_builder::check_alter_slot_version(self.server_version(), two_phase)?; let sql = crate::sql_builder::build_alter_slot_sql(slot_name, two_phase, failover)?; debug!("Altering replication slot: {}", sql); @@ -844,13 +842,9 @@ impl NativeConnection { Ok(result) } - fn build_drop_slot_sql(slot_name: &str, wait: bool) -> Result { - crate::sql_builder::build_drop_slot_sql(slot_name, wait) - } - /// Drop a replication slot. pub fn drop_replication_slot(&mut self, slot_name: &str, wait: bool) -> Result<()> { - let sql = Self::build_drop_slot_sql(slot_name, wait)?; + let sql = crate::sql_builder::build_drop_slot_sql(slot_name, wait)?; debug!("Dropping replication slot: {}", sql); let result = self.exec(&sql)?; if !result.is_ok() { @@ -866,16 +860,13 @@ impl NativeConnection { Ok(()) } - fn build_read_slot_sql(slot_name: &str) -> Result { - crate::sql_builder::build_read_slot_sql(slot_name) - } - /// Read information about a replication slot. pub fn read_replication_slot( &mut self, slot_name: &str, ) -> Result { - let sql = Self::build_read_slot_sql(slot_name)?; + crate::sql_builder::check_read_slot_version(self.server_version())?; + let sql = crate::sql_builder::build_read_slot_sql(slot_name)?; debug!("Reading replication slot: {}", sql); let result = self.exec(&sql)?; if !result.is_ok() { @@ -936,6 +927,7 @@ impl NativeConnection { /// Start a base backup with options. pub fn base_backup(&mut self, options: &BaseBackupOptions) -> Result { + crate::sql_builder::check_base_backup_version(self.server_version(), options)?; let sql = crate::sql_builder::build_base_backup_sql(options)?; debug!("Starting base backup: {}", sql); @@ -1323,80 +1315,6 @@ mod tests { ); } - // === build_create_slot_sql === - - #[test] - fn test_slot_sql_delegates_to_builder() { - // Delegates to sql_builder::build_create_slot_sql (exhaustively covered by create_slot_sql_cases there). - let opts = ReplicationSlotOptions { - failover: true, - ..Default::default() - }; - let sql = NativeConnection::build_create_slot_sql( - "slot", - SlotType::Logical, - Some("pgoutput"), - &opts, - ) - .unwrap(); - assert_eq!( - sql, - r#"CREATE_REPLICATION_SLOT "slot" LOGICAL "pgoutput" (FAILOVER);"# - ); - } - - // === build_drop_slot_sql === - - #[test] - fn test_build_drop_slot_sql_without_wait() { - assert_eq!( - NativeConnection::build_drop_slot_sql("my_slot", false).unwrap(), - r#"DROP_REPLICATION_SLOT "my_slot";"# - ); - } - - #[test] - fn test_build_drop_slot_sql_with_wait() { - assert_eq!( - NativeConnection::build_drop_slot_sql("my_slot", true).unwrap(), - r#"DROP_REPLICATION_SLOT "my_slot" WAIT;"# - ); - } - - #[test] - fn test_build_drop_slot_sql_injection() { - assert_eq!( - NativeConnection::build_drop_slot_sql(r#"evil"slot"#, false).unwrap(), - r#"DROP_REPLICATION_SLOT "evil""slot";"# - ); - } - - #[test] - fn test_build_drop_slot_sql_injection_with_wait() { - assert_eq!( - NativeConnection::build_drop_slot_sql(r#"evil"slot"#, true).unwrap(), - r#"DROP_REPLICATION_SLOT "evil""slot" WAIT;"# - ); - } - - // === build_read_slot_sql === - - #[test] - fn test_build_read_slot_sql_basic() { - assert_eq!( - NativeConnection::build_read_slot_sql("my_slot").unwrap(), - r#"READ_REPLICATION_SLOT "my_slot";"# - ); - } - - #[test] - fn test_build_read_slot_sql_injection() { - assert_eq!( - NativeConnection::build_read_slot_sql(r#"evil"slot"#).unwrap(), - r#"READ_REPLICATION_SLOT "evil""slot";"# - ); - } - // === ensure_replication_mode, is_alive, server_version, close_connection, Drop === #[tokio::test] @@ -1421,6 +1339,49 @@ mod tests { assert_eq!(conn.server_version(), 160000); } + #[tokio::test] + async fn test_create_slot_preflight_rejects_failover_below_pg17() { + let mut conn = NativeConnection::null_for_testing(); + let opts = ReplicationSlotOptions { + failover: true, + ..Default::default() + }; + let err = conn + .create_replication_slot_with_options("s", SlotType::Logical, Some("pgoutput"), &opts) + .unwrap_err(); + assert!(err.to_string().contains("FAILOVER"), "{err}"); + assert!(err.to_string().contains("17+"), "{err}"); + } + + #[tokio::test] + async fn test_alter_slot_preflight_rejects_below_pg17() { + let mut conn = NativeConnection::null_for_testing(); + let err = conn + .alter_replication_slot("s", None, Some(true)) + .unwrap_err(); + assert!(err.to_string().contains("ALTER_REPLICATION_SLOT"), "{err}"); + assert!(err.to_string().contains("17+"), "{err}"); + } + + #[tokio::test] + async fn test_base_backup_preflight_rejects_incremental_below_pg17() { + let mut conn = NativeConnection::null_for_testing(); + let opts = BaseBackupOptions { + incremental: true, + ..Default::default() + }; + let err = conn.base_backup(&opts).unwrap_err(); + assert!(err.to_string().contains("INCREMENTAL"), "{err}"); + } + + #[tokio::test] + async fn test_read_slot_preflight_passes_on_pg16_then_hits_socket() { + // READ_REPLICATION_SLOT is PG15+, so the PG16 preflight passes; the call + // then fails at the null socket. Exercises the preflight line's Ok path. + let mut conn = NativeConnection::null_for_testing(); + assert!(conn.read_replication_slot("s").is_err()); + } + #[tokio::test] async fn test_close_connection_null_conn() { let mut conn = NativeConnection::null_for_testing(); diff --git a/src/protocol.rs b/src/protocol.rs index fa967b4..c11b84f 100644 --- a/src/protocol.rs +++ b/src/protocol.rs @@ -1365,49 +1365,53 @@ pub(crate) fn key_columns_for_relation( } } -/// Extract schema/table name, replica identity, and key columns for a relation -/// from the cache in `state`. -#[allow(clippy::type_complexity)] +/// Schema/table name, replica identity, and key columns for a relation, plus a borrow of the cached [`RelationInfo`] to decode tuple data against. Resolved from the cache in `state` for an INSERT/UPDATE/DELETE. +pub(crate) struct RelationMetadata<'a> { + pub(crate) schema: Arc, + pub(crate) table: Arc, + pub(crate) replica_identity: ReplicaIdentity, + pub(crate) key_columns: Vec>, + pub(crate) relation: &'a RelationInfo, +} + +/// Look up [`RelationMetadata`] for `relation_id` in `state`, or `None` if the +/// relation has not been seen. pub(crate) fn relation_metadata( state: &ReplicationState, relation_id: u32, key_type: Option, -) -> Option<( - Arc, - Arc, - ReplicaIdentity, - Vec>, - &RelationInfo, -)> { +) -> Option> { let relation = state.get_relation(relation_id)?; - let schema_name = Arc::clone(&relation.namespace); - let table_name = Arc::clone(&relation.relation_name); - - let replica_identity = - ReplicaIdentity::from_byte(relation.replica_identity).unwrap_or(ReplicaIdentity::Default); - - let key_columns = key_columns_for_relation(relation, key_type); - - Some(( - schema_name, - table_name, - replica_identity, - key_columns, + Some(RelationMetadata { + schema: Arc::clone(&relation.namespace), + table: Arc::clone(&relation.relation_name), + replica_identity: ReplicaIdentity::from_byte(relation.replica_identity) + .unwrap_or(ReplicaIdentity::Default), + key_columns: key_columns_for_relation(relation, key_type), relation, - )) + }) } /// Convert a parsed logical replication message into a [`ChangeEvent`], reading /// and updating the relation cache in `state`. /// -/// This is the connection-independent core of the decode path. `Relation` -/// messages update `state` and return `Ok(None)` (or a `Relation` event on a -/// detected schema change), data messages read `state` and return -/// `Ok(Some(event))`, and a data message for an unknown relation logs a warning -/// and returns `Ok(None)`. `lsn` stamps the emitted event. The network -/// `EventStream` delegates to this function, so byte-only consumers get -/// identical results without a connection. +/// This is the connection-independent core of the decode path. Both the network +/// `EventStream` and [`PgOutputDecoder`] delegate to it, so every consumer gets +/// identical results whether or not a connection is held. +/// +/// `Ok(None)` covers two distinct cases the caller cannot tell apart: +/// - a first-seen `Relation` message (cache is populated, no event emitted), and +/// - a data message for a relation not yet in `state` (dropped, logged at WARN). +/// +/// A `Relation` whose schema differs from the cached one emits a `Relation` +/// event before the cache is updated. `lsn` stamps every emitted event. +/// +/// Kept `pub(crate)`: the public entry point is [`PgOutputDecoder`]. This lives +/// in the unconditional `protocol` module (not the network-gated `stream` +/// module), so the ChangeEvent assembly it performs also compiles into the +/// `no_std`/parser-only build — a deliberate size cost for that target in +/// exchange for connection-free decoding. pub(crate) fn message_to_change_event( state: &mut ReplicationState, message: StreamingReplicationMessage, @@ -1507,21 +1511,19 @@ pub(crate) fn message_to_change_event( new_tuple, key_type, } => { - if let Some((schema_name, table_name, replica_identity, key_columns, relation)) = - relation_metadata(state, relation_id, key_type) - { - let old_data = old_tuple.map(|t| t.into_row_data(relation)); - let new_data = new_tuple.into_row_data(relation); + if let Some(meta) = relation_metadata(state, relation_id, key_type) { + let old_data = old_tuple.map(|t| t.into_row_data(meta.relation)); + let new_data = new_tuple.into_row_data(meta.relation); ChangeEvent { event_type: EventType::Update { - schema: schema_name, - table: table_name, + schema: meta.schema, + table: meta.table, relation_oid: relation_id, old_data, new_data, - replica_identity, - key_columns, + replica_identity: meta.replica_identity, + key_columns: meta.key_columns, }, lsn: Lsn::new(lsn), metadata: None, @@ -1537,19 +1539,17 @@ pub(crate) fn message_to_change_event( old_tuple, key_type, } => { - if let Some((schema_name, table_name, replica_identity, key_columns, relation)) = - relation_metadata(state, relation_id, Some(key_type)) - { - let old_data = old_tuple.into_row_data(relation); + if let Some(meta) = relation_metadata(state, relation_id, Some(key_type)) { + let old_data = old_tuple.into_row_data(meta.relation); ChangeEvent { event_type: EventType::Delete { - schema: schema_name, - table: table_name, + schema: meta.schema, + table: meta.table, relation_oid: relation_id, old_data, - replica_identity, - key_columns, + replica_identity: meta.replica_identity, + key_columns: meta.key_columns, }, lsn: Lsn::new(lsn), metadata: None, @@ -1899,7 +1899,21 @@ impl PgOutputDecoder { /// `'w'` XLogData header, no CopyData wrapper), the same contract as /// [`LogicalReplicationParser::parse_wal_message`]. Returns `Ok(None)` for a /// first-seen `Relation` (cache update only) or a data message referencing an - /// unknown relation, and `Ok(Some(event))` otherwise. `lsn` stamps the event. + /// unknown relation, and `Ok(Some(event))` otherwise. + /// + /// `lsn` stamps the emitted event; pass the WAL start position of this + /// message (the network path uses `XLogData.wal_start`). When draining a slot + /// over SQL, pass the `lsn` column returned alongside `data` by + /// `pg_logical_slot_peek_binary_changes` / `pg_logical_slot_get_binary_changes`. + /// + /// # Precondition + /// Feed a *complete* stream from a `START_REPLICATION` (or slot-drain) + /// boundary. pgoutput is self-describing — a `Relation` precedes the first + /// data row for each table and repeats on schema change — so a decoder that + /// starts empty builds its own cache. A data message whose `Relation` has not + /// yet been seen is **silently dropped** (logged at WARN, returns `Ok(None)`); + /// do not resume mid-stream without replaying the preceding `Relation` + /// messages. #[inline] pub fn decode_message( &mut self, @@ -2113,6 +2127,83 @@ mod tests { } ref other => panic!("expected Insert, got {other:?}"), } + + // Concrete decode of Update/Delete/Truncate on the same warm decoder + // (its cache holds relation 42 from the leading Relation at 0x10), + // asserting golden field values instead of re-deriving them through the + // function under test — this removes the reference-oracle circularity. + let ev = d + .decode_message( + encode_message_to_bytes(&messages[2].0, VERSION as u8).freeze(), + Lsn::new(0x30), + ) + .unwrap() + .expect("update yields an event"); + assert_eq!(ev.lsn, Lsn::new(0x30)); + match ev.event_type { + EventType::Update { + ref schema, + ref table, + relation_oid, + ref old_data, + ref new_data, + ref key_columns, + .. + } => { + assert_eq!(&**schema, "public"); + assert_eq!(&**table, "users"); + assert_eq!(relation_oid, 42); + assert!(old_data.is_none(), "this UPDATE carried no old tuple"); + assert_eq!(new_data.get("id").unwrap(), "1"); + assert_eq!(new_data.get("name").unwrap(), "bob"); + // key_type None on UPDATE falls back to the relation's key columns. + assert_eq!(key_columns, &vec![Arc::from("id")]); + } + ref other => panic!("expected Update, got {other:?}"), + } + + let ev = d + .decode_message( + encode_message_to_bytes(&messages[3].0, VERSION as u8).freeze(), + Lsn::new(0x40), + ) + .unwrap() + .expect("delete yields an event"); + assert_eq!(ev.lsn, Lsn::new(0x40)); + match ev.event_type { + EventType::Delete { + ref schema, + ref table, + relation_oid, + ref old_data, + ref key_columns, + .. + } => { + assert_eq!(&**schema, "public"); + assert_eq!(&**table, "users"); + assert_eq!(relation_oid, 42); + assert_eq!(old_data.get("id").unwrap(), "1"); + assert!(old_data.get("name").unwrap().is_null(), "name was NULL"); + // key_type 'K' → replica-identity / key columns. + assert_eq!(key_columns, &vec![Arc::from("id")]); + } + ref other => panic!("expected Delete, got {other:?}"), + } + + let ev = d + .decode_message( + encode_message_to_bytes(&messages[4].0, VERSION as u8).freeze(), + Lsn::new(0x50), + ) + .unwrap() + .expect("truncate yields an event"); + assert_eq!(ev.lsn, Lsn::new(0x50)); + match ev.event_type { + EventType::Truncate(ref tables) => { + assert_eq!(tables, &vec![Arc::from("public.users")]); + } + ref other => panic!("expected Truncate, got {other:?}"), + } } #[test] diff --git a/src/sql_builder.rs b/src/sql_builder.rs index 4f732c0..ad18d69 100644 --- a/src/sql_builder.rs +++ b/src/sql_builder.rs @@ -478,6 +478,259 @@ pub fn build_base_backup_sql(options: &BaseBackupOptions) -> Result { Ok(format!("BASE_BACKUP{}", build_sql_options(&opts))) } +// The builders above are pure and hold no server-version state by design. The preflight below is the separate, equally-pure gate a *connection* applies (with its known `server_version()`) before issuing a version-gated command, so an unsupported option fails with an actionable client-side error instead of an opaque server-side `syntax error`. Only the `libpq` / `rustls-tls` backends call these, so the whole group lives in a feature-gated module: the parser-only / no_std (`--no-default-features`) build compiles it out entirely — no callers, no `dead_code`, no per-item `allow`. +#[cfg(any(feature = "libpq", feature = "rustls-tls"))] +mod version_preflight { + use super::*; + + /// PostgreSQL 15.0 on the `server_version` (`PQserverVersion`) scale + /// (major * 10000 + minor, e.g. 14.23 → 140023). `server_version == 0` (unknown) passes. + const PG15: i32 = 150000; + /// PostgreSQL 17.0 on the `server_version` scale. + const PG17: i32 = 170000; + /// PostgreSQL 18.0 on the `server_version` scale. + const PG18: i32 = 180000; + + /// Format a `PQserverVersion`-scale integer as `major.minor` for error messages. + fn format_server_version(v: i32) -> String { + format!("{}.{}", v / 10000, v % 10000) + } + + /// Preflight `CREATE_REPLICATION_SLOT` options against the server version. + /// + /// Returns a clear [`ReplicationError::config`] when the connected server is too old for a requested option, instead of letting the server reject the emitted SQL with a bare `syntax error`. `server_version == 0` (unknown) passes. + /// + /// Gates: `TWO_PHASE` requires PostgreSQL 15+, `FAILOVER` requires PostgreSQL 17+. + pub(crate) fn check_create_slot_version( + server_version: i32, + slot_type: SlotType, + options: &ReplicationSlotOptions, + ) -> Result<()> { + if server_version == 0 || !matches!(slot_type, SlotType::Logical) { + return Ok(()); + } + if options.two_phase && server_version < PG15 { + return Err(ReplicationError::config(format!( + "TWO_PHASE replication slots require PostgreSQL 15+, but the server reports {}", + format_server_version(server_version) + ))); + } + if options.failover && server_version < PG17 { + return Err(ReplicationError::config(format!( + "FAILOVER replication slots require PostgreSQL 17+, but the server reports {}", + format_server_version(server_version) + ))); + } + Ok(()) + } + + /// Preflight `ALTER_REPLICATION_SLOT` against the server version. + /// + /// Gates: the command itself requires PostgreSQL 17+; altering `TWO_PHASE` additionally requires PostgreSQL 18+ (PostgreSQL 17 rejects `two_phase` with `unrecognized option`). `server_version == 0` (unknown) passes. `FAILOVER` needs no separate gate — it was introduced with the command in PostgreSQL 17, so the command-level check already covers it. + pub(crate) fn check_alter_slot_version( + server_version: i32, + two_phase: Option, + ) -> Result<()> { + if server_version == 0 { + return Ok(()); + } + if server_version < PG17 { + return Err(ReplicationError::config(format!( + "ALTER_REPLICATION_SLOT requires PostgreSQL 17+, but the server reports {}", + format_server_version(server_version) + ))); + } + if two_phase.is_some() && server_version < PG18 { + return Err(ReplicationError::config(format!( + "altering TWO_PHASE on a replication slot requires PostgreSQL 18+, \ + but the server reports {}", + format_server_version(server_version) + ))); + } + Ok(()) + } + + /// Preflight `READ_REPLICATION_SLOT` against the server version. + /// + /// Gate: requires PostgreSQL 15+. `server_version == 0` (unknown) passes. + pub(crate) fn check_read_slot_version(server_version: i32) -> Result<()> { + if server_version != 0 && server_version < PG15 { + return Err(ReplicationError::config(format!( + "READ_REPLICATION_SLOT requires PostgreSQL 15+, but the server reports {}", + format_server_version(server_version) + ))); + } + Ok(()) + } + + /// Preflight `BASE_BACKUP` options against the server version. + /// + /// Gates: `INCREMENTAL` requires PostgreSQL 17+; any option at all switches the + /// command to the PostgreSQL 15+ parenthesized generic-option grammar, which + /// PostgreSQL 14 cannot parse (a bare `BASE_BACKUP` still works everywhere). + /// `server_version == 0` (unknown) passes. + pub(crate) fn check_base_backup_version( + server_version: i32, + options: &BaseBackupOptions, + ) -> Result<()> { + if server_version == 0 { + return Ok(()); + } + if options.incremental && server_version < PG17 { + return Err(ReplicationError::config(format!( + "INCREMENTAL base backup requires PostgreSQL 17+, but the server reports {}", + format_server_version(server_version) + ))); + } + if server_version < PG15 && base_backup_has_options(options) { + return Err(ReplicationError::config(format!( + "BASE_BACKUP with options uses the PostgreSQL 15+ parenthesized grammar, \ + but the server reports {}; omit all options or use PostgreSQL 15+", + format_server_version(server_version) + ))); + } + Ok(()) + } + + /// Whether any `BASE_BACKUP` option is set — i.e. the emitted SQL would use the parenthesized option list rather than a bare `BASE_BACKUP`. + fn base_backup_has_options(o: &BaseBackupOptions) -> bool { + o.label.is_some() + || o.target.is_some() + || o.target_detail.is_some() + || o.progress + || o.checkpoint.is_some() + || o.wal + || o.wait + || o.compression.is_some() + || o.compression_detail.is_some() + || o.max_rate.is_some() + || o.tablespace_map + || o.verify_checksums + || o.manifest.is_some() + || o.manifest_checksums.is_some() + || o.incremental + } + + #[cfg(test)] + mod preflight_tests { + use super::*; + + fn logical_opts(two_phase: bool, failover: bool) -> ReplicationSlotOptions { + ReplicationSlotOptions { + two_phase, + failover, + ..Default::default() + } + } + + #[test] + fn preflight_create_slot_gates_two_phase_pg15() { + let opts = logical_opts(true, false); + let err = check_create_slot_version(140023, SlotType::Logical, &opts).unwrap_err(); + assert!(err.to_string().contains("TWO_PHASE"), "{err}"); + assert!(err.to_string().contains("15+"), "{err}"); + assert!(check_create_slot_version(150000, SlotType::Logical, &opts).is_ok()); + assert!(check_create_slot_version(180000, SlotType::Logical, &opts).is_ok()); + } + + #[test] + fn preflight_create_slot_gates_failover_pg17() { + let opts = logical_opts(false, true); + let err = check_create_slot_version(160000, SlotType::Logical, &opts).unwrap_err(); + assert!(err.to_string().contains("FAILOVER"), "{err}"); + assert!(err.to_string().contains("17+"), "{err}"); + assert!(check_create_slot_version(170000, SlotType::Logical, &opts).is_ok()); + } + + #[test] + fn preflight_create_slot_unknown_version_passes() { + // server_version 0 (undeterminable) never blocks. + let opts = logical_opts(true, true); + assert!(check_create_slot_version(0, SlotType::Logical, &opts).is_ok()); + } + + #[test] + fn preflight_create_slot_physical_ignores_logical_gates() { + // two_phase/failover gates are logical-only; the builder separately rejects failover on physical slots. + let opts = logical_opts(true, true); + assert!(check_create_slot_version(140000, SlotType::Physical, &opts).is_ok()); + } + + #[test] + fn preflight_create_slot_default_options_pass_on_any_version() { + let opts = ReplicationSlotOptions::default(); + assert!(check_create_slot_version(140000, SlotType::Logical, &opts).is_ok()); + assert!(check_create_slot_version(0, SlotType::Logical, &opts).is_ok()); + } + + #[test] + fn preflight_alter_slot_requires_pg17() { + let err = check_alter_slot_version(160000, None).unwrap_err(); + assert!(err.to_string().contains("ALTER_REPLICATION_SLOT"), "{err}"); + assert!(err.to_string().contains("17+"), "{err}"); + assert!(check_alter_slot_version(170000, None).is_ok()); + assert!(check_alter_slot_version(0, None).is_ok()); + } + + #[test] + fn preflight_alter_slot_two_phase_requires_pg18() { + let err = check_alter_slot_version(170004, Some(true)).unwrap_err(); + assert!(err.to_string().contains("TWO_PHASE"), "{err}"); + assert!(err.to_string().contains("18+"), "{err}"); + assert!(check_alter_slot_version(180000, Some(true)).is_ok()); + // A non-two_phase alter (e.g. failover-only) is fine on PG17. + assert!(check_alter_slot_version(170000, None).is_ok()); + } + + #[test] + fn preflight_read_slot_requires_pg15() { + let err = check_read_slot_version(140023).unwrap_err(); + assert!(err.to_string().contains("READ_REPLICATION_SLOT"), "{err}"); + assert!(err.to_string().contains("15+"), "{err}"); + assert!(check_read_slot_version(150000).is_ok()); + assert!(check_read_slot_version(0).is_ok()); + } + + #[test] + fn preflight_base_backup_options_require_pg15() { + let with_opt = BaseBackupOptions { + progress: true, + ..Default::default() + }; + assert!(check_base_backup_version(140023, &with_opt).is_err()); + assert!(check_base_backup_version(150000, &with_opt).is_ok()); + // Bare BASE_BACKUP (no options) parses on any version. + let bare = BaseBackupOptions::default(); + assert!(check_base_backup_version(140023, &bare).is_ok()); + } + + #[test] + fn preflight_base_backup_incremental_requires_pg17() { + let inc = BaseBackupOptions { + incremental: true, + ..Default::default() + }; + let err = check_base_backup_version(160000, &inc).unwrap_err(); + assert!(err.to_string().contains("INCREMENTAL"), "{err}"); + assert!(check_base_backup_version(170000, &inc).is_ok()); + assert!(check_base_backup_version(0, &inc).is_ok()); + } + + #[test] + fn format_server_version_renders_major_minor() { + assert_eq!(format_server_version(140023), "14.23"); + assert_eq!(format_server_version(150000), "15.0"); + assert_eq!(format_server_version(170004), "17.4"); + } + } +} + +#[cfg(any(feature = "libpq", feature = "rustls-tls"))] +pub(crate) use version_preflight::{ + check_alter_slot_version, check_base_backup_version, check_create_slot_version, + check_read_slot_version, +}; + /// Options for building a `CREATE SUBSCRIPTION` SQL statement. /// /// All fields borrow from the caller — no allocation or cloning required. diff --git a/src/stream.rs b/src/stream.rs index d5306cc..a2adcc7 100644 --- a/src/stream.rs +++ b/src/stream.rs @@ -4205,12 +4205,11 @@ mod tests { ); stream.state.add_relation(relation); - let (schema, table, replica_id, key_cols, _rel) = - crate::protocol::relation_metadata(&stream.state, 100, Some('K')).unwrap(); - assert_eq!(&*schema, "myschema"); - assert_eq!(&*table, "mytable"); - assert_eq!(replica_id, ReplicaIdentity::Default); - assert_eq!(key_cols, vec![Arc::from("pk_col")]); + let meta = crate::protocol::relation_metadata(&stream.state, 100, Some('K')).unwrap(); + assert_eq!(&*meta.schema, "myschema"); + assert_eq!(&*meta.table, "mytable"); + assert_eq!(meta.replica_identity, ReplicaIdentity::Default); + assert_eq!(meta.key_columns, vec![Arc::from("pk_col")]); } #[test] @@ -4720,12 +4719,11 @@ mod tests { ); stream.state.add_relation(relation); - let (schema, table, _ri, _keys, _rel) = - crate::protocol::relation_metadata(&stream.state, 100, None).unwrap(); + let meta = crate::protocol::relation_metadata(&stream.state, 100, None).unwrap(); // With empty namespace, full_name is ".just_table", split gives ["", "just_table"] - assert_eq!(&*table, "just_table"); + assert_eq!(&*meta.table, "just_table"); // Schema should still work - assert_eq!(&*schema, ""); + assert_eq!(&*meta.schema, ""); } /// Build a synthetic WAL message: 'w' + start_lsn(8) + end_lsn(8) + send_time(8) + payload