From 009e5c1f8311a3caf686a383e8e27e48ea60c023 Mon Sep 17 00:00:00 2001 From: serprex <159546+serprex@users.noreply.github.com> Date: Mon, 3 Aug 2026 23:51:21 +0000 Subject: [PATCH 1/8] Optimize vacuum/analyze Non-shape catalog updates can be applied without boundary --- plans/filter.md | 11 + src/bin/stream.rs | 10 + src/filter/catalog_tracker.rs | 71 ++++ src/filter/engine.rs | 221 +++++++++++- src/filter/main_data.rs | 17 + tests/common/inproc_harness.rs | 5 + tests/vacuum_catalog_churn.rs | 610 +++++++++++++++++++++++++++++++++ 7 files changed, 940 insertions(+), 5 deletions(-) create mode 100644 tests/vacuum_catalog_churn.rs diff --git a/plans/filter.md b/plans/filter.md index e8f9e81c..fc24c0a6 100644 --- a/plans/filter.md +++ b/plans/filter.md @@ -51,6 +51,8 @@ State: - `pg_class_filenode: HashMap` — current `pg_class` filenode per database. Empty bootstrap falls through to `rel == 1259` (initial mapped relfilenode) +- `opaque_filenodes` + `opaque_filenode_by_oid` — track statistics catalog + filenodes across rewrites - `relmap_updates`, `pg_class_writes_{decoded,undecoded,oid_in_prefix}`, `seeded_from_source` — diagnostic counters in the manifest @@ -188,6 +190,15 @@ drained members for the same reason: the speculative shapes keyed to them have to die with the tree, on the pump, before a later boundary promotes them +## Statistics traffic + +- Track statistics catalogs separately from catalogs that describe relation + shape +- Ignore statistics-only writes when deciding whether a transaction needs a + boundary +- Keep boundaries for mixed or incompletely observed transactions +- Verify behavior with maintenance workloads and existing DDL cases + ## Rewrite path `src/rewrite.rs::noop_replace` takes complete record buffer (header + diff --git a/src/bin/stream.rs b/src/bin/stream.rs index a4f0b3d1..56382f60 100644 --- a/src/bin/stream.rs +++ b/src/bin/stream.rs @@ -931,6 +931,16 @@ async fn run_session( .seed_from_source(sql_client) .await .context("seed_from_source")?; + let observed_from = stream + .filter_mut() + .seed_observed_from_source(sql_client) + .await + .context("seed observed-from xid")?; + tracing::info!( + target: "walshadow", + observed_from, + "transactions from this xid on are observed whole", + ); tracing::info!( target: "walshadow", added, diff --git a/src/filter/catalog_tracker.rs b/src/filter/catalog_tracker.rs index e270fce0..95d0c52b 100644 --- a/src/filter/catalog_tracker.rs +++ b/src/filter/catalog_tracker.rs @@ -40,6 +40,18 @@ const REL_MAP_FILE_SIZE: usize = 4 + 4 + MAX_MAPPINGS * 8 + 4; // magic + n + ma /// `pg_class.oid`, fixed PG catalog OID pub const PG_CLASS_OID: u32 = 1259; +/// Catalogs that store statistics, including indexes and toast heaps +/// Keep `pg_statistic_ext` outside this list because DDL writes its definition +const OPAQUE_CATALOG_OIDS: &[u32] = &[ + 2619, // pg_statistic + 2696, // pg_statistic_relid_att_inh_index + 2840, // pg_toast_2619 + 2841, // pg_toast_2619_index + 3429, // pg_statistic_ext_data + 3430, // pg_toast_3429 + 3431, // pg_toast_3429_index + 3433, // pg_statistic_ext_data_stxoid_inh_index +]; /// `pg_namespace.oid`; writes to it force capture-all (relcache invals /// enumerate rels only for pg_class/pg_attribute/pg_index/pg_constraint /// changes — PG `src/backend/utils/cache/inval.c` — while namespace rename @@ -59,6 +71,10 @@ pub struct CatalogTracker { /// PG_NAMESPACE_OID`. Unmapped catalog: VACUUM FULL relocates it via /// its own pg_class row, harvested below. pg_namespace_filenode: HashMap, + /// Current statistics catalog filenodes by database + opaque_filenodes: HashSet<(u32, u32)>, + /// Current filenode for each statistics catalog oid + opaque_filenode_by_oid: HashMap<(u32, u32), u32>, relmap_updates: u64, /// pg_class heap writes the decoder couldn't reconstruct (truncated / /// malformed `t_hoff`). OID-prefix-compressed records count in @@ -251,6 +267,9 @@ impl CatalogTracker { if row.oid == PG_NAMESPACE_OID { self.pg_namespace_filenode.insert(db, row.relfilenode); } + if OPAQUE_CATALOG_OIDS.contains(&row.oid) { + self.learn_opaque(db, row.oid, row.relfilenode); + } } if row.oid >= FIRST_NORMAL_OBJECT_ID { user_oid = Some(row.oid); @@ -276,6 +295,23 @@ impl CatalogTracker { } } + /// Return true for a statistics catalog filenode + pub fn is_opaque_catalog(&self, db: u32, rel: u32) -> bool { + if self.opaque_filenodes.contains(&(db, rel)) { + return true; + } + OPAQUE_CATALOG_OIDS.contains(&rel) && !self.opaque_filenode_by_oid.contains_key(&(db, rel)) + } + + fn learn_opaque(&mut self, db: u32, oid: u32, filenode: u32) { + if let Some(prev) = self.opaque_filenode_by_oid.insert((db, oid), filenode) + && prev != filenode + { + self.opaque_filenodes.remove(&(db, prev)); + } + self.opaque_filenodes.insert((db, filenode)); + } + /// True when `(db, rel)` is pg_namespace's current heap — the /// capture-all trigger set. /// @@ -378,6 +414,9 @@ impl CatalogTracker { if catalog_oid == PG_NAMESPACE_OID { self.pg_namespace_filenode.insert(db_node, filenode); } + if OPAQUE_CATALOG_OIDS.contains(&catalog_oid) { + self.learn_opaque(db_node, catalog_oid, filenode); + } } self.seeded_from_source += added as u64; Ok(added) @@ -523,6 +562,38 @@ mod tests { assert!(!t.is_catalog(5, 0)); } + #[test] + fn statistics_catalogs_are_opaque_at_their_bootstrap_filenode() { + let t = CatalogTracker::new(); + assert!(t.is_opaque_catalog(5, 2619)); // pg_statistic + assert!(t.is_opaque_catalog(5, 2696)); // its index + assert!(t.is_opaque_catalog(5, 2840)); // its toast heap + assert!(!t.is_opaque_catalog(5, 1259)); // pg_class + assert!(!t.is_opaque_catalog(5, 16400)); // user rel + } + + #[test] + fn vacuum_full_moves_the_opaque_filenode() { + let mut t = CatalogTracker::new(); + // Simulate VACUUM FULL moving pg_statistic + let data = pg_class_block_data(2619, 40000); + let rec = heap_block_record_with_main( + RmId::Heap, + 0x20, + 5, + 1259, + data, + xl_heap_update_no_compression(), + ); + t.observe(&rec); + assert!(t.is_opaque_catalog(5, 40000)); + assert!(!t.is_opaque_catalog(5, 2619), "old filenode may be reused",); + assert!( + t.is_opaque_catalog(6, 2619), + "filenode tracking is per database", + ); + } + #[test] fn relmap_update_adds_post_rewrite_filenodes() { let mut t = CatalogTracker::new(); diff --git a/src/filter/engine.rs b/src/filter/engine.rs index a7e948a4..1f23627e 100644 --- a/src/filter/engine.rs +++ b/src/filter/engine.rs @@ -15,19 +15,22 @@ use std::sync::{Arc, Mutex}; use walrus::pg::walparser::{RelFileNode, RmId, XLogRecord, XLogRecordBlock}; +use crate::decode::heap_decoder::{XLOG_HEAP_INPLACE, XLOG_HEAP_OPMASK}; use crate::decode::wal_xact::{ XLOG_XACT_ABORT, XLOG_XACT_ABORT_PREPARED, XLOG_XACT_ASSIGNMENT, XLOG_XACT_COMMIT, XLOG_XACT_COMMIT_PREPARED, XLOG_XACT_INVALIDATIONS, XLOG_XACT_OPMASK, XactPayloadError, parse_xact_assignment, parse_xact_invalidations, parse_xact_payload, }; -use crate::filter::catalog_tracker::{CatalogTracker, CatalogTrackerStats}; +use tokio_postgres::Client; + +use crate::filter::catalog_tracker::{CatalogTracker, CatalogTrackerStats, SeedError}; use crate::filter::classify::{Class, classify}; use crate::filter::dirty_tree::{DirtyState, DirtyTree}; use crate::filter::main_data; use crate::filter::manifest::ManifestStats; use crate::record::{AffectedOid, BoundaryInfo, BoundaryKind, Route, rmgr_label}; use crate::schema::FIRST_NORMAL_OBJECT_ID; -use ahash::HashMap; +use ahash::{HashMap, HashSet, HashSetExt}; #[derive(Debug, Default, Clone, Copy)] pub struct FilterStats { @@ -174,6 +177,10 @@ pub struct Filter { /// `XLOG_XACT_INVALIDATIONS` set, plus subxid → top links; drained at /// commit / abort dirty: DirtyTree, + /// Transactions that wrote only statistics + stats_writers: HashSet, + /// First xid known to be fully visible to this run + observed_from_xid: Option, smgr_markers: Arc>, /// Followed database. Routing and catalog-filenode tracking stay /// cluster-wide; this scopes descriptor-capture input only. `None` @@ -188,6 +195,8 @@ impl Filter { tracker: CatalogTracker::new(), stats: FilterStats::default(), dirty: DirtyTree::default(), + stats_writers: HashSet::new(), + observed_from_xid: None, smgr_markers: Arc::new(Mutex::new(SmgrMarkers::default())), target_db_oid: None, } @@ -204,6 +213,30 @@ impl Filter { self.smgr_markers.clone() } + /// Record an xid after which transactions are fully observed + pub fn observe_from_xid(&mut self, xid: u32) { + let earlier = match self.observed_from_xid { + Some(have) if (have.wrapping_sub(xid) as i32) <= 0 => have, + _ => xid, + }; + self.observed_from_xid = Some(earlier); + } + + /// Seed observation state from source snapshot + pub async fn seed_observed_from_source(&mut self, client: &Client) -> Result { + let row = client + .query_one( + "SELECT (pg_snapshot_xmax(pg_current_snapshot())::text::numeric \ + % 4294967296)::bigint", + &[], + ) + .await?; + let next_xid: i64 = row.get(0); + let next_xid = next_xid as u32; + self.observe_from_xid(next_xid); + Ok(next_xid) + } + pub fn decide(&mut self, record: &XLogRecord) -> Route { // Offline callers (segment filter tool) have no LSN and no capture; // a malformed commit payload only degrades boundary metadata there, @@ -261,14 +294,14 @@ impl Filter { // (would hold at unrelated commits). Route is database-blind: // foreign catalog records still replay on shadow let (route, catalog_touch_db) = match class { - Class::Catalog => (Route::ToShadow, catalog_write_db(record)), + Class::Catalog => (Route::ToShadow, self.descriptor_touch_db(record)), // Relmap update (VACUUM FULL mapped catalog) is Special-class, // and carries its database in main_data Class::Special => (Route::ToShadow, obs.catalog_db_oid), Class::User => { if any_block_is_catalog(&self.tracker, &record.blocks) { // tracker has filenodes the bootstrap classify rule misses - (Route::ToShadow, catalog_write_db(record)) + (Route::ToShadow, self.descriptor_touch_db(record)) } else { (Route::ToDecoder, None) } @@ -276,7 +309,8 @@ impl Filter { Class::Empty => match main_data::relation_for_empty(record) { Some(rel) => { if self.tracker.is_catalog(rel.db_node, rel.rel_node) { - (Route::ToShadow, Some(rel.db_node)) + let opaque = self.tracker.is_opaque_catalog(rel.db_node, rel.rel_node); + (Route::ToShadow, (!opaque).then_some(rel.db_node)) } else { (Route::ToDecoder, None) } @@ -294,11 +328,21 @@ impl Filter { .expect("smgr markers poisoned") .insert(rfn, source_lsn); } + // Running-xacts records provide observation points during streaming + if record.header.resource_manager_id == RmId::Standby as u8 + && record.header.info & 0xF0 == main_data::XLOG_RUNNING_XACTS + && let Some(next_xid) = main_data::parse_running_xacts_next_xid(&record.main_data) + { + self.observe_from_xid(next_xid); + } let xid = record.header.xact_id; // Subxid → top link rides the subxact's first record at // wal_level=logical (`XLR_BLOCK_ID_TOPLEVEL_XID`); learn before // touch and admission so both resolve the true root self.dirty.link(xid, record.toplevel_xid); + if xid != 0 && self.is_stats_write(record) { + self.stats_writers.insert(xid); + } // Foreign and shared catalog writes route to shadow and update the // cluster-wide tracker above, but feed no descriptor of the // followed database, so they must not dirty its capture tree @@ -374,6 +418,12 @@ impl Filter { let (merged, members) = self.dirty .drain_tree(header_xid, payload.twophase_xid, &payload.subxacts); + let root = payload.twophase_xid.unwrap_or(header_xid); + let stats_only = self.stats_only_tree(root, &members, merged.is_some()); + // Remove statistics markers for all transaction members + self.forget_stats_writers(&members); + self.forget_stats_writers(&[header_xid, root]); + self.forget_stats_writers(&payload.subxacts); if !is_commit { // Speculative catalog state the tree wrote dies with it. Named // on the record so the drop lands on the pump, ahead of any @@ -421,6 +471,10 @@ impl Filter { capture_all = true; } let dirty_hit = merged.is_some(); + // Statistics-only transactions do not need invalidation recapture + if stats_only && !capture_all { + inval_oids.clear(); + } if !dirty_hit && inval_oids.is_empty() && !capture_all { return Ok(XactEnd::default()); } @@ -494,6 +548,11 @@ impl Filter { return Ok(None); } let root = self.dirty.root(xid); + // Skip invalidation-only work for statistics-only transactions + if !namespace_hit && !flush && self.stats_only_tree(root, &[xid], self.dirty.is_dirty(xid)) + { + return Ok(None); + } let dirty = self.dirty.touch(xid, source_lsn); dirty.unenumerated |= namespace_hit || flush; for oid in &oids { @@ -525,6 +584,50 @@ impl Filter { }))) } + /// Return true for a statistics write in followed database + fn is_stats_write(&self, record: &XLogRecord) -> bool { + let Some(rel) = record.blocks.first().map(|b| b.header.location.rel) else { + return false; + }; + if !self.is_target_db(rel.db_node) { + return false; + } + if self.tracker.is_opaque_catalog(rel.db_node, rel.rel_node) { + return true; + } + record.header.resource_manager_id == RmId::Heap as u8 + && record.header.info & XLOG_HEAP_OPMASK == XLOG_HEAP_INPLACE + && self.tracker.is_catalog(rel.db_node, rel.rel_node) + } + + /// Return true when transaction history is fully observed + fn fully_observed(&self, xid: u32) -> bool { + // Compare transaction IDs with wraparound + self.observed_from_xid + .is_some_and(|from| (xid.wrapping_sub(from) as i32) >= 0) + } + + /// Return true when tree contains only fully observed statistics writes + fn stats_only_tree(&self, root: u32, members: &[u32], dirty_hit: bool) -> bool { + !dirty_hit + && self.fully_observed(root) + && (self.stats_writers.contains(&root) + || members.iter().any(|m| self.stats_writers.contains(m))) + } + + fn forget_stats_writers(&mut self, members: &[u32]) { + for m in members { + self.stats_writers.remove(m); + } + } + + /// Return database for catalog writes that affect descriptors + fn descriptor_touch_db(&self, record: &XLogRecord) -> Option { + let db = catalog_write_db(record)?; + let rel = record.blocks.first()?.header.location.rel; + (!self.tracker.is_opaque_catalog(rel.db_node, rel.rel_node)).then_some(db) + } + /// Per-database relation / catalog scope: the record's database is /// provably the followed one. An unwired filter proves nothing fn is_target_db(&self, db: u32) -> bool { @@ -1069,6 +1172,114 @@ mod tests { assert_eq!(b.oids[0].pg_class_touch, None, "inval-sourced oid"); } + /// Build a running-xacts record + fn running_xacts_rec(next_xid: u32) -> XLogRecord<'static> { + let mut md = vec![0u8; 24]; + md[12..16].copy_from_slice(&next_xid.to_le_bytes()); + let mut r = rec(RmId::Standby, &[]); + r.header.info = main_data::XLOG_RUNNING_XACTS; + r.main_data = std::borrow::Cow::Owned(md); + r + } + + /// Build an in-place pg_class statistics write + fn pg_class_inplace(xid: u32) -> XLogRecord<'static> { + let mut r = rec_with_xid(RmId::Heap, &[(5, 1259)], xid); + r.header.info = XLOG_HEAP_INPLACE; + r + } + + /// Statistics writes and their invalidations do not need a boundary + #[test] + fn analyze_commit_raises_no_boundary() { + let mut f = target_filter(); + f.decide_record(&running_xacts_rec(700), 10, 0xD116) + .unwrap(); + f.decide_record(&rec_with_xid(RmId::Heap, &[(5, 2619)], 746), 100, 0xD116) + .unwrap(); + f.decide_record(&pg_class_inplace(746), 110, 0xD116) + .unwrap(); + let invals = xact_invals_rec(746, &[(-2, 5, 16384), (-2, 5, 16389)]); + assert!( + f.decide_record(&invals, 120, 0xD116) + .unwrap() + .boundary + .is_none(), + "command boundary for a statistics-only transaction", + ); + let commit = xact_end_full( + XLOG_XACT_COMMIT, + 746, + &[], + &[(-2, 5, 16384), (-2, 5, 16389)], + None, + ); + assert!( + f.decide_record(&commit, 130, 0xD116) + .unwrap() + .boundary + .is_none(), + "commit boundary for a statistics-only transaction", + ); + } + + #[test] + fn analyze_before_the_running_xacts_watermark_keeps_its_boundary() { + let mut f = target_filter(); + // Older transaction may have records before stream start + f.decide_record(&running_xacts_rec(900), 10, 0xD116) + .unwrap(); + f.decide_record(&rec_with_xid(RmId::Heap, &[(5, 2619)], 746), 100, 0xD116) + .unwrap(); + f.decide_record(&pg_class_inplace(746), 110, 0xD116) + .unwrap(); + let commit = xact_end_full(XLOG_XACT_COMMIT, 746, &[], &[(-2, 5, 16384)], None); + let b = f + .decide_record(&commit, 130, 0xD116) + .unwrap() + .boundary + .expect("boundary"); + assert_eq!(b.oids[0].oid, 16384); + } + + #[test] + fn ddl_alongside_analyze_keeps_its_boundary() { + let mut f = target_filter(); + f.decide_record(&running_xacts_rec(700), 10, 0xD116) + .unwrap(); + // Keep boundary when DDL and ANALYZE share a transaction + f.decide_record(&rec_with_xid(RmId::Heap, &[(5, 1249)], 746), 100, 0xD116) + .unwrap(); + f.decide_record(&rec_with_xid(RmId::Heap, &[(5, 2619)], 746), 110, 0xD116) + .unwrap(); + let commit = xact_end_full(XLOG_XACT_COMMIT, 746, &[], &[(-2, 5, 16384)], None); + let b = f + .decide_record(&commit, 130, 0xD116) + .unwrap() + .boundary + .expect("boundary"); + assert_eq!(b.tree_first_touch, 100, "pg_attribute write dirtied first"); + assert_eq!(b.oids[0].oid, 16384); + } + + #[test] + fn statistics_writes_alone_never_dirty() { + let mut f = target_filter(); + let stats = rec_with_xid(RmId::Heap, &[(5, 2619)], 7); + assert_eq!( + f.decide_record(&stats, 100, 0xD116).unwrap().route, + Route::ToShadow, + "statistics still replay on shadow", + ); + let commit = xact_end(XLOG_XACT_COMMIT, 7, &[], None); + assert!( + f.decide_record(&commit, 200, 0xD116) + .unwrap() + .boundary + .is_none(), + ); + } + #[test] fn inval_only_commit_is_boundary_defense() { let mut f = target_filter(); diff --git a/src/filter/main_data.rs b/src/filter/main_data.rs index 53353482..be657632 100644 --- a/src/filter/main_data.rs +++ b/src/filter/main_data.rs @@ -97,6 +97,23 @@ pub fn parse_xl_heap_truncate(md: &[u8]) -> Option { }) } +/// Info byte for running-xacts records +pub const XLOG_RUNNING_XACTS: u8 = 0x10; + +/// Offset of next xid in running-xacts payload +const RUNNING_XACTS_NEXT_XID_OFFSET: usize = 12; +/// Minimum running-xacts payload size +const MIN_SIZE_OF_XACT_RUNNING_XACTS: usize = 24; + +/// Read next xid from a running-xacts payload +pub fn parse_running_xacts_next_xid(md: &[u8]) -> Option { + if md.len() < MIN_SIZE_OF_XACT_RUNNING_XACTS { + return None; + } + let at = RUNNING_XACTS_NEXT_XID_OFFSET; + Some(u32::from_le_bytes(md[at..at + 4].try_into().unwrap())) +} + /// `xl_relmap_update` header (PG `src/include/utils/relmapper.h`): /// `Oid dbid; Oid tsid; int32 nbytes; char data[FLEXIBLE_ARRAY_MEMBER]`. /// `dbid == 0` is the shared map (`global/pg_filenode.map`) diff --git a/tests/common/inproc_harness.rs b/tests/common/inproc_harness.rs index fcaab502..5c829aa6 100644 --- a/tests/common/inproc_harness.rs +++ b/tests/common/inproc_harness.rs @@ -703,6 +703,11 @@ async fn build_pipeline_inner( .seed_from_source(sql_client) .await .expect("seed_from_source"); + stream + .filter_mut() + .seed_observed_from_source(sql_client) + .await + .expect("seed observed-from xid"); } let shadow_conninfo = socket_conninfo( diff --git a/tests/vacuum_catalog_churn.rs b/tests/vacuum_catalog_churn.rs new file mode 100644 index 00000000..79e04563 --- /dev/null +++ b/tests/vacuum_catalog_churn.rs @@ -0,0 +1,610 @@ +//! Check that maintenance traffic does not create catalog boundaries +//! +//! Run with `--nocapture` to print boundary details + +#![cfg(target_os = "linux")] + +#[path = "common/inproc_harness.rs"] +mod h; + +use std::collections::{BTreeMap, BTreeSet}; +use std::fs; +use std::io::Write as _; +use std::pin::Pin; +use std::process::Command; +use std::time::{Duration, Instant}; + +use walrus::pg::replication::conn::PgConfig; +use walrus::pg::replication::tls::{SslMode, TlsParams}; +use walshadow::boundary_hold::{BoundaryGateConfig, CatalogBoundaryGate}; +use walshadow::record::{Record, RecordSink, Route, SinkError, WAL_SEG_SIZE, rmgr_label}; +use walshadow::schema::FIRST_NORMAL_OBJECT_ID; +use walshadow::segment_sink::DirSegmentSink; +use walshadow::shadow::{Shadow, ShadowConfig}; +use walshadow::shadow_stream::ShadowStreamSink; +use walshadow::source_feed::{SourceFeed, StandbyStatus}; +use walshadow::wal_stream::WalStream; + +fn pg_available() -> bool { + Command::new("initdb") + .arg("--version") + .output() + .map(|o| o.status.success()) + .unwrap_or(false) +} + +fn make_source(tmp: &tempfile::TempDir, port: u16) -> Shadow { + let mut cfg = ShadowConfig::new(tmp.path().join("source-data"), tmp.path().join("filtered")); + cfg.port = port; + cfg.socket_dir = tmp.path().join("sock"); + cfg.ctl_timeout = Duration::from_secs(60); + fs::create_dir_all(&cfg.filter_out_dir).unwrap(); + fs::create_dir_all(&cfg.socket_dir).unwrap(); + Shadow::new(cfg) +} + +/// Set WAL and vacuum options for controlled test workloads +fn append_source_conf(sh: &Shadow) { + let path = sh.config().data_dir.join("postgresql.conf"); + let mut f = fs::OpenOptions::new().append(true).open(&path).unwrap(); + writeln!(f, "\n# vacuum census overrides").unwrap(); + writeln!(f, "wal_level = logical").unwrap(); + writeln!(f, "max_wal_senders = 4").unwrap(); + writeln!(f, "autovacuum = off").unwrap(); + writeln!(f, "fsync = off").unwrap(); + writeln!(f, "full_page_writes = off").unwrap(); +} + +struct StopOnDrop<'a> { + sh: &'a Shadow, +} + +impl Drop for StopOnDrop<'_> { + fn drop(&mut self) { + let _ = self.sh.stop(); + } +} + +/// Boundary observed by census +#[derive(Debug)] +struct BoundaryRow { + lsn: u64, + kind: String, + /// Catalog relations written by transaction tree + rels: BTreeSet, + capture_all: bool, + /// Relation oids selected for recapture + oids: Vec, +} + +/// Count filter decisions and boundary sources +#[derive(Default)] +struct Census { + records: u64, + to_shadow: u64, + to_decoder: u64, + /// Catalog writes by filenode and WAL operation + catalog_writes: BTreeMap<(u32, String), u64>, + /// Catalog writes grouped by transaction + dirty: BTreeMap>, + boundaries: Vec, + max_next_lsn: u64, +} + +impl Census { + fn observe(&mut self, record: &Record<'_>) { + self.records += 1; + self.max_next_lsn = self.max_next_lsn.max(record.next_lsn); + match record.route { + Route::ToShadow => self.to_shadow += 1, + Route::ToDecoder => self.to_decoder += 1, + } + let xid = record.parsed.header.xact_id; + let op = format!( + "{}/{:#04X}", + rmgr_label(record.parsed.header.resource_manager_id), + record.parsed.header.info & 0xF0, + ); + for blk in &record.parsed.blocks { + let rel = blk.header.location.rel.rel_node; + if rel == 0 || rel >= FIRST_NORMAL_OBJECT_ID { + continue; + } + *self.catalog_writes.entry((rel, op.clone())).or_default() += 1; + if xid != 0 { + self.dirty.entry(xid).or_default().insert(rel); + } + } + if let Some(info) = &record.boundary_info { + let mut rels = self.dirty.remove(&xid).unwrap_or_default(); + for member in &info.members { + if let Some(sub) = self.dirty.remove(member) { + rels.extend(sub); + } + } + self.boundaries.push(BoundaryRow { + lsn: record.source_lsn, + kind: format!("{:?}", info.kind), + rels, + capture_all: info.capture_all, + oids: info.oids.iter().map(|a| a.oid).collect(), + }); + } + } + + fn report(&self, phase: &str, names: &Names) { + let name_of = |rel: u32| names.filenode(rel); + println!( + "\n=== {phase}: {} records ({} to shadow, {} to decoder), {} boundaries", + self.records, + self.to_shadow, + self.to_decoder, + self.boundaries.len(), + ); + let mut by_count: Vec<_> = self.catalog_writes.iter().collect(); + by_count.sort_by_key(|((rel, op), count)| (std::cmp::Reverse(**count), *rel, op.clone())); + for ((rel, op), count) in by_count { + println!(" {count:6} {} {op}", name_of(*rel)); + } + for b in &self.boundaries { + let rels: Vec = b.rels.iter().map(|r| name_of(*r)).collect(); + let oids: Vec = b.oids.iter().map(|o| names.oid(*o)).collect(); + println!( + " boundary {:#X} {} capture_all={} recaptures [{}] via writes to [{}]", + b.lsn, + b.kind, + b.capture_all, + oids.join(", "), + rels.join(", "), + ); + } + } + + /// Catalog relations that dirtied at least one boundary + fn boundary_rels(&self) -> BTreeSet { + self.boundaries + .iter() + .flat_map(|b| b.rels.clone()) + .collect() + } +} + +impl RecordSink for Census { + fn on_record<'a>( + &'a mut self, + record: &'a Record<'a>, + ) -> Pin> + Send + 'a>> { + Box::pin(async move { + self.observe(record); + Ok(()) + }) + } +} + +/// Report how far sink has consumed +trait Drained { + fn max_next_lsn(&self) -> u64; +} + +impl Drained for Census { + fn max_next_lsn(&self) -> u64 { + self.max_next_lsn + } +} + +/// Census with publication holds enabled +struct HoldingCensus { + census: Census, + gate: CatalogBoundaryGate, + holds: u64, + held: Duration, +} + +impl Drained for HoldingCensus { + fn max_next_lsn(&self) -> u64 { + self.census.max_next_lsn + } +} + +impl RecordSink for HoldingCensus { + fn on_record<'a>( + &'a mut self, + record: &'a Record<'a>, + ) -> Pin> + Send + 'a>> { + Box::pin(async move { + self.census.observe(record); + if record.catalog_boundary { + let parked = Instant::now(); + self.gate + .hold(record.source_lsn, record.next_lsn, || true) + .await?; + self.holds += 1; + self.held += parked.elapsed(); + } + Ok(()) + }) + } +} + +/// Resolve relation names from filenodes and oids +struct Names { + by_filenode: BTreeMap, + by_oid: BTreeMap, +} + +impl Names { + fn load(sh: &Shadow) -> Self { + let rows = sh + .psql_one( + "SELECT string_agg(pg_relation_filenode(oid)::text || ' ' || oid::text \ + || ' ' || relname, E'\\n') \ + FROM pg_class WHERE pg_relation_filenode(oid) IS NOT NULL", + ) + .expect("relation map"); + let mut by_filenode = BTreeMap::new(); + let mut by_oid = BTreeMap::new(); + for line in rows.lines() { + let mut parts = line.splitn(3, ' '); + let (Some(node), Some(oid), Some(name)) = (parts.next(), parts.next(), parts.next()) + else { + continue; + }; + by_filenode.insert(node.parse().expect("filenode"), name.to_string()); + by_oid.insert(oid.parse().expect("oid"), name.to_string()); + } + Self { + by_filenode, + by_oid, + } + } + + fn filenode(&self, node: u32) -> String { + self.by_filenode + .get(&node) + .cloned() + .unwrap_or_else(|| format!("filenode {node}")) + } + + fn oid(&self, oid: u32) -> String { + self.by_oid + .get(&oid) + .cloned() + .unwrap_or_else(|| format!("oid {oid}")) + } +} + +fn current_db_oid(sh: &Shadow) -> u32 { + sh.psql_one("SELECT oid::int8 FROM pg_database WHERE datname = current_database()") + .expect("db oid") + .parse() + .expect("integer") +} + +fn wal_insert_lsn(sh: &Shadow) -> u64 { + let s = sh + .psql_one("SELECT pg_current_wal_insert_lsn()") + .expect("lsn"); + walshadow::pg::parse_pg_lsn(&s).expect("parse lsn") +} + +/// Run workload and pump its WAL into sink +async fn pump_phase( + sh: &Shadow, + feed: &mut SourceFeed, + stream: &mut WalStream, + segs: &mut DirSegmentSink, + buf: &mut Vec, + sink: &mut S, + sql: &str, +) -> Duration { + sh.apply_schema_dump(sql).expect("workload"); + sh.psql_one("INSERT INTO marker(at) VALUES (now())") + .expect("marker"); + let target = wal_insert_lsn(sh); + + let started = Instant::now(); + let deadline = started + Duration::from_secs(60); + while sink.max_next_lsn() < target && Instant::now() < deadline { + let next = tokio::time::timeout( + Duration::from_secs(2), + feed.next_chunk(StandbyStatus::collapsed(stream.dispatched_lsn()), buf), + ) + .await; + let chunk = match next { + Ok(Ok(Some(c))) => c, + Ok(Ok(None)) => break, + Ok(Err(e)) => panic!("source feed: {e:#}"), + Err(_) => continue, + }; + stream + .push(chunk.start_lsn, chunk.data, sink, segs) + .await + .expect("push"); + } + let elapsed = started.elapsed(); + assert!( + sink.max_next_lsn() >= target, + "phase drained to {:#X}, target {target:#X}", + sink.max_next_lsn(), + ); + elapsed +} + +/// Run pump phase with a new census +async fn phase( + sh: &Shadow, + feed: &mut SourceFeed, + stream: &mut WalStream, + segs: &mut DirSegmentSink, + buf: &mut Vec, + sql: &str, +) -> Census { + let mut census = Census::default(); + pump_phase(sh, feed, stream, segs, buf, &mut census, sql).await; + census +} + +/// Attach replication feed and filter to source +async fn attach(source: &Shadow, app_name: &str) -> (SourceFeed, WalStream) { + let cfg = source.config(); + let pgcfg = PgConfig { + host: cfg.socket_dir.to_string_lossy().into_owned(), + port: cfg.port, + user: "postgres".into(), + password: None, + database: "postgres".into(), + application_name: app_name.into(), + sslmode: SslMode::Disable, + tls: TlsParams::default(), + }; + let mut feed = SourceFeed::connect(&pgcfg) + .await + .expect("feed connect") + .with_status_interval(Duration::from_millis(500)); + let ident = feed.identify_system().await.expect("IDENTIFY_SYSTEM"); + let aligned = WalStream::align_down(ident.xlogpos, WAL_SEG_SIZE); + let mut stream = WalStream::new(ident.timeline, WAL_SEG_SIZE, aligned).unwrap(); + stream.filter_mut().set_target_db(current_db_oid(source)); + { + let sql_client = feed.sql_client().await.expect("sql client"); + stream + .filter_mut() + .tracker_mut() + .seed_from_source(sql_client) + .await + .expect("seed_from_source"); + stream + .filter_mut() + .seed_observed_from_source(sql_client) + .await + .expect("seed observed-from xid"); + } + feed.start_physical_replication(None, aligned, ident.timeline) + .await + .expect("START_REPLICATION"); + (feed, stream) +} + +/// Verify maintenance workloads do not create boundaries +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn maintenance_traffic_costs_no_catalog_boundary() { + if !pg_available() { + eprintln!("skip: no initdb on PATH"); + return; + } + let tmp = tempfile::tempdir().unwrap(); + let source = make_source(&tmp, h::PG_SOURCE_PORT); + source.initdb().expect("initdb"); + source.write_base_conf().expect("base conf"); + append_source_conf(&source); + source.start().expect("start"); + let _stop = StopOnDrop { sh: &source }; + + source + .apply_schema_dump( + "CREATE TABLE churn (id bigint primary key, payload text);\n\ + CREATE TABLE marker (at timestamptz);\n\ + INSERT INTO churn SELECT g, repeat('x', 200) FROM generate_series(1, 20000) g;\n\ + DELETE FROM churn WHERE id % 3 = 0;\n", + ) + .expect("seed schema"); + + let names = Names::load(&source); + let (mut feed, mut stream) = attach(&source, "vacuum-census").await; + let mut segs = DirSegmentSink::new(tmp.path().join("out")).expect("out dir"); + let mut buf = Vec::with_capacity(64 * 1024); + + // Drain schema setup before measurement + phase( + &source, + &mut feed, + &mut stream, + &mut segs, + &mut buf, + "SELECT 1;\n", + ) + .await; + + let dml = phase( + &source, + &mut feed, + &mut stream, + &mut segs, + &mut buf, + "UPDATE churn SET payload = repeat('y', 200) WHERE id % 7 = 0;\n\ + INSERT INTO churn SELECT g, repeat('z', 200) FROM generate_series(20001, 22000) g;\n", + ) + .await; + dml.report("dml only", &names); + + let analyze = phase( + &source, + &mut feed, + &mut stream, + &mut segs, + &mut buf, + "ANALYZE churn;\n", + ) + .await; + analyze.report("analyze", &names); + + let vacuum = phase( + &source, + &mut feed, + &mut stream, + &mut segs, + &mut buf, + "VACUUM churn;\n", + ) + .await; + vacuum.report("vacuum", &names); + + let vacuum_analyze = phase( + &source, + &mut feed, + &mut stream, + &mut segs, + &mut buf, + "VACUUM (ANALYZE) churn;\n", + ) + .await; + vacuum_analyze.report("vacuum analyze", &names); + + let name_of = |rel: &u32| names.filenode(*rel); + for (phase_name, census) in [ + ("dml only", &dml), + ("analyze", &analyze), + ("vacuum", &vacuum), + ("vacuum analyze", &vacuum_analyze), + ] { + let rels: Vec = census.boundary_rels().iter().map(name_of).collect(); + assert!( + census.boundaries.is_empty(), + "{phase_name}: {} catalog boundaries, dirtied by [{}] — each one stalls the pump", + census.boundaries.len(), + rels.join(", "), + ); + } +} + +/// Verify maintenance workloads do not park a live pump +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn maintenance_traffic_parks_the_pump_for_nothing() { + if !h::pg_available() { + eprintln!("skip: no initdb on PATH"); + return; + } + if !h::pg_basebackup_available() { + eprintln!("skip: no pg_basebackup on PATH"); + return; + } + let tmp = tempfile::tempdir().unwrap(); + let walsender_port = h::reserve_port(); + let (clusters, shadow_state) = h::bootstrap_clusters( + &tmp, + "CREATE TABLE churn (id bigint primary key, payload text);\n\ + CREATE TABLE marker (at timestamptz);\n\ + INSERT INTO churn SELECT g, repeat('x', 200) FROM generate_series(1, 20000) g;\n\ + DELETE FROM churn WHERE id % 3 = 0;\n", + h::PG_SOURCE_PORT, + h::PG_SHADOW_PORT, + walsender_port, + ) + .await; + let source = &clusters.source; + let _stop_source = StopOnDrop { sh: source }; + let _stop_shadow = StopOnDrop { + sh: &clusters.shadow, + }; + // Disable background maintenance during controlled phases + source + .apply_schema_dump("ALTER SYSTEM SET autovacuum = off;\nSELECT pg_reload_conf();\n") + .expect("autovacuum off"); + + let (mut feed, mut stream) = attach(source, "vacuum-hold").await; + stream.set_bytes_sink(Box::new(ShadowStreamSink::new(shadow_state.clone()))); + let mut segs = DirSegmentSink::new(clusters.shadow_filter_dir.clone()).expect("filter dir"); + let mut buf = Vec::with_capacity(64 * 1024); + let mut sink = HoldingCensus { + census: Census::default(), + gate: CatalogBoundaryGate::new(shadow_state, BoundaryGateConfig::default()), + holds: 0, + held: Duration::ZERO, + }; + + let report = |name: &str, sink: &mut HoldingCensus, elapsed: Duration| { + let (records, holds, held) = (sink.census.records, sink.holds, sink.held); + println!( + "{name:>16}: {records:6} records in {elapsed:>10.2?}, \ + {holds} holds costing {held:.2?}", + ); + sink.census = Census::default(); + sink.holds = 0; + sink.held = Duration::ZERO; + (holds, held) + }; + + let elapsed = pump_phase( + source, + &mut feed, + &mut stream, + &mut segs, + &mut buf, + &mut sink, + "SELECT 1;\n", + ) + .await; + report("bootstrap wal", &mut sink, elapsed); + + let elapsed = pump_phase( + source, + &mut feed, + &mut stream, + &mut segs, + &mut buf, + &mut sink, + "UPDATE churn SET payload = repeat('y', 200) WHERE id % 7 = 0;\n\ + INSERT INTO churn SELECT g, repeat('z', 200) FROM generate_series(20001, 22000) g;\n", + ) + .await; + let (dml_holds, _) = report("dml only", &mut sink, elapsed); + + let elapsed = pump_phase( + source, + &mut feed, + &mut stream, + &mut segs, + &mut buf, + &mut sink, + "ANALYZE churn;\nVACUUM (ANALYZE) churn;\nANALYZE;\n", + ) + .await; + let (maintenance_holds, maintenance_held) = report("maintenance", &mut sink, elapsed); + + // Use DDL to confirm boundary holds still work + let elapsed = pump_phase( + source, + &mut feed, + &mut stream, + &mut segs, + &mut buf, + &mut sink, + "ALTER TABLE churn ADD COLUMN c int;\n", + ) + .await; + let (ddl_holds, ddl_held) = report("add column", &mut sink, elapsed); + + assert_eq!(dml_holds, 0, "DML alone must not park the pump"); + assert_eq!( + maintenance_holds, 0, + "VACUUM / ANALYZE parked the pump {maintenance_holds} times for {maintenance_held:.2?}", + ); + assert!( + ddl_holds > 0, + "ADD COLUMN must still park: the gate is what proves the census above", + ); + println!( + "one publication hold costs {:.2?}", + ddl_held / ddl_holds as u32 + ); +} From 5b0451e00b5eb6efcbeb1a2ec4070eddb33b1c56 Mon Sep 17 00:00:00 2001 From: serprex <159546+serprex@users.noreply.github.com> Date: Tue, 4 Aug 2026 00:36:35 +0000 Subject: [PATCH 2/8] Implement wakeups for holds Avoids 20ms stuttering while processing DDL --- plans/shadow.md | 5 ++ plans/source.md | 17 +++++- src/source/boundary_hold.rs | 39 ++++++++++++- src/source/shadow_stream.rs | 106 ++++++++++++++++++++++++++++++++-- tests/vacuum_catalog_churn.rs | 2 +- 5 files changed, 159 insertions(+), 10 deletions(-) diff --git a/plans/shadow.md b/plans/shadow.md index 886912bc..a1bf5be7 100644 --- a/plans/shadow.md +++ b/plans/shadow.md @@ -279,6 +279,11 @@ Aggregate view (`ShadowStreamState::aggregate() → AggregateLsn`) exposes `min_flush_lsn`, `min_apply_lsn`, `active_connections`, `dropped_total` for status loop + metrics +Writes ride a batching tick, with `request_status` waking the listener +for the one path that cannot wait one out — the publication hold, which +needs its queued WAL at shadow before the reply it prods for can mean +anything ([source.md](source.md)) + Backpressure: per-connection send queue caps at `slow_threshold` bytes; overflow drops socket & lets shadow reconnect — completed segments via archive (`restore_command`), in-progress segment via `wire_buf` diff --git a/plans/source.md b/plans/source.md index aebf0437..eb0b0edf 100644 --- a/plans/source.md +++ b/plans/source.md @@ -281,9 +281,20 @@ commits never park; hold cost is DDL-rate Shadow keeps applying while the pump parks: the walsender listener flushes already-queued frames on its own task. Walreceivers report apply progress only when flush advances or on -`wal_receiver_status_interval`, so the gate prods each poll tick with a -reply-requested `'k'` keepalive (`ShadowStreamState::request_status`) -and observes apply at poll cadence (ms). Waiter is result-bearing: +`wal_receiver_status_interval`, so the gate prods with a reply-requested +`'k'` keepalive (`ShadowStreamState::request_status`). + +Neither direction waits out a timer. `request_status` wakes the listener +so the prod — and the queued WAL behind it, which shadow cannot apply +before it receives — goes out now instead of on the listener's next +batching tick; the reply lands in `observe_status`, which wakes the gate +when apply advances. Bulk WAL keeps the batching tick: waking per enqueue +costs more in listener/pump lock traffic than the latency is worth, and +only a hold needs bytes out early. `poll_interval` stays as the backstop +that paces the deadline and worker-liveness checks, so a lost wake +degrades to the old cadence rather than hanging. What remains is shadow's +own replay — sub-millisecond warm, ~10ms for the first catalog record +after an idle period. Waiter is result-bearing: decoder-worker death (`QueueingRecordSink::worker_alive`, channel closed on fatal error or panic; parked root cause preferred over the generic hold error), walreceiver loss past the deadline, and diff --git a/src/source/boundary_hold.rs b/src/source/boundary_hold.rs index 52437e4c..873bc9da 100644 --- a/src/source/boundary_hold.rs +++ b/src/source/boundary_hold.rs @@ -105,6 +105,9 @@ impl CatalogBoundaryGate { worker_alive: impl Fn() -> bool, ) -> Result<(), SinkError> { let start = Instant::now(); + // Subscribed before the first read, so a status landing mid-check + // wakes the wait rather than being missed + let mut applied = self.state.lock().await.applied_rx(); loop { let agg = self.state.lock().await.aggregate(); if agg.min_apply_lsn.is_some_and(|apply| apply >= next_lsn) { @@ -136,7 +139,11 @@ impl CatalogBoundaryGate { )); } self.state.lock().await.request_status(); - tokio::time::sleep(self.config.poll_interval).await; + // Shadow's reply wakes this; `poll_interval` is the backstop + // that keeps a lost wake — or a walreceiver that never answers + // — degrading to the old cadence instead of hanging, and paces + // the `worker_alive` / deadline checks above + let _ = tokio::time::timeout(self.config.poll_interval, applied.changed()).await; } } @@ -321,6 +328,36 @@ mod tests { assert_eq!(gate.stats.failures.load(Ordering::Relaxed), 0); } + /// Release rides the status wake, not the poll: with the backstop set + /// far past the status, only the wake can release inside the timeout + #[tokio::test] + async fn hold_releases_on_the_status_wake_not_the_poll() { + let s = state(); + let id = s.lock().await.register_connection(0x1000); + let gate = CatalogBoundaryGate::new( + s.clone(), + BoundaryGateConfig { + hold_timeout: Duration::from_secs(5), + poll_interval: Duration::from_secs(4), + }, + ); + let waiter = tokio::spawn({ + let s = s.clone(); + async move { + tokio::time::sleep(Duration::from_millis(20)).await; + s.lock().await.observe_status(id, 0x2000, 0x2000, 0x2000); + } + }); + let started = Instant::now(); + gate.hold(0x1F00, 0x2000, || true).await.expect("released"); + waiter.await.unwrap(); + assert!( + started.elapsed() < Duration::from_secs(1), + "waited {:?}, so the poll released it, not the wake", + started.elapsed(), + ); + } + #[tokio::test] async fn hold_prods_walreceiver_with_reply_requested_keepalive() { let s = state(); diff --git a/src/source/shadow_stream.rs b/src/source/shadow_stream.rs index 9b1d7044..2a7db383 100644 --- a/src/source/shadow_stream.rs +++ b/src/source/shadow_stream.rs @@ -25,7 +25,7 @@ use std::time::Duration; use smallvec::SmallVec; use thiserror::Error; use tokio::net::{TcpListener, TcpStream, UnixListener, UnixStream}; -use tokio::sync::Mutex; +use tokio::sync::{Mutex, watch}; use walrus::pg::replication::server::{self, ServerError, WalSenderConn, decode_standby_status}; use walrus::pg::replication::stream::{encode_keepalive_frame_into, encode_wal_data_frame_into}; @@ -112,6 +112,13 @@ pub struct ShadowStreamState { /// gap and strands at segment boundaries). Reset per segment. wire_buf: Vec, wire_buf_start: u64, + /// Wakes the listener to write now rather than on its batching tick; + /// bumped by `request_status`, the one caller whose bytes a hold is + /// waiting behind + queued: watch::Sender, + /// Bumped when a client's apply LSN advances: wakes a publication hold + /// on the standby status that releases it + applied: watch::Sender, } impl ShadowStreamState { @@ -135,9 +142,30 @@ impl ShadowStreamState { dropped_total: 0, wire_buf: Vec::new(), wire_buf_start: current_lsn, + queued: watch::Sender::new(0), + applied: watch::Sender::new(0), } } + /// Wakes the listener out of its batching tick, for the paths that + /// cannot wait one out ([`request_status`](Self::request_status)). + /// `watch` latches, so a wake landing while the listener is mid-write + /// still arms its next wait — and a drain takes the whole queue, so one + /// wake covers every byte enqueued before it + pub fn queued_rx(&self) -> watch::Receiver { + self.queued.subscribe() + } + + /// Wakes when any client's apply LSN advances; the waiter re-reads + /// [`aggregate`](Self::aggregate), which stays the release authority + pub fn applied_rx(&self) -> watch::Receiver { + self.applied.subscribe() + } + + fn wake_listener(&self) { + self.queued.send_modify(|n| *n += 1); + } + pub fn aggregate(&self) -> AggregateLsn { let (active, min_flush, min_apply) = self .connections @@ -256,7 +284,11 @@ impl ShadowStreamState { let _ = write_lsn; if let Some(c) = self.connections.get_mut(&id) { c.flush_lsn = c.flush_lsn.max(flush_lsn); + let advanced = apply_lsn > c.apply_lsn; c.apply_lsn = c.apply_lsn.max(apply_lsn); + if advanced { + self.applied.send_modify(|n| *n += 1); + } } } @@ -336,10 +368,17 @@ impl ShadowStreamState { } /// Enqueue a reply-requested `'k'` keepalive on every active - /// connection. Shadow's walreceiver answers immediately with fresh - /// flush/apply LSNs — non-forced replies otherwise fire only when the - /// flush position advances or `wal_receiver_status_interval` elapses, - /// so a publication hold waiting on apply progress prods through this. + /// connection and wake the listener to write it now. Shadow's + /// walreceiver answers immediately with fresh flush/apply LSNs — + /// non-forced replies otherwise fire only when the flush position + /// advances or `wal_receiver_status_interval` elapses, so a publication + /// hold waiting on apply progress prods through this. + /// + /// The wake carries the queue's WAL bytes out with the keepalive: a + /// hold waits on replay of records shadow cannot apply before it + /// receives them, and the listener's tick is a batching timer sized for + /// bulk streaming, not for this. Bulk traffic keeps that batching — + /// only the prod jumps the queue pub fn request_status(&mut self) { let server_wal_end = self.server_wal_end; let ids: Vec = self @@ -353,6 +392,7 @@ impl ShadowStreamState { encode_keepalive_frame_into(out, server_wal_end, true); }); } + self.wake_listener(); } } @@ -571,8 +611,25 @@ where let mut last_write = tokio::time::Instant::now(); let mut ticker = tokio::time::interval(flush_interval); ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); + // Writes ride the enqueue wake; the ticker is the idle-keepalive timer + // and the backstop for a wake the queue raced. Batching survives: a + // wake drains everything queued, and bytes enqueued during a write + // land in the next drain + let mut queued = state.lock().await.queued_rx(); + // Registration backfilled this connection before the subscribe, so the + // first wait must not swallow bytes already queued + queued.mark_changed(); loop { tokio::select! { + _ = queued.changed() => { + let pending = state.lock().await.drain_send_queue(id); + if let Some(bytes) = pending + && !bytes.is_empty() + { + conn.write_framed(&bytes).await?; + last_write = tokio::time::Instant::now(); + } + } _ = ticker.tick() => { let pending = { let mut s = state.lock().await; @@ -626,6 +683,45 @@ mod tests { ShadowStreamState::new(1, "12345".into(), 0x1000, 1024 * 1024) } + #[test] + fn status_request_wakes_the_listener_with_the_queued_wal() { + let mut s = fresh_state(); + let id = s.register_connection(0x1000); + let rx = s.queued_rx(); + s.enqueue(id, vec![b'd', 0, 0, 0, 4]); + assert!( + !rx.has_changed().expect("sender alive"), + "bulk WAL keeps the listener's batching tick", + ); + s.request_status(); + assert!(rx.has_changed().expect("sender alive")); + let queued = s.drain_send_queue(id).expect("queue"); + assert!( + queued.len() > 5, + "the wake carries the queued WAL out with the keepalive", + ); + } + + #[test] + fn only_apply_progress_wakes_a_hold() { + let mut s = fresh_state(); + let id = s.register_connection(0x1000); + let mut rx = s.applied_rx(); + s.observe_status(id, 0x2000, 0x2000, 0x1000); + assert!( + !rx.has_changed().expect("sender alive"), + "flush progress alone releases nothing", + ); + s.observe_status(id, 0x2000, 0x2000, 0x1800); + assert!(rx.has_changed().expect("sender alive")); + rx.mark_unchanged(); + s.observe_status(id, 0x2000, 0x2000, 0x1800); + assert!( + !rx.has_changed().expect("sender alive"), + "a repeated status is not progress", + ); + } + #[test] fn aggregate_lsn_with_no_connections_is_default() { let s = fresh_state(); diff --git a/tests/vacuum_catalog_churn.rs b/tests/vacuum_catalog_churn.rs index 79e04563..32fd838d 100644 --- a/tests/vacuum_catalog_churn.rs +++ b/tests/vacuum_catalog_churn.rs @@ -589,7 +589,7 @@ async fn maintenance_traffic_parks_the_pump_for_nothing() { &mut segs, &mut buf, &mut sink, - "ALTER TABLE churn ADD COLUMN c int;\n", + "ALTER TABLE churn ADD COLUMN c int;\nALTER TABLE churn ADD COLUMN d int;\nALTER TABLE churn ADD COLUMN e int;\n", ) .await; let (ddl_holds, ddl_held) = report("add column", &mut sink, elapsed); From 68899c8775ca4957c58f031cbc034025c0df8d1f Mon Sep 17 00:00:00 2001 From: serprex <159546+serprex@users.noreply.github.com> Date: Tue, 4 Aug 2026 17:22:17 +0000 Subject: [PATCH 3/8] fix bench/ to work with podman & support longer runs --- bench/ec2/.gitignore | 3 + bench/ec2/README.md | 28 +- bench/ec2/aws-env.sh | 79 ++++- bench/ec2/ec2-bench/deploy.sh | 9 +- bench/ec2/ec2-walshadow/deploy.sh | 30 +- bench/ec2/lib.sh | 11 + bench/ec2/terraform/main.tf | 3 +- bench/ec2/terraform/variables.tf | 5 + bench/run_bench_suite.sh | 13 +- plans/future/ddl_fuzz.md | 486 ++++++++++++++++++++++++++++++ 10 files changed, 640 insertions(+), 27 deletions(-) create mode 100644 plans/future/ddl_fuzz.md diff --git a/bench/ec2/.gitignore b/bench/ec2/.gitignore index 07c83026..907e9a35 100644 --- a/bench/ec2/.gitignore +++ b/bench/ec2/.gitignore @@ -2,6 +2,9 @@ *.pem state.env +# Local AWS profile settings +aws.local.env + # copied-back perf/eBPF profiles (large) profiles/ diff --git a/bench/ec2/README.md b/bench/ec2/README.md index 0ff73ebb..a7432384 100644 --- a/bench/ec2/README.md +++ b/bench/ec2/README.md @@ -24,6 +24,22 @@ shell: node folders keep `cloud-init.yaml` and, where needed, `deploy.sh` / (creds) and `lib.sh` (ssh/state.env). `*.pem`, `state.env` and terraform state are gitignored. +Record account and profile in `aws.local.env`, which stays gitignored and is +required by `aws-env.sh`. Account check aborts when credentials resolve to a +different account, then Terraform checks same account through +`allowed_account_ids`: + +```bash +echo 'BENCH_AWS_ACCOUNT=' > aws.local.env +echo 'BENCH_AWS_PROFILE=

