replication version preflight

.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

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

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(&regular_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(&regular_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");
+}