Snapshot minting's interval Env is not being leveraged.

.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 <app>_<shard>.
 # STREAMGRES_SCHEMAS=public,xyne_0
 # The publication the change feed streams; the default is retained for

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

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<String>,
+    /// 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_<shard>` schema by default).
     pub schemas: Vec<String>,
@@ -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)?,

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()

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:?}"
+                ));
+            }
+        }
+    }
+}

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,15 +109,19 @@ 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)
     }
 }
 
 /// The single owner of an engine (see the module docs).
 ///
 /// - `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<E: Engine> {
     /// their freeing (an allocation per row) happens off the engine's
     /// thread; the frames hold the ones that matter by reference.
     landed: Vec<Vec<(DataFrameKey, DataFrameRow)>>,
-    recent: VecDeque<Delivered>,
+    recent: HashMap<TableName, VecDeque<Delivered>>,
+    recent_order: VecDeque<(Lsn, TableName)>,
     position: Lsn,
     floor: Lsn,
     stats: SyncStats,
@@ -140,7 +147,8 @@ impl<E: Engine> Runtime<E> {
             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<E: Engine> Runtime<E> {
 
     /// 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<E: Engine> Runtime<E> {
         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<E: Engine> Runtime<E> {
                 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<E: Engine> Runtime<E> {
             .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);
+            }
         }
     }
 }