' >> aws.local.env +aws sso login --profile=

# renew expired session +``` + +After the account check, `aws-env.sh` resolves that profile to session keys +(`aws configure export-credentials`) and exports them, so terraform runs on +exactly the creds the check validated — its Go SDK rejects a stale SSO token +that the CLI would still serve from cache. + ## stack.sh — the main interface ```bash @@ -50,12 +66,18 @@ the source table at startup — so a second setup's run would disturb the first. ## Per-setup ### walshadow -Build the image first (the deploy ships a locally-built image rather than building on the box): +Build the image first (the deploy ships a locally-built image rather than building on the box). +`PG_MAJOR` must match `ec2-source-pg` (`postgres:17`) — the shadow data dir comes +from a BASE_BACKUP of the source, so PG 18 binaries cannot open it; `deploy.sh` +compares the two and refuses to start on a mismatch: ```bash -docker build -f docker/Dockerfile -t walshadow:local . # from repo root +docker build -f docker/Dockerfile --build-arg PG_MAJOR=17 -t walshadow:local . # run from repository root ./stack.sh up walshadow # source-pg + clickhouse + daemon ./stack.sh down # daemon only (base kept) ``` +Building with podman works: it tags locally-built images `localhost/walshadow:local`, +and `docker load` on the node keeps that prefix, so the deploys resolve whichever +tag actually landed (`remote_image_tag` in `lib.sh`) instead of assuming `$IMAGE`. `deploy.sh` ships `walshadow:local` (`docker save | ssh | docker load`), writes `ch-config.toml` (ClickHouse private IP, `flush_timeout_ms`), and runs the daemon. @@ -87,6 +109,8 @@ a read-only hot standby that streams WAL. Re-running takes a fresh base backup. ../run_bench_suite.sh walshadow-run # DEST defaults to clickhouse # physical standby: DEST=postgres ../run_bench_suite.sh pg-run # reads ec2-pg-standby +# Run sustained and interleaved loads for five minutes; long run already uses 10 30-second rounds +RUN_SECS=300 ../run_bench_suite.sh walshadow-5min ``` ## Notes diff --git a/bench/ec2/aws-env.sh b/bench/ec2/aws-env.sh index 0758f770..da250b0c 100644 --- a/bench/ec2/aws-env.sh +++ b/bench/ec2/aws-env.sh @@ -1,22 +1,81 @@ #!/usr/bin/env bash -# Source this before running aws commands: source ../aws-env.sh +# Load before running AWS commands: source ../aws-env.sh +# +# Refuse to run against unexpected account +# Set account and profile in ignored aws.local.env +# +# Example: +# BENCH_AWS_ACCOUNT= +# BENCH_AWS_PROFILE=

