diff --git a/.env.example b/.env.example index 4535926..e84b895 100644 --- a/.env.example +++ b/.env.example @@ -13,6 +13,11 @@ STREAMGRES_PG_DSN=postgresql://postgres:postgres@localhost:5432/xyne_dev_db # snapshot minting. Change this host/port to PgBouncer to pool SQL reads; it # must route to the same PostgreSQL database. Defaults to the direct DSN. STREAMGRES_READ_PG_DSN=postgresql://postgres:postgres@localhost:5432/xyne_dev_db +# Optional direct primary connection for one periodic WAL snapshot writer per +# database. Use a separate login granted EXECUTE on +# pg_catalog.pg_log_standby_snapshot(); all pods must name the same database. +# STREAMGRES_MASTER_PG_DSN=postgresql://snapshot_user:password@primary:5432/xyne_dev_db +# STREAMGRES_STANDBY_SNAPSHOT_INTERVAL_MS=5000 # Schemas whose tables the catalog carries; default: public and _. # STREAMGRES_SCHEMAS=public,xyne_0 # The publication the change feed streams; the default is retained for diff --git a/docs/deploy.md b/docs/deploy.md index 8ac93e9..391700a 100644 --- a/docs/deploy.md +++ b/docs/deploy.md @@ -80,6 +80,29 @@ branch name and the short commit SHA. directly to PostgreSQL for the logical feed and snapshot minting. Replication must not use the read pool. +When the feed runs on a physical standby, set `STREAMGRES_MASTER_PG_DSN` to +the same database on its primary. Each server opens one additional direct +connection. One server holds a session advisory lock and calls +`pg_log_standby_snapshot()` immediately on leadership, then every +`STREAMGRES_STANDBY_SNAPSHOT_INTERVAL_MS` milliseconds (default 5000). +Other servers wait for that lock and take over when the leader's database +session closes. Leave the primary DSN unset to disable the worker. Do not +route this connection through PgBouncer transaction pooling. All servers +must use the same database and interval. + +The primary connection can use a separate login without replication or table +write privileges. Grant it only the additional function privilege it needs: + +```sql +CREATE ROLE streamgres_snapshot LOGIN NOSUPERUSER NOCREATEDB NOCREATEROLE NOREPLICATION; +GRANT EXECUTE ON FUNCTION pg_catalog.pg_log_standby_snapshot() + TO streamgres_snapshot; +``` + +Set its password and connection rules separately. If the role has a connection +limit, allow at least one connection per server so followers can wait for the +lock. + **The server writes nothing to the database it follows.** It reads rows and the replication stream. Its position in the stream comes from each commit and from the keepalives PostgreSQL sends anyway. So `STREAMGRES_PG_DSN` can point diff --git a/src/client/config.rs b/src/client/config.rs index f1a4f46..f0484f4 100644 --- a/src/client/config.rs +++ b/src/client/config.rs @@ -29,6 +29,11 @@ pub struct Config { /// Defaults to `dsn`; it must reach the same database as the direct DSN /// because reads import snapshots minted through the direct connection. pub read_dsn: String, + /// Dedicated, direct primary connection for periodic standby snapshot + /// records. When absent, the snapshot worker is disabled. + pub master_pg_dsn: Option, + /// Time between primary standby snapshot records (5000 ms by default). + pub standby_snapshot_interval: Duration, /// `STREAMGRES_SCHEMAS`: the schemas whose tables the catalog carries /// (`public` plus the app's `zero_` schema by default). pub schemas: Vec, @@ -227,6 +232,11 @@ impl Config { let dsn = first(&["STREAMGRES_PG_DSN", "ZERO_UPSTREAM_DB"]) .ok_or("STREAMGRES_PG_DSN (or ZERO_UPSTREAM_DB) must name the database")?; let read_dsn = first(&["STREAMGRES_READ_PG_DSN"]).unwrap_or_else(|| dsn.clone()); + let master_pg_dsn = first(&["STREAMGRES_MASTER_PG_DSN"]); + let standby_snapshot_interval = millis("STREAMGRES_STANDBY_SNAPSHOT_INTERVAL_MS", 5_000)?; + if master_pg_dsn.is_some() && standby_snapshot_interval.is_zero() { + return Err("STREAMGRES_STANDBY_SNAPSHOT_INTERVAL_MS must be positive".to_owned()); + } let query_url = first(&["STREAMGRES_QUERY_URL", "ZERO_QUERY_URL"]) .ok_or("STREAMGRES_QUERY_URL (or ZERO_QUERY_URL) must name the query endpoint")?; let mutate_url = first(&["STREAMGRES_MUTATE_URL", "ZERO_MUTATE_URL"]) @@ -318,6 +328,8 @@ impl Config { base_path, dsn, read_dsn, + master_pg_dsn, + standby_snapshot_interval, schemas, slot: threads::slot_name(), slot_cleanup_age: millis("STREAMGRES_SLOT_CLEANUP_AGE_MS", 0)?, diff --git a/src/client/mod.rs b/src/client/mod.rs index 85ac6a8..990cb69 100644 --- a/src/client/mod.rs +++ b/src/client/mod.rs @@ -59,6 +59,7 @@ pub mod plan; pub mod protocol; pub mod sampler; pub mod schema; +mod standby_snapshot; pub mod transform; pub mod warm; pub mod wire; @@ -131,6 +132,15 @@ pub fn serve(config: Config) -> Result<(), String> { &settings.dsn, &settings.ddl_trigger, ))?; + if let Some(dsn) = &config.master_pg_dsn { + standby_snapshot::start( + dsn, + config.standby_snapshot_interval, + settings.keepalive, + shutdown.clone(), + &server, + )?; + } let reads = tokio::runtime::Builder::new_multi_thread() .worker_threads(settings.read_threads) .enable_all() diff --git a/src/client/standby_snapshot.rs b/src/client/standby_snapshot.rs new file mode 100644 index 0000000..e3e1e34 --- /dev/null +++ b/src/client/standby_snapshot.rs @@ -0,0 +1,103 @@ +//! One primary-side standby snapshot writer per database, elected with a +//! session advisory lock. The lock and every call use the same direct session. + +use std::time::Duration; + +use tokio_postgres::{Config, NoTls}; + +use crate::log::{log_info, log_warn}; +use crate::shutdown::Shutdown; +use crate::sync::pg::Keepalive; + +// A stable pair shared by every Streamgres pod connected to this database. +const LOCK_CLASS: i32 = 0x5354_4752; // STGR +const LOCK_RESOURCE: i32 = 0x534e_4150; // SNAP +const RETRY_DELAY: Duration = Duration::from_secs(5); +const CALL_TIMEOUT: Duration = Duration::from_secs(15); + +pub fn start( + dsn: &str, + interval: Duration, + keepalive: Keepalive, + shutdown: Shutdown, + runtime: &tokio::runtime::Runtime, +) -> Result<(), String> { + let mut config: Config = dsn + .parse() + .map_err(|error| format!("STREAMGRES_MASTER_PG_DSN: {error}"))?; + config.application_name("streamgres-standby-snapshot"); + config.connect_timeout(Duration::from_secs(10)); + keepalive.apply(&mut config); + runtime.spawn(run(config, interval, shutdown)); + Ok(()) +} + +async fn run(config: Config, interval: Duration, shutdown: Shutdown) { + loop { + let connected = tokio::select! { + _ = shutdown.wait() => return, + result = config.connect(NoTls) => result, + }; + match connected { + Ok((client, connection)) => { + // Driving the connection is essential: it also detects a broken + // session while a follower waits for the advisory lock. + let driver = tokio::spawn(connection); + let outcome = lead(&client, interval, &shutdown).await; + drop(client); + driver.abort(); // Close the session and release its lock. + if shutdown.requested() { + return; + } + if let Err(error) = outcome { + log_warn!("primary standby snapshot worker: {error}"); + } + } + Err(error) => log_warn!("primary standby snapshot connection: {error}"), + } + tokio::select! { + _ = shutdown.wait() => return, + _ = tokio::time::sleep(RETRY_DELAY) => {}, + } + } +} + +async fn lead( + client: &tokio_postgres::Client, + interval: Duration, + shutdown: &Shutdown, +) -> Result<(), String> { + tokio::select! { + _ = shutdown.wait() => return Ok(()), + result = client.query( + "SELECT pg_catalog.pg_advisory_lock($1::integer, $2::integer)", + &[&LOCK_CLASS, &LOCK_RESOURCE], + ) => result.map_err(|error| format!("acquiring leader lock: {error}"))?, + }; + log_info!("primary standby snapshot leader acquired the database lock"); + + let mut ticks = tokio::time::interval(interval); + ticks.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); + loop { + tokio::select! { + _ = shutdown.wait() => return Ok(()), + _ = ticks.tick() => {}, + } + let called = tokio::select! { + _ = shutdown.wait() => return Ok(()), + result = tokio::time::timeout( + CALL_TIMEOUT, + client.simple_query("SELECT pg_catalog.pg_log_standby_snapshot()"), + ) => result, + }; + match called { + Ok(Ok(_)) => {} + Ok(Err(error)) => return Err(format!("logging standby snapshot: {error}")), + Err(_) => { + return Err(format!( + "logging standby snapshot exceeded {CALL_TIMEOUT:?}" + )); + } + } + } +} diff --git a/src/sync/runtime.rs b/src/sync/runtime.rs index 50f3501..8938b34 100644 --- a/src/sync/runtime.rs +++ b/src/sync/runtime.rs @@ -21,14 +21,14 @@ //! anything. Nothing waits: a read behind the stream is caught up, and a //! read ahead of it cannot exist. //! -//! The delivered writes are kept in a buffer bounded below by the +//! The delivered writes are kept by table in buffers bounded below by the //! **floor**: the lowest location a read can still be positioned at, //! which the driver learns from the storage and sets here, taken together //! with the floors the reads in flight were issued under. A read the //! driver could not run (no snapshot available yet) is parked and handed //! out again the next time the stream moves. -use std::collections::{HashMap, VecDeque}; +use std::collections::{HashMap, HashSet, VecDeque}; use std::fmt; use crate::ivm::{ @@ -77,6 +77,7 @@ struct Delivered { /// - `rows_completed`: such writes' images the feed left columns out of, /// completed from the result's image of the row. /// - `writes_buffered`: writes remembered for reads to be brought up. +/// - `writes_scanned`: retained writes examined while bringing reads up. #[derive(Debug, Clone, Default, PartialEq, Eq)] pub struct SyncStats { pub reads_issued: u64, @@ -88,6 +89,7 @@ pub struct SyncStats { pub rows_completed: u64, pub rows_added: u64, pub writes_buffered: u64, + pub writes_scanned: u64, } impl fmt::Display for SyncStats { @@ -107,7 +109,8 @@ impl fmt::Display for SyncStats { )?; writeln!(f, "rows completed ............. {}", self.rows_completed)?; writeln!(f, "rows added late ............ {}", self.rows_added)?; - write!(f, "writes buffered ............ {}", self.writes_buffered) + writeln!(f, "writes buffered ............ {}", self.writes_buffered)?; + write!(f, "writes scanned ............. {}", self.writes_scanned) } } @@ -115,7 +118,10 @@ impl fmt::Display for SyncStats { /// /// - `engine`: the engine, exclusively owned. /// - `in_flight`: reads handed to the driver and not yet landed, by id. -/// - `recent`: delivered writes above the floor, oldest first. +/// - `recent`: delivered writes above the floor, grouped by table and in +/// commit order within each table. +/// - `recent_order`: the same writes' positions and tables in global commit +/// order, so trimming touches only writes it removes. /// - `position`: the location of the last write routed (or progress mark /// seen): where the engine is. /// - `floor`: the lowest location a read issued from now on can be at. @@ -127,7 +133,8 @@ pub struct Runtime { /// their freeing (an allocation per row) happens off the engine's /// thread; the frames hold the ones that matter by reference. landed: Vec>, - recent: VecDeque, + recent: HashMap>, + recent_order: VecDeque<(Lsn, TableName)>, position: Lsn, floor: Lsn, stats: SyncStats, @@ -140,7 +147,8 @@ impl Runtime { engine, in_flight: IdMap::default(), landed: Vec::new(), - recent: VecDeque::new(), + recent: HashMap::new(), + recent_order: VecDeque::new(), position: Lsn(0), floor: Lsn(0), stats: SyncStats::default(), @@ -187,7 +195,7 @@ impl Runtime { /// How many delivered writes are kept for bringing reads up. pub fn buffered(&self) -> usize { - self.recent.len() + self.recent_order.len() } /// How many reads are out (parked ones included). @@ -219,10 +227,12 @@ impl Runtime { debug_assert!(at >= self.position, "the stream delivers in commit order"); self.position = self.position.max(at); self.stats.writes_buffered += 1; - self.recent.push_back(Delivered { + let table = write.table().clone(); + self.recent.entry(table.clone()).or_default().push_back(Delivered { at, write: write.clone(), }); + self.recent_order.push_back((at, table)); let mut step = Step { updates: self.engine.route(write), selects: Vec::new(), @@ -418,10 +428,10 @@ impl Runtime { order_rows(&query.order_by, image, worst) == std::cmp::Ordering::Greater }) }; - for delivered in &self.recent { - if delivered.at <= at || delivered.write.table() != table { - continue; - } + let writes = self.recent.get(table); + let start = writes.map_or(0, |writes| writes.partition_point(|write| write.at <= at)); + self.stats.writes_scanned += writes.map_or(0, |writes| writes.len() - start) as u64; + for delivered in writes.into_iter().flat_map(|writes| writes.iter().skip(start)) { let key = delivered.write.pkey_value(); let position = index.get(key).copied(); let Some(image) = delivered.write.new_row_image() else { @@ -476,12 +486,24 @@ impl Runtime { .values() .map(|flight| flight.floor) .fold(self.floor, Lsn::min); - while self - .recent - .front() - .is_some_and(|delivered| delivered.at <= keep_above) - { - self.recent.pop_front(); + // One table visit removes all its old writes; later order entries + // for that table only need to leave the global queue. + let mut pruned_tables = HashSet::new(); + while self.recent_order.front().is_some_and(|(at, _)| *at <= keep_above) { + let (_, table) = self.recent_order.pop_front().expect("front was present"); + if !pruned_tables.insert(table.clone()) { + continue; + } + let empty = { + let writes = self.recent.get_mut(&table).expect("write has a table queue"); + while writes.front().is_some_and(|write| write.at <= keep_above) { + writes.pop_front(); + } + writes.is_empty() + }; + if empty { + self.recent.remove(&table); + } } } }