+# Run aws sso login --profile=

when session expires # # Credential resolution, in order: +# - AWS_PROFILE, then BENCH_AWS_PROFILE # - ~/.aws/credentials in shell-export form (export AWS_ACCESS_KEY_ID=...): -# source it as env vars and point CLI file paths at /dev/null so the CLI -# uses env vars instead of trying (and failing) to parse it as INI -# - otherwise normal AWS resolution via AWS_PROFILE (eg SSO: -# aws sso login --profile=

&& export AWS_PROFILE=

) -if [ -f ~/.aws/credentials ]; then +# load environment variables and hide non-INI file from CLI +_bench_ec2_dir="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" +if [ ! -f "$_bench_ec2_dir/aws.local.env" ]; then + echo "missing $_bench_ec2_dir/aws.local.env" >&2 + echo " set BENCH_AWS_ACCOUNT and BENCH_AWS_PROFILE in that file" >&2 + unset _bench_ec2_dir + return 1 +fi +# shellcheck disable=SC1091 +source "$_bench_ec2_dir/aws.local.env" +unset _bench_ec2_dir + +: "${BENCH_AWS_ACCOUNT:?set BENCH_AWS_ACCOUNT in bench/ec2/aws.local.env}" +export BENCH_AWS_ACCOUNT +# Pass expected account to Terraform +export TF_VAR_account_id="$BENCH_AWS_ACCOUNT" + +# Detect INI-formatted credentials +bench_aws_ini_credentials() { grep -qE '^[[:space:]]*\[' ~/.aws/credentials 2>/dev/null; } + +: "${AWS_PROFILE:=${BENCH_AWS_PROFILE:-}}" +if [ -n "$AWS_PROFILE" ]; then + export AWS_PROFILE + bench_aws_ini_credentials || export AWS_SHARED_CREDENTIALS_FILE=/dev/null +elif [ -f ~/.aws/credentials ] && ! bench_aws_ini_credentials; then set -a # shellcheck disable=SC1090 source ~/.aws/credentials set +a - unset AWS_PROFILE export AWS_SHARED_CREDENTIALS_FILE=/dev/null export AWS_CONFIG_FILE=/dev/null - # config file is disabled above, so a region must come from env - export AWS_DEFAULT_REGION="${AWS_DEFAULT_REGION:-ap-south-1}" else - : "${AWS_PROFILE:?no ~/.aws/credentials; export AWS_PROFILE (after aws sso login --profile=

)}" + echo "no profile for AWS account $BENCH_AWS_ACCOUNT" >&2 + echo " echo BENCH_AWS_PROFILE=

>> bench/ec2/aws.local.env && aws sso login --profile=

" >&2 + return 1 fi +# Keep region available when config file is disabled +export AWS_DEFAULT_REGION="${AWS_DEFAULT_REGION:-ap-south-1}" + +# Stop before changing an unexpected account +bench_aws_account_check() { + local got hint="${AWS_PROFILE:+ --profile=$AWS_PROFILE}" + got="$(aws sts get-caller-identity --query Account --output text 2>&1)" || { + echo "aws sts get-caller-identity failed: $got" >&2 + echo " expired SSO session? aws sso login$hint" >&2 + return 1 + } + [ "$got" = "$BENCH_AWS_ACCOUNT" ] || { + echo "wrong AWS account: creds${AWS_PROFILE:+ (profile $AWS_PROFILE)} are in $got, bench infra is in $BENCH_AWS_ACCOUNT" >&2 + echo " set BENCH_AWS_PROFILE in bench/ec2/aws.local.env to a profile for $BENCH_AWS_ACCOUNT," >&2 + echo " or set BENCH_AWS_ACCOUNT to work in $got" >&2 + return 1 + } + export AWS_ACCOUNT_ID="$got" +} +bench_aws_account_check + +# Pass checked credentials to Terraform +bench_aws_export_creds() { + local env_out + env_out="$(aws configure export-credentials --format env-no-export 2>/dev/null)" || return 0 + set -a + eval "$env_out" + set +a + unset AWS_PROFILE +} +bench_aws_export_creds diff --git a/bench/ec2/ec2-bench/deploy.sh b/bench/ec2/ec2-bench/deploy.sh index d0fc3f9f..b4d242f1 100755 --- a/bench/ec2/ec2-bench/deploy.sh +++ b/bench/ec2/ec2-bench/deploy.sh @@ -22,11 +22,14 @@ docker build -f "$REPO_ROOT/docker/Dockerfile.bench" -t "$IMAGE" "$REPO_ROOT" wait_cloud_init -if [ "${FORCE:-0}" != "1" ] && "${SSH[@]}" "sudo docker image inspect $IMAGE >/dev/null 2>&1"; then - echo "image $IMAGE already on host (FORCE=1 to resend)" +REMOTE_IMAGE="$(remote_image_tag "$IMAGE")" +if [ "${FORCE:-0}" != "1" ] && [ -n "$REMOTE_IMAGE" ]; then + echo "image $REMOTE_IMAGE already on host (FORCE=1 to resend)" else echo "shipping $IMAGE (docker save | ssh | docker load)…" docker save "$IMAGE" | gzip | "${SSH[@]}" 'gunzip | sudo docker load' + REMOTE_IMAGE="$(remote_image_tag "$IMAGE")" + [ -n "$REMOTE_IMAGE" ] || { echo "$IMAGE missing on host after load" >&2; exit 1; } fi # Ship the sibling state.env files so --network private can resolve endpoints. @@ -47,7 +50,7 @@ done echo "installing walshadow-ec2-bench wrapper…" "${SSH[@]}" "cat | sudo tee /usr/local/bin/walshadow-ec2-bench >/dev/null && sudo chmod +x /usr/local/bin/walshadow-ec2-bench" <). +# Require stack.sh setup and local `walshadow:local` image, +# built for source PostgreSQL major version: +# docker build -f docker/Dockerfile --build-arg PG_MAJOR=17 -t walshadow:local +# Reject incompatible image versions below # # Endpoint IPs are read from the sibling state.env files; override with # SOURCE_PRIVATE_IP=... CH_HOST=... ./deploy.sh @@ -58,11 +59,28 @@ for i in $(seq 1 30); do done # Ship the image unless it's already present on the host (use FORCE=1 to resend). -if [ "${FORCE:-0}" != "1" ] && "${SSH[@]}" "sudo docker image inspect $IMAGE >/dev/null 2>&1"; then - echo "image $IMAGE already on host (FORCE=1 to resend)" +REMOTE_IMAGE="$(remote_image_tag "$IMAGE")" +if [ "${FORCE:-0}" != "1" ] && [ -n "$REMOTE_IMAGE" ]; then + echo "image $REMOTE_IMAGE already on host (FORCE=1 to resend)" else echo "shipping $IMAGE (docker save | ssh | docker load)..." docker save "$IMAGE" | gzip | "${SSH[@]}" 'gunzip | sudo docker load' + REMOTE_IMAGE="$(remote_image_tag "$IMAGE")" + [ -n "$REMOTE_IMAGE" ] || { echo "$IMAGE missing on host after load" >&2; exit 1; } +fi + +# Backup data requires matching PostgreSQL major versions +IMG_MAJOR="$("${SSH[@]}" "sudo docker run --rm --entrypoint postgres $REMOTE_IMAGE -V" | grep -oE '[0-9]+' | head -1)" +SRC_VERSION_NUM="$("${SSH[@]}" "sudo docker run --rm --entrypoint psql $REMOTE_IMAGE -h '$SRC_PRIV' -U postgres -tAc 'SHOW server_version_num'" | tr -dc '0-9')" +if [ -z "$IMG_MAJOR" ] || [ -z "$SRC_VERSION_NUM" ]; then + echo "could not read PG majors (image $REMOTE_IMAGE, source $SRC_PRIV)" >&2 + exit 1 +fi +SRC_MAJOR=$((SRC_VERSION_NUM / 10000)) +if [ "$IMG_MAJOR" != "$SRC_MAJOR" ]; then + echo "PG major mismatch: image is PG $IMG_MAJOR, source is PG $SRC_MAJOR" >&2 + echo " rebuild: docker build -f docker/Dockerfile --build-arg PG_MAJOR=$SRC_MAJOR -t $IMAGE " >&2 + exit 1 fi # ch-config.toml: the repo config with the CH host swapped to the private IP. @@ -107,7 +125,7 @@ EOF -v /opt/walshadow/ch-config.toml:/etc/walshadow/ch-config.toml:ro \ -v walshadow-data:/var/lib/walshadow \ -p 9484:9484 \ - $IMAGE --trace-sample-ratio '$TRACE_SAMPLE_RATIO' >/dev/null && echo started" + $REMOTE_IMAGE --trace-sample-ratio '$TRACE_SAMPLE_RATIO' >/dev/null && echo started" # Grafana + Prometheus: only uploaded/recreated when FORCE_METRICS=1. if [ "${FORCE_METRICS:-0}" = "1" ]; then diff --git a/bench/ec2/lib.sh b/bench/ec2/lib.sh index 98f09563..b1b5f1d6 100644 --- a/bench/ec2/lib.sh +++ b/bench/ec2/lib.sh @@ -15,6 +15,17 @@ node_ssh_setup() { SCP=(scp -i "$PEM" -o StrictHostKeyChecking=accept-new) } +# Print matching remote image tag, including Podman's localhost prefix +remote_image_tag() { + local ref + for ref in "$1" "localhost/$1"; do + if "${SSH[@]}" "sudo docker image inspect '$ref' >/dev/null 2>&1"; then + echo "$ref" + return + fi + done +} + # Block until cloud-init has finished on the node (SSH must be set up). wait_cloud_init() { echo "waiting for SSH + cloud-init…" diff --git a/bench/ec2/terraform/main.tf b/bench/ec2/terraform/main.tf index a8ff9a88..a06d6511 100644 --- a/bench/ec2/terraform/main.tf +++ b/bench/ec2/terraform/main.tf @@ -23,7 +23,8 @@ terraform { } provider "aws" { - region = var.region + region = var.region + allowed_account_ids = [var.account_id] default_tags { tags = { diff --git a/bench/ec2/terraform/variables.tf b/bench/ec2/terraform/variables.tf index e532aedd..03703a7a 100644 --- a/bench/ec2/terraform/variables.tf +++ b/bench/ec2/terraform/variables.tf @@ -3,6 +3,11 @@ variable "region" { default = "ap-south-1" } +# Restrict provider to expected account +variable "account_id" { + type = string +} + # "none" keeps just the base up variable "streamer" { type = string diff --git a/bench/run_bench_suite.sh b/bench/run_bench_suite.sh index afe6e32f..e05a92b8 100755 --- a/bench/run_bench_suite.sh +++ b/bench/run_bench_suite.sh @@ -12,8 +12,8 @@ # Pick the destination with DEST: # DEST=clickhouse (default) — walshadow / peerdb pipelines (reads ec2-clickhouse) # DEST=postgres — PG→PG physical standby (reads ec2-pg-standby) -# Override any bench's flags via the *_ARGS env vars below, the target with -# NETWORK=private, or skip the rebuild with SKIP_BUILD=1. +# Override flags with *_ARGS, load duration with RUN_SECS, target with NETWORK, +# or skip rebuild with SKIP_BUILD=1 set -euo pipefail HERE="$(cd "$(dirname "$0")" && pwd)" @@ -34,10 +34,13 @@ OUT="$RESULTS_DIR/$NAME" [ -e "$OUT" ] && { echo "error: $OUT already exists — choose a different name" >&2; exit 1; } # Per-bench flags — override via env, e.g. SUSTAINED_ARGS="--bench sustained --rate 1000". -# Durations are kept modest for a quick pass; bump --xact-secs / --duration-secs for longer runs. +# Set RUN_SECS to override sustained and interleaved load duration +# Leave unset for short runs; interleaved-long always runs 10 rounds of 30 seconds +SUSTAINED_SECS="${RUN_SECS:-20}" +INTERLEAVED_SECS="${RUN_SECS:-30}" SINGLE_ARGS="${SINGLE_ARGS:---bench single-row --iterations 100 --warmup 10}" -SUSTAINED_ARGS="${SUSTAINED_ARGS:---bench sustained --rate 30000 --duration-secs 20 --concurrency 90}" -INTERLEAVED_ARGS="${INTERLEAVED_ARGS:---bench interleaved --xact-threads 90 --rounds 1 --xact-secs 30}" +SUSTAINED_ARGS="${SUSTAINED_ARGS:---bench sustained --rate 30000 --duration-secs $SUSTAINED_SECS --concurrency 90}" +INTERLEAVED_ARGS="${INTERLEAVED_ARGS:---bench interleaved --xact-threads 90 --rounds 1 --xact-secs $INTERLEAVED_SECS}" LONG_THROUGHPUT="${LONG_THROUGHPUT:---bench interleaved --xact-threads 1 --rounds 10 --xact-secs 30}" if [ "${SKIP_BUILD:-0}" != "1" ]; then diff --git a/plans/future/ddl_fuzz.md b/plans/future/ddl_fuzz.md new file mode 100644 index 00000000..d0738d26 --- /dev/null +++ b/plans/future/ddl_fuzz.md @@ -0,0 +1,486 @@ +# DDL/DML semantic fuzzing + +Build confidence that legal PostgreSQL schema and row transitions either +converge in ClickHouse or stop with an explicit, pre-side-effect rejection. +Generate typed transaction programs, run them through real PostgreSQL WAL and +walshadow, then compare source and ClickHouse state. Complement, do not replace, +byte-oriented parser fuzzing in [FUZZ.md](FUZZ.md) + +Primary bug class is semantic success: PostgreSQL commits, walshadow reaches an +ack, but ClickHouse holds wrong schema or rows. Panic freedom and parser safety +cannot detect it + +## Scope and contract + +Exercise permanent tables first. Expand across unlogged tables, partitions, +tablespaces, prepared transactions, restarts, and concurrent sessions after +single-table oracle stabilizes + +Classify every generated program before execution: + +- `Converge` — source and ClickHouse logical row sets plus supported schema + projection must match after emitter ack +- `RejectBeforeEffects` — unsupported transition must stop before any part of + source transaction reaches ClickHouse +- `Policy` — configured behavior intentionally differs, for example DROP under + `retain`; compare against policy-specific model instead of source +- `KnownGap` — minimized reproducer kept runnable and reported, never counted as + successful confidence + +Do not accept WARN plus continued ingestion as rejection. Do not accept an +unmapped-row counter as safe discard for a `Converge` case. Treat ack without +state equivalence as failure + +Compare final committed state. ClickHouse cannot expose a PostgreSQL transaction +atomically while walshadow executes separate INSERT / ALTER / TRUNCATE queries; +intermediate reader visibility stays outside this plan. Restart campaigns still +check convergence from every generated execution cut + +## Known transition corpus + +Land deterministic seeds before random campaigns. This separates known product +limits from new interactions and gives shrinker stable endpoints + +| transition | expected current outcome | contract decision needed | +|---|---|---| +| `ALTER TABLE s.t RENAME TO u; INSERT INTO s.u ...` | descriptor name changes without `SchemaEvent`; mapping remains under `s.t`, trailing row routes unmapped | move source mapping key and decide destination rename policy, or reject | +| `ALTER TABLE s.t SET SCHEMA u; UPDATE u.t ...` | same relation-name gap, namespace mapping may also change destination database | migrate route and destination, or reject | +| `ALTER SCHEMA s RENAME TO u` followed by DML | capture-all refreshes descriptors but emits no relation events | emit relation moves, or reject affected mappings | +| `ALTER COLUMN v TYPE int8` from `int4`, then insert above `i32::MAX` | CH type and mapping remain `Int32`; apply only warns for `type_changes` | migrate type before rows, or reject whole xact | +| `ALTER COLUMN v TYPE varchar(40)` from `varchar(10)` | both shapes map to CH `String`; data usually converges despite warning | classify representation-preserving transition explicitly | +| `ALTER COLUMN v TYPE text` from `varchar` with post-DDL DML | pending timeline can decode row, CH representation stays `String` | preserve as supported same-target transition | +| `ALTER COLUMN v DROP NOT NULL; INSERT (..., NULL)` | CH column stays non-nullable; emitter writes type default for NULL | alter nullability, or reject before row apply | +| `ALTER COLUMN v SET NOT NULL` | CH column stays nullable | decide whether weaker CH schema is supported projection | +| switch replica identity from PK `id` to unique `email`, then DELETE | CH `ORDER BY id` stays fixed; old tuple may carry only `email`, tombstone defaults `id` | rebuild destination key, retain immutable target key, or reject | +| drop PK constraint, drop its column | CH still uses column in `ORDER BY`; CH DROP COLUMN fails after any earlier xact effects | preflight whole plan against target key | +| add PK to table created keyless | CH retains `ORDER BY (_lsn)`; updates do not collapse by new PK | rebuild key or keep case unsupported | +| `DROP TABLE; CREATE TABLE` with same name under each drop strategy | `retain` / `warn` preserve old rows and `CREATE IF NOT EXISTS` no-ops; `drop` should round-trip | encode policy-specific lifecycle oracle | +| `CREATE UNLOGGED TABLE; INSERT ...` | catalog may create CH table; user DML has no durable WAL | reject mapping or mark no-row policy explicitly | +| `ALTER TABLE ... SET UNLOGGED`, mutate, `SET LOGGED` | unlogged interval disappears; stale CH rows can survive conversion | reject persistence transition | +| attach populated partition, then write through parent | heap WAL names leaf; pinned parent target receives no fan-in and attach emits no row backfill | define partition routing/backfill semantics | +| more than pending-capture boundary cap around physical in-place drift | timeline degrades; affected rows fail ambiguity fence | preserve fail-closed behavior, assert no side effects | +| `PREPARE TRANSACTION`, restart, `COMMIT PREPARED` | process-local xact state is lost | expected `KnownGap` until [two_phase_commit.md](two_phase_commit.md) lands | +| create or move into non-default tablespace, then DML | shadow path materialization can fail; bootstrap skips source rows | expected `KnownGap` until [TABLESPACES.md](TABLESPACES.md) lands | + +Also seed supported controls: + +- CREATE + INSERT / multi-VALUES / COPY in one transaction +- ADD nullable column between row writes +- ADD column with fast default, including toasted and Tier 3 defaults +- RENAME COLUMN between UPDATEs +- DROP COLUMN with DML before and after, excluding CH sorting key +- TRUNCATE between inserts, including toasted values +- CREATE + ALTER + COPY with multiple pending descriptor slots +- top-level abort, savepoint rollback, savepoint release, committed subxact +- benign typmod / storage drift with post-DDL DML +- rewrite via `VACUUM FULL`, `CLUSTER`, and representation-preserving ALTER +- same programs across daemon restart at each execution barrier + +## Program IR + +Generate operations from typed state, never random SQL bytes. Stable identities +must survive names and object generations changing + +```rust +struct Program { + setup: Setup, + transactions: Vec, +} + +struct Transaction { + actions: Vec, + finish: Finish, +} + +enum Finish { + Commit, + Rollback, + Prepare { gid: Gid, resolve: PreparedResolution }, +} + +enum Action { + Insert { rel: RelId, row: Row }, + InsertMany { rel: RelId, rows: Vec }, + Copy { rel: RelId, rows: Vec }, + Update { rel: RelId, key: Key, changes: Vec }, + Delete { rel: RelId, key: Key }, + + CreateTable { rel: RelId, spec: TableSpec }, + DropTable { rel: RelId }, + RenameTable { rel: RelId, name: Name }, + MoveSchema { rel: RelId, schema: SchemaId }, + SetPersistence { rel: RelId, persistence: Persistence }, + Truncate { rels: Vec, cascade: bool }, + + AddColumn { rel: RelId, column: ColumnSpec }, + DropColumn { rel: RelId, column: ColumnId }, + RenameColumn { rel: RelId, column: ColumnId, name: Name }, + AlterType { rel: RelId, column: ColumnId, ty: PgType }, + SetNullability { rel: RelId, column: ColumnId, nullable: bool }, + SetDefault { rel: RelId, column: ColumnId, value: Option }, + SetStorage { rel: RelId, column: ColumnId, storage: Storage }, + + AddUniqueIndex { rel: RelId, index: IndexId, columns: Vec }, + DropIndex { index: IndexId }, + SetReplicaIdentity { rel: RelId, identity: Identity }, + AddPrimaryKey { rel: RelId, columns: Vec }, + DropPrimaryKey { rel: RelId }, + + Savepoint { id: SavepointId }, + RollbackTo { id: SavepointId }, + Release { id: SavepointId }, +} +``` + +Model state per relation: + +- stable `RelId`, current PostgreSQL oid/generation, schema, and name +- current `relkind`, persistence, partition parent / child relation +- stable `ColumnId`, current attnum, name, type, typmod, nullability, default, + storage, dropped status +- current PK and replica identity +- expected source row set +- current destination, mapping, CH columns, and CH sorting key +- lifecycle policy and expected transition classification + +Renderer chooses only PostgreSQL-legal actions from current state. Keep invalid +DDL generation in a separate PostgreSQL-error campaign; syntax and dependency +errors provide no replication confidence + +Use small bounded programs for shrink quality: + +- one or two tables initially +- one to four columns from restricted type alphabet +- one to four transactions +- one to eight actions per transaction +- small row sets with deliberate boundary values + +Start type alphabet with bool, int4, int8, text, varchar, numeric, timestamp, +and bytea. Include NULL, empty values, integer width edges, numeric precision +edges, repeated strings, and external-TOAST-sized strings. Add arrays, JSON, +UUID, inet, intervals, domains, enums, and custom types after core oracle holds + +## Architecture + +Use two related harnesses + +### Pure coverage-guided targets + +Extend `fuzz/` crate from [FUZZ.md](FUZZ.md) with semantic targets. Keep each +invocation deterministic and free of PostgreSQL / ClickHouse processes + +| target | input | oracle | +|---|---|---| +| `ddl_schema_transition` | descriptor pair or chain | every changed field maps to explicit `Supported`, `NoChEffect`, `Policy`, or `Unsupported`; no silent unclassified drift | +| `ddl_mapping_transition` | descriptor chain + starting mapping | supported columns and names match reference mapping after each transition | +| `ddl_route_transition` | relation lifecycle + namespace config | every `Converge` heap resolves one destination; rename/drop/recreate never silently maps to stale generation | +| `ddl_drain_order` | heaps, schema controls, truncates, subxact order | merged walk preserves control-before-dependent-row and pre-control durability order | +| `ddl_plan_idempotence` | abstract sealed plan + restart cut | replayed mutation sequence converges to uninterrupted model state | +| `ddl_reject_atomicity` | abstract plan containing unsupported transition | validator rejects before first modeled side effect | + +Use `arbitrary` for structure generation. Normalize impossible combinations to +smaller valid cases rather than returning early; a target dominated by rejected +inputs produces misleading edge coverage + +### Real-cluster state machine + +Add `tests/ddl_fuzz_e2e.rs` and helpers under `tests/common/ddl_fuzz/`, reusing +`inproc_harness.rs`. Use seeded generator or `proptest`, not libFuzzer process +loop. External PostgreSQL and ClickHouse branches are invisible to libFuzzer, +and process/WAL switching cost makes per-input target invocation unsuitable + +Run multiple isolated programs under unique source namespaces before one WAL +switch. Drain once, then query every case. If stream fails, record current +program prefix and replay it alone. If state comparison fails, replay failing +case on fresh clusters and shrink there + +Keep campaign mode separate from normal deterministic CI: + +- fixed regression seeds always run under existing runtime skip gate +- bounded deterministic seed range runs in regular live CI +- broader seed range runs in scheduled PG 16 / 17 / 18 jobs +- long campaign records corpus and artifacts without changing source tree + +Do not use sleeps as correctness barriers. Wait for dispatched LSN, shadow +replay, emitter ack, and pipeline drain. Query CH only after relevant commit LSN +is acknowledged + +## Pure DDL planning seam + +Extract schema decision from live `DdlApplicator`. Current code combines event +classification, mapping mutation, SQL rendering, and CH I/O, which forces a +live client to test transition semantics + +Target shape: + +```rust +struct DdlPlan { + mutations: Vec, + mapping_delta: MappingDelta, + disposition: DdlDisposition, +} + +enum DdlDisposition { + Apply, + NoChEffect, + Policy, + Unsupported { reason: UnsupportedDdl }, +} + +fn plan_schema_transition( + old: Option<&RelDescriptor>, + new: Option<&RelDescriptor>, + mapping: Option<&TableMapping>, + config: &DdlConfigSnapshot, +) -> Result; +``` + +`DdlMutation` should express intent rather than raw SQL: + +- create / drop / rename table +- add / drop / rename / modify column +- rebuild-required key or engine change +- no-CH-effect metadata change + +Render SQL only after plan validates. Apply `mapping_delta` only after CH +mutation succeeds. Let transaction planner inspect `DdlDisposition` before any +heap or control side effect. This seam makes unsupported type, key, persistence, +and relation-name transitions fuzzable and gives `RejectBeforeEffects` a real +enforcement point + +Do not require this extraction before first real-cluster corpus. Deterministic +known-gap tests can land against current behavior, marked `KnownGap`. Require +pure planning seam before claiming broad semantic fuzz coverage + +## Differential oracles + +### Row state + +Reserve immutable surrogate key for first campaign and exclude it from DDL. +Query source into canonical typed rows ordered by key. Query effective CH state +using `_lsn` and `_is_deleted`, not physical part rows. Normalize: + +- explicit NULL distinct from type default +- bytea as hex +- numeric as canonical decimal text +- timestamps in UTC with declared precision +- floats by bit pattern, preserving NaN and infinities +- text and custom output as bytes where collation or formatting can vary + +Once stable-key campaign passes, enable key mutations and compare through +model-owned logical row identities. Do not group CH by newly expected source +key when target still uses old sorting key; that would hide key-transition bugs + +### Schema state + +Compare current source descriptor projection with `system.columns`: + +- destination existence and name +- mapped column existence and order-independent name set +- CH type and nullability through type bridge +- default where missing-value semantics rely on it +- synthetic column tail +- sorting key / engine metadata when source key transition is in contract + +Ignore source CHECK, FK, trigger, and ordinary index metadata unless it changes +replica identity or routing + +### Lifecycle state + +Track source relation generation independently of name. Assert: + +- DROP under `drop` removes destination and runtime-derived mapping +- recreate does not inherit rows from old source generation +- `retain` / `warn` behavior matches configured policy model +- rename or schema move cannot turn a mapped generation into silent unmapped + discard +- partition attach / detach follows declared fan-in or unsupported contract + +### Progress and diagnostics + +For every `Converge` case assert: + +- emitter ack reaches barrier LSN +- no pipeline fatal +- no rows route unmapped +- no unsupported relation or operation counter moves +- no rejected type-change counter moves +- no ambiguity overlaps emitted rows +- source and CH row/schema oracles match + +For `RejectBeforeEffects`, snapshot relevant CH state before transaction and +assert exact equality afterward. A fatal after pre-DDL rows became durable is +partial application, not successful rejection + +### Restart equivalence + +Compare uninterrupted result with restart at generated cuts: + +- after source commit, before pump observes commit +- after plan seal, before execution +- after pre-DDL data durability fence +- after each CH DDL mutation +- after TRUNCATE +- after post-DDL INSERT durability, before ack persistence +- between PREPARE and COMMIT / ROLLBACK PREPARED + +Use existing kill/restart hooks where available. Add deterministic failpoints +only at product-owned boundaries; do not infer cuts from wall-clock delays + +## Shrinking and artifacts + +Represent each program as versioned JSON independent of random generator. Every +failure artifact contains: + +- generator seed and program JSON +- rendered SQL by session +- expected classification +- source and CH schema snapshots +- canonical source and CH rows +- relevant LSNs and ack position +- walshadow counters and first fatal message +- PostgreSQL and ClickHouse versions +- walshadow config, especially namespace and drop policy + +Implement semantic shrink order: + +1. remove unrelated transactions +2. remove unrelated tables +3. remove actions while preserving references +4. collapse savepoint / subxact structure +5. remove columns not used by failing transition +6. reduce row count +7. shrink values toward NULL, zero, empty, width boundary, and short text +8. simplify types while preserving transition class + +Repair references after removal through stable `RelId` / `ColumnId`; never let +shrinker turn most candidates into invalid SQL. Emit minimized SQL beside JSON +and promote confirmed bugs into deterministic integration tests + +Keep large evolving corpora outside git. Check in small semantic seeds and +minimized regressions. Existing parser corpus policy in [FUZZ.md](FUZZ.md) +continues unchanged + +## Semantic coverage + +LLVM edge coverage measures walshadow code only; record operation-space +coverage explicitly. Persist counts for: + +```text +DDL kind +x DML kind before / after +x same / separate transaction +x commit / abort / savepoint / prepare +x descriptor transition class +x no rewrite / in-place / filenode rewrite +x nullable / non-nullable +x key unchanged / changed / dropped +x toast / inline +x mapped / auto-created / excluded +x drop strategy +x restart cut +x PostgreSQL major +``` + +Require all legal single operations and ordered operation pairs before growing +program depth. Weight generator toward uncovered pairs and prior bug +neighborhoods. Do not claim confidence from raw case count; report transition +coverage plus unique minimized failures + +Track outcome buckets separately: + +- converged +- expected policy divergence +- rejected before effects +- rejected after partial effects +- acked mismatch +- fatal mismatch +- unmapped discard +- generator / PostgreSQL rejection + +Generator rejection rate must stay low. High rejection means model emits +invalid SQL, not that product survived difficult cases + +## Campaign sequencing + +### Phase 0, contract and deterministic matrix + +- encode `Converge` / `RejectBeforeEffects` / `Policy` / `KnownGap` +- land known transition corpus and supported controls +- add canonical row and schema snapshot helpers +- record counters around each case +- resolve stale assertions or comments claiming DROP / RENAME COLUMN remain + unimplemented + +### Phase 1, descriptor and mapping model + +- define transition classifier over every `RelDescriptor` field +- add `ddl_schema_transition` and `ddl_mapping_transition` +- classify relname, persistence, replica identity, defaults, nullability, + typmod, physical type, relkind, and toast changes explicitly +- turn every unclassified descriptor drift into fuzz assertion + +### Phase 2, pure DDL plan + +- extract `plan_schema_transition` +- validate unsupported transition before execution +- add abstract CH schema model and SQL-render unit tests +- add route lifecycle, reject atomicity, and idempotent replay fuzz targets + +### Phase 3, generated live transactions + +- add one-table stable-key generator +- run fixed seeds under PG 16 / 17 / 18 +- add savepoints, subxacts, multiple tables, and clean-xact interleave +- promote every failure after semantic minimization + +### Phase 4, destructive and physical transitions + +- enable key changes, drop/recreate, rewrites, TOAST, unlogged transitions, + partitions, tablespaces +- connect explicit expected gaps to their owning plans +- add restart cuts and CH reconnect failures + +### Phase 5, sustained campaign + +- schedule bounded seed ranges per PG major +- persist JSON corpus and semantic coverage report +- deduplicate failures by normalized program plus first divergent oracle +- periodically replay full checked-in regression corpus against supported CH + versions + +## Acceptance + +- Every `RelDescriptor` field change has explicit transition classification; + fuzz target cannot produce silent unclassified drift +- Every supported DDL/DML ordered pair has at least one real-PG regression seed + on PG 16, 17, and 18 +- Generated `Converge` cases assert rows, schema, ack, and zero silent-discard / + unsupported counters +- Generated unsupported cases reject before any transaction side effect in CH +- Restart replay matches uninterrupted final state for every supported control + cut +- Known gaps remain executable, minimized, and reported separately; closing one + moves its seed to `Converge` or `RejectBeforeEffects` +- Failure JSON replays deterministically without original random seed +- Scheduled report publishes semantic pair coverage, outcome buckets, and new + minimized failures, not only execution count + +## Cross-links + +- [FUZZ.md](FUZZ.md) — byte parsing, CRC, codec, and C-boundary fuzzing +- [coverage100.md](coverage100.md) — line coverage and live DDL branch matrix +- [catalog_capture_completeness.md](catalog_capture_completeness.md) — relation / + namespace rename event gaps +- [pinned_ddl_baseline.md](pinned_ddl_baseline.md) — cold-start DDL baseline and + CH-existence drift +- [two_phase_commit.md](two_phase_commit.md) — prepared transaction restart + durability +- [TABLESPACES.md](TABLESPACES.md) — non-default tablespace bootstrap and shadow + path gaps +- [pipeline_backpressure_and_scaling.md](pipeline_backpressure_and_scaling.md) — + DDL barrier scope and pipeline ordering +- [../desc_log.md](../desc_log.md) — pending descriptor timeline and ambiguity + fence +- [../emitter.md](../emitter.md) — transaction plan, route freeze, DDL / + TRUNCATE execution, ack semantics From e2ac5541aa2007acc48fbdf43aee26a0a839cf56 Mon Sep 17 00:00:00 2001 From: serprex <159546+serprex@users.noreply.github.com> Date: Tue, 4 Aug 2026 19:05:56 +0000 Subject: [PATCH 4/8] refactor bench/, move run_bench_suite.sh into walshadow-ec2-bench --suite --- bench/ec2/README.md | 31 ++-- bench/ec2/ec2-bench/deploy.sh | 27 ++-- bench/ec2/ec2-peerdb/deploy.sh | 34 ++--- bench/ec2/ec2-peerdb/profile.sh | 48 ------- bench/ec2/ec2-pg-standby/deploy.sh | 9 +- bench/ec2/ec2-pg-standby/profile.sh | 51 ------- bench/ec2/ec2-source-pg/deploy.sh | 11 +- bench/ec2/ec2-walshadow/deploy.sh | 15 +- bench/ec2/ec2-walshadow/profile.sh | 43 ------ bench/ec2/lib.sh | 53 +++++-- bench/ec2/profile.sh | 105 ++++++++++++++ bench/run_bench_suite.sh | 86 ----------- bench/src/bin/ec2_bench.rs | 122 +++++++++++++++- bench/src/bin/local_bench.rs | 11 +- bench/src/lib.rs | 18 +-- bench/src/suite.rs | 215 ++++++++++++++++++++++++++++ 16 files changed, 545 insertions(+), 334 deletions(-) delete mode 100755 bench/ec2/ec2-peerdb/profile.sh delete mode 100755 bench/ec2/ec2-pg-standby/profile.sh delete mode 100755 bench/ec2/ec2-walshadow/profile.sh create mode 100755 bench/ec2/profile.sh delete mode 100755 bench/run_bench_suite.sh create mode 100644 bench/src/suite.rs diff --git a/bench/ec2/README.md b/bench/ec2/README.md index a7432384..ba9505de 100644 --- a/bench/ec2/README.md +++ b/bench/ec2/README.md @@ -20,9 +20,9 @@ into each node folder for `deploy.sh` / `profile.sh` / the bench to read. The desired node set (streamer, clickhouse, bench runner) persists in `terraform/setup.auto.tfvars`, written by `stack.sh`. Post-boot setup stays shell: node folders keep `cloud-init.yaml` and, where needed, `deploy.sh` / -`profile.sh` / `pre_down.sh`. Shared script helpers live in `aws-env.sh` -(creds) and `lib.sh` (ssh/state.env). `*.pem`, `state.env` and terraform state -are gitignored. +`pre_down.sh`. Shared scripts sit alongside `stack.sh`: `aws-env.sh` (creds), +`lib.sh` (ssh, state.env, readiness waits) and `profile.sh` (on-CPU capture for +any streamer). `*.pem`, `state.env` and terraform state are gitignored. Record account and profile in `aws.local.env`, which stays gitignored and is required by `aws-env.sh`. Account check aborts when credentials resolve to a @@ -54,7 +54,9 @@ cd bench/ec2 `terraform apply` is interactive — review the plan before confirming, especially on setup swaps (e.g. walshadow→pg destroys the ClickHouse node). Before a streamer node is destroyed or swapped, `stack.sh` copies any on-CPU -profiles off it and runs its `pre_down.sh` hook. Terraform can also be driven +profiles off it and runs its `pre_down.sh` hook. Start a capture with +`./profile.sh [secs]` just before a benchmark; it +returns immediately and teardown copies the result into the node folder. Terraform can also be driven directly: `source aws-env.sh && terraform -chdir=terraform plan`. Knobs like `instance_type` / `az` / `my_ip` are variables (see `terraform/variables.tf`); `instance_type` is global — the AZ is picked from its offerings. @@ -101,17 +103,24 @@ a read-only hot standby that streams WAL. Re-running takes a fresh base backup. ## Benchmark a setup -`walshadow-ec2-bench` reads endpoints from the relevant `state.env`. Use -`run_bench_suite.sh ` (at the bench crate root) to run all four benches into -`bench/results//` (a gitignored dir, created on demand): +`walshadow-ec2-bench` reads endpoints from the relevant `state.env`. `--suite +` runs all four benches into `bench/results//` (a gitignored dir, +created on demand; an existing name is refused). Run it from the repository +root, where the `--state-dir` and `--results-dir` defaults resolve: ```bash +B="cargo run --release --bin walshadow-ec2-bench --" # CDC engines (walshadow / peerdb) → ClickHouse: -../run_bench_suite.sh walshadow-run # DEST defaults to clickhouse +$B --suite walshadow-run # --dest defaults to clickhouse # physical standby: -DEST=postgres ../run_bench_suite.sh pg-run # reads ec2-pg-standby -# Run sustained and interleaved loads for five minutes; long run already uses 10 30-second rounds -RUN_SECS=300 ../run_bench_suite.sh walshadow-5min +$B --suite pg-run --dest postgres # reads ec2-pg-standby +# Run sustained and interleaved loads for five minutes; interleaved-long always runs 10 30-second rounds +$B --suite walshadow-5min --run-secs 300 +# one bench on its own, with its own knobs: +$B --bench interleaved --xact-secs 150 ``` +Each shape runs as a child process, so one failure does not end the pass: its +output is teed to `.txt` with a `# FAILED` footer and the suite exits +non-zero listing what failed. ## Notes - `c8i.2xlarge`s bill while running (~8× a t2.small) — `down` (or `down --all`) when idle. diff --git a/bench/ec2/ec2-bench/deploy.sh b/bench/ec2/ec2-bench/deploy.sh index b4d242f1..1545f53d 100755 --- a/bench/ec2/ec2-bench/deploy.sh +++ b/bench/ec2/ec2-bench/deploy.sh @@ -5,16 +5,16 @@ # # After this: ssh to the box and run, e.g. # walshadow-ec2-bench --network private --dest clickhouse --bench single-row --state-dir /opt/bench/ec2 -# # or all four via the shipped runner: -# BIN=walshadow-ec2-bench STATE_DIR=/opt/bench/ec2 NETWORK=private DEST=clickhouse \ -# SKIP_BUILD=1 /opt/bench/run_bench_suite.sh myrun +# # or the whole four-bench suite: +# walshadow-ec2-bench --network private --dest clickhouse --suite myrun \ +# --state-dir /opt/bench/ec2 --results-dir /opt/bench/results set -euo pipefail cd "$(dirname "$0")" source ./state.env # PUBLIC_IP, PEM, ... source ../lib.sh IMAGE="${IMAGE:-walshadow-bench:local}" -REPO_ROOT="$(cd ../../.. && pwd)" +REPO_ROOT="$(repo_root)" node_ssh_setup echo "building $IMAGE (from docker/Dockerfile.bench)…" @@ -35,9 +35,9 @@ fi # Ship the sibling state.env files so --network private can resolve endpoints. echo "shipping endpoint state.env files…" # List every level explicitly so /opt/bench and /opt/bench/ec2 are also -# ubuntu-owned (install -d only reliably applies -o to the leaf dirs) — needed -# so we can scp run_bench_suite.sh there and the runner can write results. -"${SSH[@]}" 'sudo install -d -o ubuntu /opt/bench /opt/bench/ec2 /opt/bench/ec2/ec2-source-pg /opt/bench/ec2/ec2-clickhouse /opt/bench/ec2/ec2-pg-standby' +# ubuntu-owned (install -d only reliably applies -o to the leaf dirs) — the scp +# below writes as ubuntu, and results stay readable without sudo. +"${SSH[@]}" 'sudo install -d -o ubuntu /opt/bench /opt/bench/results /opt/bench/ec2 /opt/bench/ec2/ec2-source-pg /opt/bench/ec2/ec2-clickhouse /opt/bench/ec2/ec2-pg-standby' for n in ec2-source-pg ec2-clickhouse ec2-pg-standby; do if [ -f "../$n/state.env" ]; then "${SCP[@]}" "../$n/state.env" "ubuntu@$PUBLIC_IP:/opt/bench/ec2/$n/state.env" @@ -46,22 +46,19 @@ for n in ec2-source-pg ec2-clickhouse ec2-pg-standby; do done # Install a wrapper: `walshadow-ec2-bench …` → runs the image with host -# networking and /opt/bench/ec2 mounted at the same path (so --state-dir works). +# networking and /opt/bench mounted at the same path, so --state-dir reads the +# shipped state.env files and --suite writes results back to the host. echo "installing walshadow-ec2-bench wrapper…" "${SSH[@]}" "cat | sudo tee /usr/local/bin/walshadow-ec2-bench >/dev/null && sudo chmod +x /usr/local/bin/walshadow-ec2-bench" <:" -echo " BIN=walshadow-ec2-bench STATE_DIR=/opt/bench/ec2 NETWORK=private DEST=clickhouse SKIP_BUILD=1 /opt/bench/run_bench_suite.sh myrun" -echo " # for the pg standby: DEST=postgres …" +echo " walshadow-ec2-bench --network private --dest clickhouse --suite myrun --state-dir /opt/bench/ec2 --results-dir /opt/bench/results" +echo " # for the pg standby: --dest postgres …" diff --git a/bench/ec2/ec2-peerdb/deploy.sh b/bench/ec2/ec2-peerdb/deploy.sh index 0e62c941..620d31e9 100755 --- a/bench/ec2/ec2-peerdb/deploy.sh +++ b/bench/ec2/ec2-peerdb/deploy.sh @@ -18,21 +18,12 @@ source ../lib.sh node_ssh_setup -SRC_PRIV="${SOURCE_PRIVATE_IP:-$(read_state_var ../ec2-source-pg/state.env SOURCE_PRIVATE_IP)}" -CH_PRIV="${CH_PRIVATE_IP:-$(read_state_var ../ec2-clickhouse/state.env PRIVATE_IP)}" -[ -n "$SRC_PRIV" ] || { echo "source PG private IP unknown (provision ec2-source-pg first)" >&2; exit 1; } -[ -n "$CH_PRIV" ] || { echo "clickhouse private IP unknown (provision ec2-clickhouse first)" >&2; exit 1; } +SRC_PRIV="${SOURCE_PRIVATE_IP:-$(require_state_ip ec2-source-pg SOURCE_PRIVATE_IP)}" +CH_PRIV="${CH_PRIVATE_IP:-$(require_state_ip ec2-clickhouse PRIVATE_IP)}" echo "source PG: $SRC_PRIV:5432 clickhouse: $CH_PRIV:9000" -echo "waiting for SSH on the host…" -ssh_ok=0 -for i in $(seq 1 30); do "${SSH[@]}" true 2>/dev/null && { ssh_ok=1; break; }; sleep 10; done -[ "$ssh_ok" = 1 ] || { echo "host not reachable over SSH after ~300s" >&2; exit 1; } - -# Block until cloud-init has actually finished (Docker install + PeerDB clone), -# rather than racing it. --wait returns non-zero if cloud-init errored. -echo "waiting for cloud-init to finish (Docker install + PeerDB clone)…" -"${SSH[@]}" 'sudo cloud-init status --wait' || { echo "cloud-init did not complete cleanly on the host" >&2; exit 1; } +# cloud-init installs Docker and clones PeerDB +wait_cloud_init "${SSH[@]}" 'command -v docker >/dev/null && [ -d /opt/peerdb ]' \ || { echo "host missing docker or /opt/peerdb after cloud-init" >&2; exit 1; } @@ -51,13 +42,7 @@ echo "bringing up the PeerDB stack (docker compose up -d; first run pulls severa # PeerDB SQL server (Postgres wire) on :9900. PSQL='sudo docker run --rm -i --network host postgres:17-alpine psql "host=127.0.0.1 port=9900 user=peerdb password=peerdb dbname=peerdb sslmode=disable"' -echo "waiting for the PeerDB SQL server on :9900…" -sql_ok=0 -for i in $(seq 1 40); do - "${SSH[@]}" "$PSQL -tAc 'select 1'" 2>/dev/null | grep -q 1 && { sql_ok=1; break; } - sleep 10 -done -[ "$sql_ok" = 1 ] || { echo "PeerDB SQL server not reachable on :9900 after ~400s" >&2; exit 1; } +retry_remote 40 10 "the PeerDB SQL server on :9900" "$PSQL -tAc 'select 1' | grep -q 1" echo "peerdb-server up" # The PeerDB quickstart runs one-shot inits that race their dependencies on a @@ -67,11 +52,8 @@ echo "peerdb-server up" # Re-assert both, idempotently, before configuring anything. echo "ensuring MinIO staging bucket 'peerdbbucket' exists…" "${SSH[@]}" "cd /opt/peerdb && sudo docker compose exec -T minio sh -c 'mc alias set m http://localhost:9000 _peerdb_minioadmin _peerdb_minioadmin >/dev/null 2>&1; mc mb -p m/peerdbbucket' 2>&1 | tail -1" || true -echo "ensuring Temporal search attribute 'MirrorName' is registered…" -for i in $(seq 1 6); do - "${SSH[@]}" "cd /opt/peerdb && sudo docker compose exec -T temporal-admin-tools temporal operator search-attribute create --namespace default --name MirrorName --type Keyword --address temporal:7233" 2>/dev/null && break - sleep 10 -done +retry_remote 6 10 "Temporal search attribute 'MirrorName'" \ + "cd /opt/peerdb && sudo docker compose exec -T temporal-admin-tools temporal operator search-attribute create --namespace default --name MirrorName --type Keyword --address temporal:7233" || true if [ "${CONFIGURE_MIRROR:-1}" = "1" ]; then # Drop an existing mirror first so re-running deploy.sh actually re-applies @@ -119,4 +101,4 @@ echo "=== deployed ===" echo "PeerDB UI: http://$PUBLIC_IP:3000" echo "PeerDB SQL: psql 'host=$PUBLIC_IP port=9900 user=peerdb password=peerdb dbname=peerdb'" echo "logs: ssh -i $PEM ubuntu@$PUBLIC_IP 'cd /opt/peerdb && sudo docker compose logs -f'" -echo "profile: ./profile.sh [secs] # start an on-CPU capture before the bench; teardown copies it back" +echo "profile: ../profile.sh peerdb [secs] # start an on-CPU capture before the bench; teardown copies it back" diff --git a/bench/ec2/ec2-peerdb/profile.sh b/bench/ec2/ec2-peerdb/profile.sh deleted file mode 100755 index 17475c51..00000000 --- a/bench/ec2/ec2-peerdb/profile.sh +++ /dev/null @@ -1,48 +0,0 @@ -#!/usr/bin/env bash -# Start an on-CPU profile of PeerDB for N seconds (default 120), in the -# background, then return — run this just before kicking off the benchmark so -# the capture covers it. Scoped to the PeerDB processes (NOT system-wide): -# * perf → every PeerDB compose container's PID (the whole stack's CPU, -# excluding the Docker daemon / kernel / OS) → /opt/profile/perf-.data -# * eBPF (bcc) → the flow-worker (the CDC engine; bcc profiles one process) -# → /opt/profile/oncpu-flowworker-.folded -# stack.sh down copies /opt/profile back to this machine. -# -# Usage: ./profile.sh [seconds] -# Note: tools are installed by cloud-init; in-container Go binaries may -# symbolize only partially from the host. -set -euo pipefail -cd "$(dirname "$0")" -source ./state.env # PUBLIC_IP, PEM - -DUR="${1:-120}" -SSH=(ssh -i "$PEM" -o StrictHostKeyChecking=accept-new -o ConnectTimeout=15 "ubuntu@$PUBLIC_IP") - -echo "starting on-CPU profile of PeerDB on $PUBLIC_IP for ${DUR}s (background)…" -"${SSH[@]}" "DUR='$DUR' bash -s" <<'PROF' -set -e -DUR="${DUR:-120}" -TS="$(date +%Y%m%d-%H%M%S)" -OUT=/opt/profile -sudo install -d -o ubuntu "$OUT" -sudo sysctl -w kernel.perf_event_paranoid=-1 kernel.kptr_restrict=0 >/dev/null 2>&1 || true - -# PIDs of every PeerDB container (excludes the Docker daemon / OS). -PIDS=$(cd /opt/peerdb && sudo docker compose ps -q 2>/dev/null \ - | xargs -r -I{} sudo docker inspect -f '{{.State.Pid}}' {} 2>/dev/null | paste -sd,) -[ -n "$PIDS" ] || { echo "no PeerDB container PIDs found — is the stack up?" >&2; exit 1; } -# flow-worker PID for the single-process eBPF profiler. -FW=$(sudo docker inspect -f '{{.State.Pid}}' "$(cd /opt/peerdb && sudo docker compose ps -q flow-worker 2>/dev/null)" 2>/dev/null || echo "") -echo "PeerDB pids (perf): $PIDS flow-worker pid (eBPF): ${FW:-none}" - -# Detach so it survives this SSH session; chown output back to ubuntu at the end. -sudo nohup bash -c " - { [ -n \"$FW\" ] && profile-bpfcc -F 99 -f -p $FW $DUR > $OUT/oncpu-flowworker-$TS.folded 2>$OUT/oncpu-$TS.log ; } & - perf record -F 99 -g -p $PIDS -o $OUT/perf-$TS.data -- sleep $DUR 2>>$OUT/perf-$TS.log \ - || echo 'perf record failed (see log)' >>$OUT/perf-$TS.log - wait - chown -R ubuntu $OUT -" >/dev/null 2>&1 & -echo "capturing ${DUR}s → $OUT/perf-$TS.data + oncpu-flowworker-$TS.folded (background)" -PROF -echo "started — now kick off the benchmark. ../stack.sh down copies the profiles back." diff --git a/bench/ec2/ec2-pg-standby/deploy.sh b/bench/ec2/ec2-pg-standby/deploy.sh index 1753d3ac..7a6bcb0d 100755 --- a/bench/ec2/ec2-pg-standby/deploy.sh +++ b/bench/ec2/ec2-pg-standby/deploy.sh @@ -17,15 +17,10 @@ source ../lib.sh PG_IMAGE="${PG_IMAGE:-postgres:17-bookworm}" node_ssh_setup -SRC_PRIV="${SOURCE_PRIVATE_IP:-$(read_state_var ../ec2-source-pg/state.env SOURCE_PRIVATE_IP)}" -[ -n "$SRC_PRIV" ] || { echo "source PG private IP unknown (provision ec2-source-pg first)" >&2; exit 1; } +SRC_PRIV="${SOURCE_PRIVATE_IP:-$(require_state_ip ec2-source-pg SOURCE_PRIVATE_IP)}" echo "primary: $SRC_PRIV:5432 standby image: $PG_IMAGE" -echo "waiting for SSH + cloud-init…" -ssh_ok=0 -for i in $(seq 1 30); do "${SSH[@]}" true 2>/dev/null && { ssh_ok=1; break; }; sleep 10; done -[ "$ssh_ok" = 1 ] || { echo "host not reachable over SSH after ~300s" >&2; exit 1; } -"${SSH[@]}" 'sudo cloud-init status --wait' || { echo "cloud-init did not finish cleanly" >&2; exit 1; } +wait_cloud_init # Fresh base backup each deploy: stop any old standby, recreate the data volume, # then pg_basebackup -R as the postgres user (so the data dir ownership is right diff --git a/bench/ec2/ec2-pg-standby/profile.sh b/bench/ec2/ec2-pg-standby/profile.sh deleted file mode 100755 index 3dc9c372..00000000 --- a/bench/ec2/ec2-pg-standby/profile.sh +++ /dev/null @@ -1,51 +0,0 @@ -#!/usr/bin/env bash -# On-CPU profile of the STANDBY Postgres for N seconds (default 120), in the -# background, then return — run just before the benchmark so the capture covers -# it. We profile the destination (standby), not the primary: physical -# replication's work is the WAL replay on the standby, whereas the primary's CPU -# is the insert workload (common to every engine). Scoped to the standby's -# Postgres processes (NOT system-wide): -# * perf → every process in the pg-standby container (postmaster, walreceiver, -# startup, checkpointer, …) → /opt/profile/perf-.data -# * eBPF (bcc) → the startup/recovery process (the WAL-apply worker; bcc -# profiles one process) → /opt/profile/oncpu-startup-.folded -# stack.sh down copies /opt/profile back to this machine. -# -# Usage: ./profile.sh [seconds] -set -euo pipefail -cd "$(dirname "$0")" -source ./state.env # PUBLIC_IP, PEM - -DUR="${1:-120}" -SSH=(ssh -i "$PEM" -o StrictHostKeyChecking=accept-new -o ConnectTimeout=15 "ubuntu@$PUBLIC_IP") - -echo "starting on-CPU profile of the standby on $PUBLIC_IP for ${DUR}s (background)…" -"${SSH[@]}" "DUR='$DUR' bash -s" <<'PROF' -set -e -DUR="${DUR:-120}" -TS="$(date +%Y%m%d-%H%M%S)" -OUT=/opt/profile -sudo install -d -o ubuntu "$OUT" -sudo sysctl -w kernel.perf_event_paranoid=-1 kernel.kptr_restrict=0 >/dev/null 2>&1 || true - -# The standby is the only Postgres on this box, so grab its processes by name -# (host PIDs — the container shares the host kernel). Robust regardless of the -# container name; falls back to `docker top pg-standby` if pgrep finds nothing. -PIDS=$(sudo pgrep -x postgres 2>/dev/null | paste -sd,) -[ -n "$PIDS" ] || PIDS=$(sudo docker top pg-standby -eo pid --no-headers 2>/dev/null | awk '{print $1}' | paste -sd,) -[ -n "$PIDS" ] || { echo "no postgres processes found on this box — is the standby up?" >&2; exit 1; } -# The startup/recovery process replays the streamed WAL — the apply work. -SU=$(sudo pgrep -f 'postgres: startup' 2>/dev/null | head -1 || true) -[ -n "$SU" ] || SU=$(sudo pgrep -f 'postgres: .*recover' 2>/dev/null | head -1 || true) -echo "standby pids (perf): $PIDS startup/replay pid (eBPF): ${SU:-none}" - -sudo nohup bash -c " - { [ -n \"$SU\" ] && profile-bpfcc -F 99 -f -p $SU $DUR > $OUT/oncpu-startup-$TS.folded 2>$OUT/oncpu-$TS.log ; } & - perf record -F 99 -g -p $PIDS -o $OUT/perf-$TS.data -- sleep $DUR 2>>$OUT/perf-$TS.log \ - || echo 'perf record failed (see log)' >>$OUT/perf-$TS.log - wait - chown -R ubuntu $OUT -" >/dev/null 2>&1 & -echo "capturing ${DUR}s → $OUT/perf-$TS.data + oncpu-startup-$TS.folded (background)" -PROF -echo "started — now kick off the benchmark (DEST=postgres). ../stack.sh down copies the profiles back." diff --git a/bench/ec2/ec2-source-pg/deploy.sh b/bench/ec2/ec2-source-pg/deploy.sh index 40e4bc72..e85c7e0b 100755 --- a/bench/ec2/ec2-source-pg/deploy.sh +++ b/bench/ec2/ec2-source-pg/deploy.sh @@ -22,18 +22,15 @@ cd "$(dirname "$0")" source ./state.env # PUBLIC_IP, PEM, ... source ../lib.sh -REPO_ROOT="$(cd ../../.. && pwd)" -INSTALL_SQL="$REPO_ROOT/sql/runtime_config_install.sql" +INSTALL_SQL="$(repo_root)/sql/runtime_config_install.sql" [ -f "$INSTALL_SQL" ] || { echo "missing $INSTALL_SQL" >&2; exit 1; } node_ssh_setup PSQL=(sudo docker exec -i source psql -v ON_ERROR_STOP=1 -U postgres -d postgres) -echo "waiting for the source Postgres container to accept connections..." -for i in $(seq 1 30); do - "${SSH[@]}" 'command -v docker >/dev/null && sudo docker exec source pg_isready -U postgres >/dev/null 2>&1' 2>/dev/null && break - sleep 10 -done +wait_cloud_init +retry_remote 30 10 "the source Postgres container to accept connections" \ + 'sudo docker exec source pg_isready -U postgres >/dev/null 2>&1' echo "installing walshadow.config_* overlay from sql/runtime_config_install.sql..." "${SSH[@]}" "${PSQL[*]}" < "$INSTALL_SQL" diff --git a/bench/ec2/ec2-walshadow/deploy.sh b/bench/ec2/ec2-walshadow/deploy.sh index 772fcab0..8959a478 100755 --- a/bench/ec2/ec2-walshadow/deploy.sh +++ b/bench/ec2/ec2-walshadow/deploy.sh @@ -44,19 +44,13 @@ TRACE_SAMPLE_RATIO="${TRACE_SAMPLE_RATIO:-0.01}" XACT_BUFFER_MAX="${XACT_BUFFER_MAX:-1073741824}" node_ssh_setup -SRC_PRIV="${SOURCE_PRIVATE_IP:-$(read_state_var ../ec2-source-pg/state.env SOURCE_PRIVATE_IP)}" +SRC_PRIV="${SOURCE_PRIVATE_IP:-$(require_state_ip ec2-source-pg SOURCE_PRIVATE_IP)}" # CH host: explicit CH_HOST (e.g. Cloud endpoint) wins, else legacy # CH_PRIVATE_IP, else the in-VPC ec2-clickhouse node. -CH_HOST="${CH_HOST:-${CH_PRIVATE_IP:-$(read_state_var ../ec2-clickhouse/state.env PRIVATE_IP)}}" -[ -n "$SRC_PRIV" ] || { echo "source PG private IP unknown (provision ec2-source-pg first)" >&2; exit 1; } -[ -n "$CH_HOST" ] || { echo "CH host unknown (set CH_HOST=… or provision ec2-clickhouse first)" >&2; exit 1; } +CH_HOST="${CH_HOST:-${CH_PRIVATE_IP:-$(require_state_ip ec2-clickhouse PRIVATE_IP)}}" echo "source PG: $SRC_PRIV:5432 clickhouse: $CH_HOST:$CH_PORT (secure=$CH_SECURE, db=$CH_DATABASE)" -echo "waiting for Docker on the host (cloud-init may still be running)..." -for i in $(seq 1 30); do - "${SSH[@]}" 'command -v docker >/dev/null && sudo docker info >/dev/null 2>&1' 2>/dev/null && break - sleep 10 -done +wait_cloud_init # Ship the image unless it's already present on the host (use FORCE=1 to resend). REMOTE_IMAGE="$(remote_image_tag "$IMAGE")" @@ -131,12 +125,11 @@ EOF if [ "${FORCE_METRICS:-0}" = "1" ]; then GRAFANA_IMAGE="${GRAFANA_IMAGE:-grafana/grafana:13.0.2}" PROM_IMAGE="${PROM_IMAGE:-prom/prometheus:v3.12.0}" - REPO_ROOT="$(cd ../../.. && pwd)" "${SSH[@]}" "sudo docker network inspect walshadow-net >/dev/null 2>&1 || sudo docker network create walshadow-net" "${SSH[@]}" "sudo docker network connect walshadow-net walshadow 2>/dev/null || true" - tar -C "$REPO_ROOT/docker" -czf - grafana prometheus \ + tar -C "$(repo_root)/docker" -czf - grafana prometheus \ | "${SSH[@]}" "sudo install -d /opt/walshadow/obs && sudo tar -C /opt/walshadow/obs -xzf -" # compose hostname 'clickhouse' -> private IP "${SSH[@]}" "sudo grep -rl clickhouse /opt/walshadow/obs 2>/dev/null | sudo xargs -r sed -i 's/clickhouse:/$CH_HOST:/g; s#//clickhouse#//$CH_HOST#g'" diff --git a/bench/ec2/ec2-walshadow/profile.sh b/bench/ec2/ec2-walshadow/profile.sh deleted file mode 100755 index af3ac785..00000000 --- a/bench/ec2/ec2-walshadow/profile.sh +++ /dev/null @@ -1,43 +0,0 @@ -#!/usr/bin/env bash -# Background on-CPU profile (perf over the container, eBPF over walshadow-stream) -# for N seconds. Usage: ./profile.sh [seconds] -set -euo pipefail -cd "$(dirname "$0")" -source ./state.env # PUBLIC_IP, PEM - -DUR="${1:-120}" -SSH=(ssh -i "$PEM" -o StrictHostKeyChecking=accept-new -o ConnectTimeout=15 "ubuntu@$PUBLIC_IP") - -echo "starting on-CPU profile of walshadow on $PUBLIC_IP for ${DUR}s (background)…" -"${SSH[@]}" "DUR='$DUR' bash -s" <<'PROF' -set -e -DUR="${DUR:-120}" -TS="$(date +%Y%m%d-%H%M%S)" -OUT=/opt/profile -sudo install -d -o ubuntu "$OUT" -sudo sysctl -w kernel.perf_event_paranoid=-1 kernel.kptr_restrict=0 >/dev/null 2>&1 || true - -PIDS=$(sudo docker top walshadow -eo pid --no-headers 2>/dev/null | awk '{print $1}' | paste -sd,) -[ -n "$PIDS" ] || PIDS=$(sudo pgrep -f 'walshadow-stream|postgres' 2>/dev/null | paste -sd,) -[ -n "$PIDS" ] || { echo "no walshadow processes found — is the daemon up?" >&2; exit 1; } -WS=$(sudo pgrep -f 'walshadow-stream' 2>/dev/null | head -1 || true) -[ -n "$WS" ] || WS=$(sudo docker inspect -f '{{.State.Pid}}' walshadow 2>/dev/null || true) -{ [ -n "$WS" ] && sudo test -d "/proc/$WS"; } \ - || { echo "walshadow-stream daemon not running — nothing to profile (is the walshadow container up?)" >&2; exit 1; } -echo "walshadow pids (perf): $PIDS walshadow-stream pid (eBPF): $WS" - -sudo nohup bash -c " - profile-bpfcc -F 99 -f -p $WS $DUR > $OUT/oncpu-walshadow-$TS.folded 2>$OUT/oncpu-$TS.log & - # Re-filter to live PIDs: perf -p aborts the record if any one has exited. - LIVE=\"\"; for x in \$(echo '$PIDS' | tr ',' ' '); do [ -d /proc/\$x ] && LIVE=\"\$LIVE,\$x\"; done; LIVE=\${LIVE#,} - # DWARF unwinding (musl build, no frame pointers); --no-buildid* avoids - # finalization that fails on this host. - perf record -F 99 --call-graph dwarf,65528 -m 64M --no-buildid --no-buildid-cache \ - -p \"\$LIVE\" -o $OUT/perf-$TS.data -- sleep $DUR 2>>$OUT/perf-$TS.log \ - || echo 'perf record failed (see log)' >>$OUT/perf-$TS.log - wait - chown -R ubuntu $OUT -" >/dev/null 2>&1 & -echo "capturing ${DUR}s → $OUT/perf-$TS.data + oncpu-walshadow-$TS.folded (background)" -PROF -echo "started — now kick off the benchmark. ../stack.sh down copies the profiles back." diff --git a/bench/ec2/lib.sh b/bench/ec2/lib.sh index b1b5f1d6..c7fd2702 100644 --- a/bench/ec2/lib.sh +++ b/bench/ec2/lib.sh @@ -1,12 +1,24 @@ #!/usr/bin/env bash -# Shared helpers for the per-node deploy.sh / profile.sh scripts. Provisioning -# is terraform (terraform/, driven by stack.sh), which writes each node's -# ./state.env (PUBLIC_IP, PRIVATE_IP, PEM, ...). Source state.env BEFORE +# Shared helpers for the per-node deploy.sh and the shared profile.sh. +# Provisioning is terraform (terraform/, driven by stack.sh), which writes each +# node's ./state.env (PUBLIC_IP, PRIVATE_IP, PEM, ...). Source state.env BEFORE # lib.sh; helpers run from the node dir. # Echo KEY=value's value from a state.env-style file. $1=path, $2=key. read_state_var() { grep -E "^$2=" "$1" 2>/dev/null | tail -1 | cut -d= -f2-; } +# Echo a sibling node's IP, failing when terraform has not written it yet. +# $1=node dir, $2=state.env key. +require_state_ip() { + local ip + ip="$(read_state_var "../$1/state.env" "$2")" + [ -n "$ip" ] || { echo "$2 unknown in ../$1/state.env — provision $1 first, or pass the host explicitly" >&2; return 1; } + echo "$ip" +} + +# Absolute repository root, from a node dir. +repo_root() { (cd ../../.. && pwd); } + # deploy.sh preamble helper: set SSH/SCP arrays from the sourced state.env # (PEM, PUBLIC_IP). Populates globals SSH, SCP. node_ssh_setup() { @@ -15,6 +27,22 @@ node_ssh_setup() { SCP=(scp -i "$PEM" -o StrictHostKeyChecking=accept-new) } +# Run a remote command until it succeeds. Fails after the whole window, so a +# node that never comes up says so instead of surfacing as a confusing error +# from whatever step ran next. +# $1=attempts $2=delay-secs $3=label $4...=remote command +retry_remote() { + local attempts="$1" delay="$2" label="$3" i + shift 3 + echo "waiting for $label…" + for ((i = 0; i < attempts; i++)); do + "${SSH[@]}" "$@" 2>/dev/null && return 0 + sleep "$delay" + done + echo "$label not ready after $((attempts * delay))s" >&2 + return 1 +} + # Print matching remote image tag, including Podman's localhost prefix remote_image_tag() { local ref @@ -26,23 +54,26 @@ remote_image_tag() { done } -# Block until cloud-init has finished on the node (SSH must be set up). +# Block until the node answers SSH and cloud-init has finished (SSH must be set +# up). Everything a deploy needs on the box comes from cloud-init, so this is +# the readiness gate for all of them. wait_cloud_init() { - echo "waiting for SSH + cloud-init…" + retry_remote 30 10 "SSH on $PUBLIC_IP" true + echo "waiting for cloud-init…" "${SSH[@]}" 'sudo cloud-init status --wait' || { echo "cloud-init did not finish cleanly" >&2; return 1; } } -# Copy on-CPU profiles (from ./profile.sh) off the box into ./profiles// +# Copy on-CPU profiles (from ../profile.sh) off the box into ./profiles// # BEFORE the node is destroyed — stack.sh runs this for the outgoing streamer. copy_remote_profiles() { [ -n "${PUBLIC_IP:-}" ] && [ -f "${PEM:-}" ] || return 0 - local ssh_p=(ssh -i "$PEM" -o StrictHostKeyChecking=accept-new -o ConnectTimeout=10 "ubuntu@$PUBLIC_IP") - "${ssh_p[@]}" 'ls /opt/profile/* >/dev/null 2>&1' || return 0 - local dest="./profiles/$(date +%Y%m%d-%H%M%S)" + node_ssh_setup + "${SSH[@]}" 'ls /opt/profile/* >/dev/null 2>&1' || return 0 + local dest + dest="./profiles/$(date +%Y%m%d-%H%M%S)" mkdir -p "$dest" echo "copying /opt/profile → $dest …" - scp -i "$PEM" -o StrictHostKeyChecking=accept-new -o ConnectTimeout=10 \ - "ubuntu@$PUBLIC_IP:/opt/profile/*" "$dest/" 2>/dev/null \ + "${SCP[@]}" "ubuntu@$PUBLIC_IP:/opt/profile/*" "$dest/" 2>/dev/null \ && echo " copied: $(ls -1 "$dest" 2>/dev/null | tr '\n' ' ')" \ || echo " (nothing copied — capture may still be running; re-run down after it finishes)" } diff --git a/bench/ec2/profile.sh b/bench/ec2/profile.sh new file mode 100755 index 00000000..fc687d22 --- /dev/null +++ b/bench/ec2/profile.sh @@ -0,0 +1,105 @@ +#!/usr/bin/env bash +# Start an on-CPU profile of a node's replication engine for N seconds (default +# 120) in the background, then return — run this just before kicking off the +# benchmark so the capture covers it. Both captures are scoped to the engine's +# processes, NOT system-wide (no Docker daemon, kernel threads or OS): +# * perf → every process of the engine → /opt/profile/perf-.data +# * eBPF (bcc, one process) → its apply loop → /opt/profile/oncpu-