From 316fa6d9b7fed15012f2b466fd1709e65bf8b0b6 Mon Sep 17 00:00:00 2001 From: Robin Portigliatti Date: Wed, 3 Jun 2026 00:26:51 +0200 Subject: [PATCH 1/2] feat(repo-sync): mirror backups + WAL between repositories Add repo-sync (Layer 1): byte-identical mirroring of completed backups and archived WAL from the active repository to every other configured repository, plus an inline `--repo-sync` option on `backup`. Repositories in a multi-repo set are configured as identical mirrors (global compression; identical bundling / block-incremental / cipher-type; shared cipher sub-key), so syncing is a pure raw byte copy of the stored objects at the same repo paths -- no decode / re-encode / re-bundle / re-block / size recompute. Consistency is validated up front: repo-sync refuses mismatched repositories and aligns a fresh encrypted target's sub-key to the source's so the copied bytes decrypt. Intended topology: the primary archives to repo1 only; repo1 then synchronises to repo2..N, run where repo1 is local. - crates/pgbr-command/src/sync/{mod,backup,archive}.rs: the engine -- assert_consistent, sync_backup_to_repo (diff/incr ancestor backfill, manifest-sentinel idempotency, verbatim backup.info merge), sync_archive_to_repo (idempotent WAL copy), and the `repo-sync` command (--type=all|wal|backup, --set=LABEL). - backup: new `repo-sync` boolean mirrors the just-completed backup to the other repositories before expire-auto. - config.yaml / help.xml: the `repo-sync` command + global option, and the stanza / repo / repo-bundle / repo-block command associations they require. - integration: new Docker scenario 12-repo-sync.sh (encrypted, bundle + block, archive-to-repo1-then-sync, byte-identical verification, restore from mirror) and Vagrant scenario 25; a repo-sync smoke check in 08-multi-repo.sh. Layer 2 (automatic server-triggered fan-out on WAL receipt over the TLS server topology) is a follow-up. Validated: fmt, clippy --all-targets -D warnings, cargo test --workspace (28 new sync tests); all 12 Docker scenarios pass; full Vagrant suite 211 passed / 0 failed. Co-Authored-By: Claude Opus 4.8 (1M context) --- crates/pgbr-build/inputs/config.yaml | 28 + crates/pgbr-build/inputs/help.xml | 48 ++ crates/pgbr-cli/src/lib.rs | 2 + crates/pgbr-command/src/archive.rs | 8 +- crates/pgbr-command/src/backup.rs | 99 ++- crates/pgbr-command/src/lib.rs | 4 +- crates/pgbr-command/src/restore.rs | 54 +- crates/pgbr-command/src/sync/archive.rs | 373 +++++++++++ crates/pgbr-command/src/sync/backup.rs | 785 ++++++++++++++++++++++++ crates/pgbr-command/src/sync/mod.rs | 597 ++++++++++++++++++ integration/scenarios/08-multi-repo.sh | 13 + integration/scenarios/12-repo-sync.sh | 164 +++++ integration/scenarios/run-all.sh | 5 + integration/vagrant/run-validation.sh | 103 ++++ 14 files changed, 2244 insertions(+), 39 deletions(-) create mode 100644 crates/pgbr-command/src/sync/archive.rs create mode 100644 crates/pgbr-command/src/sync/backup.rs create mode 100644 crates/pgbr-command/src/sync/mod.rs create mode 100644 integration/scenarios/12-repo-sync.sh diff --git a/crates/pgbr-build/inputs/config.yaml b/crates/pgbr-build/inputs/config.yaml index c610575c2..bb42f3e57 100644 --- a/crates/pgbr-build/inputs/config.yaml +++ b/crates/pgbr-build/inputs/config.yaml @@ -126,6 +126,13 @@ command: log-level-default: DEBUG parameter-allowed: true + repo-sync: + command-role: + remote: {} + lock-remote-required: true + lock-required: true + lock-type: backup + restore: command-role: local: {} @@ -403,6 +410,7 @@ option: manifest: default: latest required: true + repo-sync: {} restore: default: latest required: true @@ -440,6 +448,7 @@ option: required: false repo-rm: required: false + repo-sync: {} restore: {} stanza-create: {} stanza-delete: {} @@ -524,6 +533,12 @@ option: - diff - incr default: full + repo-sync: + allow-list: + - all + - wal + - backup + default: all info: required: false allow-list: @@ -1202,6 +1217,15 @@ option: command-role: main: {} + repo-sync: + section: global + type: boolean + default: false + command: + backup: {} + command-role: + main: {} + start-fast: section: global type: boolean @@ -1648,6 +1672,7 @@ option: repo-ls: {} repo-put: {} repo-rm: {} + repo-sync: {} restore: {} server: {} stanza-create: @@ -1855,6 +1880,7 @@ option: default: false command: backup: {} + repo-sync: {} command-role: main: {} @@ -1866,6 +1892,7 @@ option: allow-range: [1MiB, 1PiB] command: backup: {} + repo-sync: {} command-role: main: {} depend: @@ -1885,6 +1912,7 @@ option: default: false command: backup: {} + repo-sync: {} command-role: main: {} depend: diff --git a/crates/pgbr-build/inputs/help.xml b/crates/pgbr-build/inputs/help.xml index 15d733f3b..c7310fa10 100644 --- a/crates/pgbr-build/inputs/help.xml +++ b/crates/pgbr-build/inputs/help.xml @@ -1543,6 +1543,16 @@ y + + Sync the just-completed backup to other repositories. + + +

When multiple repositories are configured and this option is enabled, the backup just written to the active repository is mirrored byte-for-byte to every other configured repository immediately after a successful backup. Because repositories in a set are identical mirrors the copy is a raw byte copy with no recompression or re-encryption. The standalone repo-sync command can also be run later to reconcile repositories.

+ + + y + + Stop prior failed backup on new backup. @@ -2504,6 +2514,44 @@ + + Sync backups and WAL between repositories. + + +

Mirrors backups and archived WAL from the active repository (the highest priority repository, or the one given by {[dash]}-repo) to every other configured repository. Repositories in a set are byte-identical mirrors, so the sync is a raw byte copy of the stored objects with no recompression, re-encryption, or re-bundling.

+ +

The command is idempotent: objects already present in a destination repository are skipped, and the ancestors of a differential or incremental backup are synced first so dependency chains are never broken.

+ + + + + + + + + Restore a database cluster. diff --git a/crates/pgbr-cli/src/lib.rs b/crates/pgbr-cli/src/lib.rs index f5967857b..74a38989b 100644 --- a/crates/pgbr-cli/src/lib.rs +++ b/crates/pgbr-cli/src/lib.rs @@ -328,6 +328,8 @@ fn worker_loaded_config(resolved: &ResolvedCli) -> LoadedConfig { const MULTI_REPO_COMMANDS: &[&str] = &[ "archive-push", "archive-get", + "backup", + "repo-sync", "stanza-create", "stanza-delete", "stanza-upgrade", diff --git a/crates/pgbr-command/src/archive.rs b/crates/pgbr-command/src/archive.rs index af3fe7156..19c7c0b39 100644 --- a/crates/pgbr-command/src/archive.rs +++ b/crates/pgbr-command/src/archive.rs @@ -288,7 +288,7 @@ fn write_segment_to_dest_arg(bytes: &[u8], dest_arg: &Path) -> Result<(), Comman /// `archive_id` is `-` from the stanza's `archive.info` /// (see [`archive_id`]), matching pgBackRust's per-cluster archive directory /// and the layout the `check` command polls. -fn repo_segment_path(stanza: &str, archive_id: &str, name: &str) -> PathBuf { +pub(crate) fn repo_segment_path(stanza: &str, archive_id: &str, name: &str) -> PathBuf { PathBuf::from(format!("archive/{stanza}/{archive_id}/{name}")) } @@ -296,7 +296,7 @@ fn repo_segment_path(stanza: &str, archive_id: &str, name: &str) -> PathBuf { /// `-` (e.g. `"16-1"`). This is the per-cluster /// subdirectory archived WAL is stored under, matching the C implementation /// and the `check` command. -fn archive_id(info: &InfoArchive) -> String { +pub(crate) fn archive_id(info: &InfoArchive) -> String { format!("{}-{}", info.db_version, info.db_id) } @@ -617,7 +617,7 @@ fn is_checkable_wal_segment(name: &str) -> bool { /// [`CommandError::MissingOption`] when an encrypted repo has no /// `repo-cipher-pass`; [`CommandError::Other`] when the info file cannot be /// loaded / decrypted; [`CommandError::Storage`] on an underlying storage error. -fn load_archive_info( +pub(crate) fn load_archive_info( config: &LoadedConfig, repo_storages: &[&dyn Storage], stanza: &str, @@ -1590,7 +1590,7 @@ fn fetch_segment_with_retry( /// Whether `repo` holds `segment` for `stanza` in any stored form (plaintext or /// a compressed suffix). Used by [`get`] to pick the first repository that has /// the segment. -fn repo_has_segment(repo: &dyn Storage, stanza: &str, archive_id: &str, segment: &str) -> Result { +pub(crate) fn repo_has_segment(repo: &dyn Storage, stanza: &str, archive_id: &str, segment: &str) -> Result { if repo.exists(&repo_segment_path(stanza, archive_id, segment))? { return Ok(true); } diff --git a/crates/pgbr-command/src/backup.rs b/crates/pgbr-command/src/backup.rs index cb3c5451c..2914f492e 100644 --- a/crates/pgbr-command/src/backup.rs +++ b/crates/pgbr-command/src/backup.rs @@ -740,7 +740,12 @@ fn walk_into(storage: &dyn Storage, dir: &Path, rel_prefix: &str, out: &mut Vec< /// /// See [`run_backup`]; plus connection failures and `backup-standby=y` with no /// reachable standby. -pub fn backup(config: &LoadedConfig, repo_storage: &dyn Storage, pg_storage: &dyn Storage) -> Result<(), CommandError> { +pub fn backup( + config: &LoadedConfig, + repo_storage: &dyn Storage, + repo_storages: &[(u32, &dyn Storage)], + pg_storage: &dyn Storage, +) -> Result<(), CommandError> { let stanza = require_stanza(config)?; // Refuse to run when the operator has called `stop` for this stanza (or // `stop --force` which writes `all.stop` and blocks every stanza). The @@ -887,6 +892,13 @@ pub fn backup(config: &LoadedConfig, repo_storage: &dyn Storage, pg_storage: &dy outcome.label, outcome.file_count, outcome.total_size )); + // repo-sync: mirror the just-completed backup to every other configured + // repository, byte-for-byte. Skipped on a dry run (nothing was written to + // sync) and when --repo-sync is off or only one repo exists. + if repo_sync_enabled(config) && !policy.dry_run { + sync_to_other_repos(config, stanza, &outcome.label, repo_storage, repo_storages)?; + } + // expire-auto (default on): apply retention right after a successful backup, // unless this was a dry run (nothing was added to expire against) or the user // disabled it. Calls the expire engine directly. C ref: backup.c runs @@ -903,6 +915,37 @@ pub fn backup(config: &LoadedConfig, repo_storage: &dyn Storage, pg_storage: &dy Ok(()) } +/// Mirror a just-completed backup `label` to every configured repository other +/// than the active (source) one, byte-for-byte (C ref: the post-backup +/// repo-sync step). The active repository is excluded from the target set; an +/// empty target set (single-repo configuration) is a logged no-op. +fn sync_to_other_repos( + config: &LoadedConfig, + stanza: &str, + label: &str, + repo_storage: &dyn Storage, + repo_storages: &[(u32, &dyn Storage)], +) -> Result<(), CommandError> { + let src_index = crate::cipher::active_repo_index(config); + let targets: Vec<(u32, &dyn Storage)> = repo_storages.iter().copied().filter(|(idx, _)| *idx != src_index).collect(); + if targets.is_empty() { + log_info("repo-sync: no other repositories configured; nothing to mirror"); + return Ok(()); + } + log_info(&format!( + "repo-sync: mirroring backup {label} to {} repositor(y/ies)", + targets.len() + )); + for (dst_index, dst) in targets { + let synced = crate::sync::backup::sync_backup_to_repo(config, stanza, label, repo_storage, src_index, dst, dst_index)?; + log_info(&format!( + "repo-sync: repo{dst_index} <- backup {label}: {} object(s), {} byte(s)", + synced.items, synced.bytes + )); + } + Ok(()) +} + /// The backup-control connections resolved per the `backup-standby` policy. /// /// Each control object is a `Box` so a *local* cluster @@ -1177,6 +1220,16 @@ fn expire_auto_enabled(config: &LoadedConfig) -> bool { boolean_default_true(config, "expire-auto") } +/// Whether the inline `--repo-sync` option is enabled (default **false**). When +/// on, a successful backup is mirrored to every other configured repository +/// immediately after completion. C ref: the post-backup repo-sync hook. +fn repo_sync_enabled(config: &LoadedConfig) -> bool { + matches!( + config.options.get(&("repo-sync".to_owned(), None)), + Some(OptionValue::Boolean(true)) + ) +} + /// Default `manifest-save-threshold` (1 GiB) when the option is absent, matching /// the option model's `1GiB` default. const DEFAULT_MANIFEST_SAVE_THRESHOLD: u64 = 1024 * 1024 * 1024; @@ -5031,7 +5084,7 @@ mod tests { options: BTreeMap::new(), params: Vec::new(), }; - let err = backup(&cfg, &repo_s, &pg_s).expect_err("backup requires a stanza"); + let err = backup(&cfg, &repo_s, &[(1, &repo_s)], &pg_s).expect_err("backup requires a stanza"); match err { CommandError::MissingOption { option } => assert_eq!(option, "stanza"), other => panic!("expected MissingOption, got {other:?}"), @@ -5220,7 +5273,7 @@ mod tests { params: Vec::new(), }; - backup(&cfg, &repo_s, &pg_s).expect("backup"); + backup(&cfg, &repo_s, &[(1, &repo_s)], &pg_s).expect("backup"); let info = InfoBackup::load(&repo_s, &backup_info_path("demo")).expect("reload backup.info"); let (label, _) = info.current.iter().next().expect("one backup"); @@ -5260,7 +5313,7 @@ mod tests { seed_file(&pg_s, "backup_label.old", b"obsolete label"); seed_file(&pg_s, "postmaster.opts", b"opts"); - backup(&typed_cfg("demo", "full"), &repo_s, &pg_s).expect("backup"); + backup(&typed_cfg("demo", "full"), &repo_s, &[(1, &repo_s)], &pg_s).expect("backup"); let info = InfoBackup::load(&repo_s, &backup_info_path("demo")).expect("reload backup.info"); let (label, _) = info.current.iter().next().expect("one backup"); @@ -5304,7 +5357,7 @@ mod tests { // The `logging_collector` pointer file (rotation state). seed_file(&pg_s, "current_logfiles", b"log/postgresql.log"); - backup(&typed_cfg("demo", "full"), &repo_s, &pg_s).expect("backup"); + backup(&typed_cfg("demo", "full"), &repo_s, &[(1, &repo_s)], &pg_s).expect("backup"); let info = InfoBackup::load(&repo_s, &backup_info_path("demo")).expect("reload backup.info"); let (label, _) = info.current.iter().next().expect("one backup"); @@ -5346,7 +5399,7 @@ mod tests { // A second excluded runtime dir, also with content. seed_file(&pg_s, "pg_subtrans/0000", b"transient subtrans slru"); - backup(&typed_cfg("demo", "full"), &repo_s, &pg_s).expect("backup"); + backup(&typed_cfg("demo", "full"), &repo_s, &[(1, &repo_s)], &pg_s).expect("backup"); let info = InfoBackup::load(&repo_s, &backup_info_path("demo")).expect("reload backup.info"); let (label, _) = info.current.iter().next().expect("one backup"); @@ -5497,7 +5550,7 @@ mod tests { let held = crate::lock::lock_acquire(lock_dir.path(), "demo", LockType::Backup).expect("pre-acquire backup lock"); assert!(expected_lock.exists(), "lock file must appear while held"); - let err = backup(&cfg, &repo_s, &pg_s).expect_err("backup must fail while the backup lock is held"); + let err = backup(&cfg, &repo_s, &[(1, &repo_s)], &pg_s).expect_err("backup must fail while the backup lock is held"); assert!( err.to_string().contains("another backup is running"), "unexpected error: {err}" @@ -5506,7 +5559,7 @@ mod tests { // Releasing the concurrent lock lets a backup run to completion; the // handle drops at return so the stale lock file is cleaned up. drop(held); - backup(&cfg, &repo_s, &pg_s).expect("backup succeeds once the lock is free"); + backup(&cfg, &repo_s, &[(1, &repo_s)], &pg_s).expect("backup succeeds once the lock is free"); assert!( !expected_lock.exists(), "lock file must be removed after the command releases it" @@ -5529,7 +5582,7 @@ mod tests { // Seed the stop file for the demo stanza. std::fs::write(lock_dir.path().join("demo.stop"), b"").expect("seed stop file"); - let err = backup(&cfg, &repo_s, &pg_s).expect_err("backup must refuse when stopped"); + let err = backup(&cfg, &repo_s, &[(1, &repo_s)], &pg_s).expect_err("backup must refuse when stopped"); let msg = err.to_string(); assert!(msg.contains("stop file exists for stanza demo"), "unexpected error: {msg}"); @@ -5869,7 +5922,7 @@ mod tests { init_stanza(&repo_s, "demo"); seed_cluster(&pg_s); - backup(&typed_cfg("demo", "full"), &repo_s, &pg_s).expect("full backup"); + backup(&typed_cfg("demo", "full"), &repo_s, &[(1, &repo_s)], &pg_s).expect("full backup"); // The full label is timestamp-derived; find the single backup recorded. let info = InfoBackup::load(&repo_s, &backup_info_path("demo")).expect("reload backup.info"); @@ -6269,7 +6322,7 @@ mod tests { seed_file(&pg_s, "base/1/1259", &relation); seed_file(&pg_s, "PG_VERSION", b"14\n"); - backup(&checksum_page_cfg("demo"), &repo_s, &pg_s).expect("backup"); + backup(&checksum_page_cfg("demo"), &repo_s, &[(1, &repo_s)], &pg_s).expect("backup"); let info = InfoBackup::load(&repo_s, &backup_info_path("demo")).expect("reload backup.info"); let (label, _) = info.current.iter().next().expect("one backup"); @@ -6299,7 +6352,7 @@ mod tests { relation[BLCKSZ + 8] ^= 0x01; seed_file(&pg_s, "base/1/1259", &relation); - backup(&checksum_page_cfg("demo"), &repo_s, &pg_s).expect("backup"); + backup(&checksum_page_cfg("demo"), &repo_s, &[(1, &repo_s)], &pg_s).expect("backup"); let info = InfoBackup::load(&repo_s, &backup_info_path("demo")).expect("reload backup.info"); let (label, _) = info.current.iter().next().expect("one backup"); @@ -6325,7 +6378,7 @@ mod tests { init_stanza(&repo_s, "demo"); seed_file(&pg_s, "base/1/1259", &valid_relation(2)); - backup(&typed_cfg("demo", "full"), &repo_s, &pg_s).expect("backup"); + backup(&typed_cfg("demo", "full"), &repo_s, &[(1, &repo_s)], &pg_s).expect("backup"); let info = InfoBackup::load(&repo_s, &backup_info_path("demo")).expect("reload backup.info"); let (label, _) = info.current.iter().next().expect("one backup"); @@ -6347,7 +6400,7 @@ mod tests { b"not page aligned content of arbitrary length here .....", ); - backup(&checksum_page_cfg("demo"), &repo_s, &pg_s).expect("backup"); + backup(&checksum_page_cfg("demo"), &repo_s, &[(1, &repo_s)], &pg_s).expect("backup"); let info = InfoBackup::load(&repo_s, &backup_info_path("demo")).expect("reload backup.info"); let (label, _) = info.current.iter().next().expect("one backup"); @@ -6367,7 +6420,7 @@ mod tests { init_stanza(&repo_s, "demo"); seed_file(&pg_s, "base/1/1259", &vec![0u8; BLCKSZ * 2]); - backup(&checksum_page_cfg("demo"), &repo_s, &pg_s).expect("backup"); + backup(&checksum_page_cfg("demo"), &repo_s, &[(1, &repo_s)], &pg_s).expect("backup"); let info = InfoBackup::load(&repo_s, &backup_info_path("demo")).expect("reload backup.info"); let (label, _) = info.current.iter().next().expect("one backup"); @@ -6515,7 +6568,7 @@ mod tests { relation[8] ^= 0x01; // bit-flip in block 0's stored pd_checksum seed_file(&pg_s, "base/1/16384", &relation); - backup(&default_full_cfg("demo"), &repo_s, &pg_s).expect("backup"); + backup(&default_full_cfg("demo"), &repo_s, &[(1, &repo_s)], &pg_s).expect("backup"); let captured = String::from_utf8(pgbr_core::log::capture::drain()).expect("captured bytes utf-8"); pgbr_core::log::capture::uninstall(); @@ -6563,7 +6616,7 @@ mod tests { seed_file(&pg_s, "global/pg_control", &synth_pg_control_v14(1)); seed_file(&pg_s, "base/1/16384", &valid_relation(2)); - backup(&default_full_cfg("demo"), &repo_s, &pg_s).expect("backup"); + backup(&default_full_cfg("demo"), &repo_s, &[(1, &repo_s)], &pg_s).expect("backup"); let captured = String::from_utf8(pgbr_core::log::capture::drain()).expect("captured bytes utf-8"); pgbr_core::log::capture::uninstall(); @@ -7504,7 +7557,7 @@ mod tests { seed_file(&pg_s, "base/1/1260", b"small relation two, slightly bigger"); seed_file(&pg_s, "PG_VERSION", b"14\n"); - backup(&bundle_cfg("demo", false, None), &repo_s, &pg_s).expect("bundled backup"); + backup(&bundle_cfg("demo", false, None), &repo_s, &[(1, &repo_s)], &pg_s).expect("bundled backup"); let info = InfoBackup::load(&repo_s, &backup_info_path("demo")).expect("reload backup.info"); let (label, _) = info.current.iter().next().expect("one backup"); @@ -7535,7 +7588,7 @@ mod tests { // Limit of 100 bytes: the 4096-byte file exceeds it and stays standalone; // PG_VERSION is bundled. repo-block off so the big file is not split. - backup(&bundle_cfg("demo", false, Some(100)), &repo_s, &pg_s).expect("bundled backup"); + backup(&bundle_cfg("demo", false, Some(100)), &repo_s, &[(1, &repo_s)], &pg_s).expect("bundled backup"); let info = InfoBackup::load(&repo_s, &backup_info_path("demo")).expect("reload backup.info"); let (label, _) = info.current.iter().next().expect("one backup"); @@ -7563,7 +7616,7 @@ mod tests { seed_file(&pg_s, "base/1/1259", &big); seed_file(&pg_s, "PG_VERSION", b"14\n"); - backup(&bundle_cfg("demo", true, None), &repo_s, &pg_s).expect("block backup"); + backup(&bundle_cfg("demo", true, None), &repo_s, &[(1, &repo_s)], &pg_s).expect("block backup"); let info = InfoBackup::load(&repo_s, &backup_info_path("demo")).expect("reload backup.info"); let (label, _) = info.current.iter().next().expect("one backup"); @@ -7641,7 +7694,7 @@ mod tests { seed_file(&pg_s, "base/1/1259", &relation); // checksum_page_cfg leaves page-header-check unset -> defaults true. - backup(&checksum_page_cfg("demo"), &repo_s, &pg_s).expect("backup"); + backup(&checksum_page_cfg("demo"), &repo_s, &[(1, &repo_s)], &pg_s).expect("backup"); let info = InfoBackup::load(&repo_s, &backup_info_path("demo")).expect("reload backup.info"); let (label, _) = info.current.iter().next().expect("one backup"); @@ -7666,7 +7719,7 @@ mod tests { let mut cfg = checksum_page_cfg("demo"); cfg.options .insert(("page-header-check".to_owned(), None), OptionValue::Boolean(false)); - backup(&cfg, &repo_s, &pg_s).expect("backup"); + backup(&cfg, &repo_s, &[(1, &repo_s)], &pg_s).expect("backup"); let info = InfoBackup::load(&repo_s, &backup_info_path("demo")).expect("reload backup.info"); let (label, _) = info.current.iter().next().expect("one backup"); @@ -8172,7 +8225,7 @@ mod tests { params: Vec::new(), }; - backup(&config, &repo_s, &pg_s).expect("backup with expire-auto"); + backup(&config, &repo_s, &[(1, &repo_s)], &pg_s).expect("backup with expire-auto"); // After expire-auto with retention=1, only the single newest full survives. let info = InfoBackup::load(&repo_s, &backup_info_path("demo")).expect("reload backup.info"); diff --git a/crates/pgbr-command/src/lib.rs b/crates/pgbr-command/src/lib.rs index b75f2743c..9fb695413 100644 --- a/crates/pgbr-command/src/lib.rs +++ b/crates/pgbr-command/src/lib.rs @@ -38,6 +38,7 @@ pub mod repo; pub mod restore; pub mod server; pub mod stanza; +pub mod sync; pub mod verify; pub mod worker; @@ -175,7 +176,8 @@ pub fn dispatch_multi( "repo-get" => repo::get(config, repo_storage), "repo-put" => repo::put(config, repo_storage), "repo-rm" => repo::rm(config, repo_storage), - "backup" => backup::backup(config, repo_storage, pg_storage), + "backup" => backup::backup(config, repo_storage, repo_storages, pg_storage), + "repo-sync" => sync::command(config, repo_storage, repo_storages), "restore" => restore::restore(config, repo_storage, pg_storage), "archive-get" => archive_get_with_prefetch(config, &archive_repos, pg_storage), "archive-push" => archive::push(config, &archive_repos, pg_storage), diff --git a/crates/pgbr-command/src/restore.rs b/crates/pgbr-command/src/restore.rs index 23cb6004b..716b7ebb9 100644 --- a/crates/pgbr-command/src/restore.rs +++ b/crates/pgbr-command/src/restore.rs @@ -5206,7 +5206,13 @@ mod tests { seed_pg_file(&pg_src_s, "base/1/1260", b); seed_pg_file(&pg_src_s, "base/1/1261", &big); - crate::backup::backup(&backup_cfg(stanza, "full", false, Some(100)), &repo_s, &pg_src_s).expect("bundled backup"); + crate::backup::backup( + &backup_cfg(stanza, "full", false, Some(100)), + &repo_s, + &[(1, &repo_s)], + &pg_src_s, + ) + .expect("bundled backup"); let label = latest_label(&repo_s, stanza); let outcome = restore_inner(&cfg(Some(stanza), Some(&label)), &repo_s, &pg_dst_s).expect("restore"); @@ -5236,7 +5242,7 @@ mod tests { let mut bcfg = backup_cfg(stanza, "full", false, None); bcfg.options .insert(("compress-type".to_owned(), None), OptionValue::StringId("gz".to_owned())); - crate::backup::backup(&bcfg, &repo_s, &pg_src_s).expect("bundled gz backup"); + crate::backup::backup(&bcfg, &repo_s, &[(1, &repo_s)], &pg_src_s).expect("bundled gz backup"); let label = latest_label(&repo_s, stanza); restore_inner(&cfg(Some(stanza), Some(&label)), &repo_s, &pg_dst_s).expect("restore"); @@ -5257,7 +5263,7 @@ mod tests { seed_pg_file(&pg_src_s, "PG_VERSION", b"14\n"); seed_pg_file(&pg_src_s, "base/1/1259", &big); - crate::backup::backup(&backup_cfg(stanza, "full", true, None), &repo_s, &pg_src_s).expect("block backup"); + crate::backup::backup(&backup_cfg(stanza, "full", true, None), &repo_s, &[(1, &repo_s)], &pg_src_s).expect("block backup"); let label = latest_label(&repo_s, stanza); restore_inner(&cfg(Some(stanza), Some(&label)), &repo_s, &pg_dst_s).expect("restore"); @@ -5281,7 +5287,8 @@ mod tests { let original: Vec = (0..300 * 1024u32).map(|n| (n % 251) as u8).collect(); seed_pg_file(&pg_src_s, "base/1/1259", &original); - crate::backup::backup(&backup_cfg(stanza, "full", true, None), &repo_s, &pg_src_s).expect("full block backup"); + crate::backup::backup(&backup_cfg(stanza, "full", true, None), &repo_s, &[(1, &repo_s)], &pg_src_s) + .expect("full block backup"); let full_label = latest_label(&repo_s, stanza); // Mutate the first 8 KiB only, then take a diff. `original` is not used @@ -5291,7 +5298,8 @@ mod tests { *byte = byte.wrapping_add(1); } seed_pg_file(&pg_src_s, "base/1/1259", &modified); - crate::backup::backup(&backup_cfg(stanza, "diff", true, None), &repo_s, &pg_src_s).expect("diff block backup"); + crate::backup::backup(&backup_cfg(stanza, "diff", true, None), &repo_s, &[(1, &repo_s)], &pg_src_s) + .expect("diff block backup"); let diff_label = latest_label(&repo_s, stanza); assert_ne!(diff_label, full_label, "diff produced a new label"); @@ -5469,7 +5477,13 @@ mod tests { seed_pg_file(&pg_src_s, "base/1/1260", b); seed_pg_file(&pg_src_s, "base/1/1261", &big); - crate::backup::backup(&backup_cfg(stanza, "full", false, Some(100)), &repo_inner, &pg_src_s).expect("bundled backup"); + crate::backup::backup( + &backup_cfg(stanza, "full", false, Some(100)), + &repo_inner, + &[(1, &repo_inner)], + &pg_src_s, + ) + .expect("bundled backup"); let label = latest_label(&repo_inner, stanza); let repo_s = RecordingRepo::new(repo_inner); @@ -5515,7 +5529,13 @@ mod tests { let original: Vec = (0..300 * 1024u32).map(|n| (n % 251) as u8).collect(); seed_pg_file(&pg_src_s, "base/1/1259", &original); - crate::backup::backup(&backup_cfg(stanza, "full", true, None), &repo_inner, &pg_src_s).expect("full block backup"); + crate::backup::backup( + &backup_cfg(stanza, "full", true, None), + &repo_inner, + &[(1, &repo_inner)], + &pg_src_s, + ) + .expect("full block backup"); let full_label = latest_label(&repo_inner, stanza); let mut modified = original; @@ -5523,7 +5543,13 @@ mod tests { *byte = byte.wrapping_add(1); } seed_pg_file(&pg_src_s, "base/1/1259", &modified); - crate::backup::backup(&backup_cfg(stanza, "diff", true, None), &repo_inner, &pg_src_s).expect("diff block backup"); + crate::backup::backup( + &backup_cfg(stanza, "diff", true, None), + &repo_inner, + &[(1, &repo_inner)], + &pg_src_s, + ) + .expect("diff block backup"); let diff_label = latest_label(&repo_inner, stanza); assert_ne!(diff_label, full_label, "diff produced a new label"); @@ -5574,12 +5600,12 @@ mod tests { // incremental can reference whole. let kept = b"a relation that does not change between the full and the incr".as_slice(); seed_pg_file(&pg_src_s, "base/1/1259", kept); - crate::backup::backup(&backup_cfg_plain(stanza, "full"), &repo_inner, &pg_src_s).expect("full backup"); + crate::backup::backup(&backup_cfg_plain(stanza, "full"), &repo_inner, &[(1, &repo_inner)], &pg_src_s).expect("full backup"); let full_label = latest_label(&repo_inner, stanza); // Take an incremental without touching the file: it should reference the // full for the unchanged file. - crate::backup::backup(&backup_cfg_plain(stanza, "incr"), &repo_inner, &pg_src_s).expect("incr backup"); + crate::backup::backup(&backup_cfg_plain(stanza, "incr"), &repo_inner, &[(1, &repo_inner)], &pg_src_s).expect("incr backup"); let incr_label = latest_label(&repo_inner, stanza); assert_ne!(incr_label, full_label, "incr produced a new label"); @@ -6331,7 +6357,13 @@ mod tests { for (rel, bytes) in files { seed_pg_file(&pg_src_s, rel, bytes); } - crate::backup::backup(&backup_cfg(stanza, "full", false, Some(1024)), &repo_inner, &pg_src_s).expect("bundled backup"); + crate::backup::backup( + &backup_cfg(stanza, "full", false, Some(1024)), + &repo_inner, + &[(1, &repo_inner)], + &pg_src_s, + ) + .expect("bundled backup"); let label = latest_label(&repo_inner, stanza); // Sanity: a single bundle object was produced. diff --git a/crates/pgbr-command/src/sync/archive.rs b/crates/pgbr-command/src/sync/archive.rs new file mode 100644 index 000000000..59e03cb50 --- /dev/null +++ b/crates/pgbr-command/src/sync/archive.rs @@ -0,0 +1,373 @@ +//! repo-sync WAL engine: byte-identical mirroring of archived WAL from the +//! active repository to one destination repository. +//! +//! Every object under `archive///` in the source — full and +//! partial WAL segments, backup-history files (`..backup`), and +//! timeline-history files (`.history`) — is enumerated and raw-copied to the +//! **same** path in the destination. Because repositories in a set are +//! byte-identical mirrors (global compression; identical bundling / block / +//! cipher sub-key), the copy is a pure [`pgbr_io::copy`] of the stored bytes: +//! no decompress, no decrypt, no recompute. The archive-id directory name +//! (`-`) is resolved from the source `archive.info` via +//! [`crate::archive::load_archive_info`]. +//! +//! Idempotent: a WAL segment already present in the destination in any stored +//! form (plaintext or a compression suffix) is skipped via +//! [`crate::archive::repo_has_segment`]; non-segment objects are skipped on a +//! plain [`Storage::exists`]. `archive.info` itself is never touched — it is +//! per-repo metadata seeded by `stanza-create`, and the destination is required +//! to be an already-initialised stanza sharing the source cipher sub-key. + +use std::path::PathBuf; + +use pgbr_config::LoadedConfig; +use pgbr_postgres::lsn::parse_wal_segment; +use pgbr_storage::Storage; + +use super::{SyncKind, SyncOutcome}; +use crate::CommandError; + +/// File extensions a stored WAL object may carry, in the same order the +/// `archive-get` path probes them. Stripped from a listed object's basename to +/// recover the bare segment name for the idempotent +/// [`crate::archive::repo_has_segment`] probe. Mirrors `archive.rs`'s private +/// `COMPRESS_SUFFIXES` (kept local so sync does not depend on that constant's +/// visibility). +const COMPRESS_SUFFIXES: &[&str] = &[".gz", ".zst", ".bz2", ".lz4"]; + +/// Mirror every WAL object for a stanza to one destination repository. +/// +/// Performs a raw byte copy of every WAL object under +/// `archive///` from the source repository to one +/// destination. The archive-id (`-`) comes from the source +/// `archive.info`. Idempotent: a segment already present in the destination (in +/// any stored form) is skipped, as is any non-segment object whose exact path +/// already exists. +/// +/// `src_index` / `dst_index` are the configured repository group indexes (used +/// only for log/diagnostic context here; the cipher sub-key is shared, so the +/// raw bytes are valid in either repository unchanged). +/// +/// # Errors +/// +/// Propagates storage and info-load failures. Returns an empty outcome when the +/// source has no `archive.info` (an uninitialised stanza, nothing to mirror). +pub fn sync_archive_to_repo( + config: &LoadedConfig, + stanza: &str, + src: &dyn Storage, + src_index: u32, + dst: &dyn Storage, + dst_index: u32, +) -> Result { + // The source/destination indexes are not needed for the raw copy itself + // (paths and bytes are identical across mirrors); they exist on the + // signature for symmetry with the backup engine and future diagnostics. + let _ = (src_index, dst_index); + + // Resolve the archive-id directory from the SOURCE archive.info. A + // one-element slice puts the source at position 0 so load_archive_info + // resolves the cipher passphrase at the first configured index (the active + // --repo, i.e. the source). No archive.info → nothing has been archived + // yet, so there is nothing to mirror. + let Some(info) = crate::archive::load_archive_info(config, &[src], stanza)? else { + return Ok(SyncOutcome { + kind: SyncKind::Wal, + items: 0, + bytes: 0, + }); + }; + let aid = crate::archive::archive_id(&info); + + // Enumerate every File under the archive-id directory. WAL lives one level + // deep in the flat layout, but recurse to be robust against any nested + // layout the storage backend reports. + let aid_dir = PathBuf::from(format!("archive/{stanza}/{aid}")); + let mut files = Vec::new(); + super::list_recursive(src, &aid_dir, &mut files)?; + + let mut items = 0usize; + let mut bytes = 0u64; + for entry in files { + let Some(name) = entry.path.file_name().and_then(|n| n.to_str()) else { + continue; + }; + // archive.info / archive.info.copy are per-repo metadata seeded by + // stanza-create; never mirror them. + if name == "archive.info" || name == "archive.info.copy" { + continue; + } + + // For a real WAL segment (24-hex base, possibly with a compression + // suffix) the destination may already hold a differently-suffixed form, + // so short-circuit with repo_has_segment (which probes plaintext + every + // compression suffix). For non-segment objects (.history, .backup) the + // stored form is unambiguous, so sync_one_segment's plain exists() skip + // is sufficient. + if let Some(base) = segment_base(name) + && crate::archive::repo_has_segment(dst, stanza, &aid, base)? + { + continue; + } + + let n = sync_one_segment(src, dst, stanza, &aid, name)?; + if n > 0 { + items += 1; + bytes += n; + } + } + + Ok(SyncOutcome { + kind: SyncKind::Wal, + items, + bytes, + }) +} + +/// The bare WAL-segment name for a stored object `name`, or `None` when `name` +/// is not a WAL segment (e.g. a `.history` or `.backup` file). +/// +/// A stored segment is a 24-hex name optionally followed by a compression +/// suffix (`000000010000000000000001`, `000000010000000000000001.gz`). Strips +/// any known compression suffix and returns the base when it parses as a WAL +/// segment; partial segments (`….partial`) and non-segment objects return +/// `None` so the caller falls back to an exact-path idempotency check. +fn segment_base(name: &str) -> Option<&str> { + let base = COMPRESS_SUFFIXES + .iter() + .find_map(|suffix| name.strip_suffix(suffix)) + .unwrap_or(name); + if parse_wal_segment(base).is_some() { Some(base) } else { None } +} + +/// Raw-copy one stored WAL object `name` (the basename, including any +/// compression suffix, e.g. `…01.gz`) from the source archive-id directory to +/// the same path in the destination, skipping if already present. Returns the +/// bytes copied (`0` on skip). +/// +/// The bytes are copied verbatim — no decompress / decrypt / recompute — because +/// the repositories are byte-identical mirrors. +/// +/// # Errors +/// +/// Propagates storage / I/O failures from the open / copy / create-path calls. +fn sync_one_segment(src: &dyn Storage, dst: &dyn Storage, stanza: &str, archive_id: &str, name: &str) -> Result { + let src_path = crate::archive::repo_segment_path(stanza, archive_id, name); + let dst_path = src_path.clone(); // identical layout across mirrors + if dst.exists(&dst_path)? { + return Ok(0); + } + if let Some(parent) = dst_path.parent() { + dst.create_path(parent, true)?; + } + let mut reader = src.open_read(&src_path)?; + let mut writer = dst.open_write(&dst_path)?; + let n = pgbr_io::copy(&mut reader, &mut writer)?; + writer.flush()?; + writer.close()?; + Ok(n) +} + +/// Storage-rooted path of a stored WAL object, mirroring +/// [`crate::archive::repo_segment_path`] for the test fixtures. +#[cfg(test)] +fn test_segment_path(stanza: &str, archive_id: &str, name: &str) -> PathBuf { + PathBuf::from(format!("archive/{stanza}/{archive_id}/{name}")) +} + +#[cfg(test)] +#[allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)] +mod tests { + use std::collections::BTreeMap; + use std::path::Path; + + use pgbr_config::{ConfigCommandRole, LoadedConfig}; + use pgbr_info::InfoArchive; + use pgbr_storage::{Posix, Storage}; + use tempfile::TempDir; + + use super::{SyncKind, sync_archive_to_repo, sync_one_segment, test_segment_path}; + + const STANZA: &str = "demo"; + const ARCHIVE_ID: &str = "16-1"; + const SEGMENT: &str = "000000010000000000000001"; + const SEGMENT_2: &str = "000000010000000000000002"; + const WAL_BODY: &[u8] = b"fake-wal-segment-contents"; + const WAL_BODY_2: &[u8] = b"another-segment-payload"; + + /// A `LoadedConfig` for the `repo-sync` command with no options set, so the + /// repositories resolve as the implicit single unencrypted repo at index 1. + fn config() -> LoadedConfig { + LoadedConfig { + command: "repo-sync".to_owned(), + command_role: ConfigCommandRole::Main, + stanza: Some(STANZA.to_owned()), + options: BTreeMap::new(), + params: Vec::new(), + } + } + + /// Seed an unencrypted `archive.info` for `STANZA` into `repo` so the + /// archive-id resolves to [`ARCHIVE_ID`] (`db_version` `"16"`, `db_id` `1`). + fn seed_archive_info(repo: &Posix) { + repo.create_path(Path::new(&format!("archive/{STANZA}")), true) + .expect("create archive dir"); + InfoArchive { + backrest_format: 5, + backrest_version: "2.58".to_owned(), + db_id: 1, + db_system_id: 6_873_049_345_984_568_091, + db_version: "16".to_owned(), + history: BTreeMap::new(), + } + .save(repo, Path::new(&format!("archive/{STANZA}/archive.info"))) + .expect("save archive.info"); + } + + /// Write `bytes` to a stored WAL object `name` under the source archive-id + /// directory. + fn put_segment(repo: &Posix, name: &str, bytes: &[u8]) { + let path = test_segment_path(STANZA, ARCHIVE_ID, name); + if let Some(parent) = path.parent() { + repo.create_path(parent, true).expect("create parent"); + } + let mut w = repo.open_write(&path).expect("open_write"); + w.write(bytes).expect("write"); + w.close().expect("close"); + } + + /// Read every byte of a stored WAL object `name` from `repo`, or `None` when + /// it is absent. + fn read_segment(repo: &Posix, name: &str) -> Option> { + let path = test_segment_path(STANZA, ARCHIVE_ID, name); + if !repo.exists(&path).expect("exists") { + return None; + } + let mut r = repo.open_read(&path).expect("open_read"); + Some(r.read_all().expect("read_all")) + } + + fn repo_pair() -> (TempDir, TempDir, Posix, Posix) { + let src = tempfile::tempdir().expect("src tempdir"); + let dst = tempfile::tempdir().expect("dst tempdir"); + let src_s = Posix::new(src.path()); + let dst_s = Posix::new(dst.path()); + (src, dst, src_s, dst_s) + } + + #[test] + fn copies_wal_segments_to_destination() { + let (_src, _dst, src_s, dst_s) = repo_pair(); + seed_archive_info(&src_s); + put_segment(&src_s, SEGMENT, WAL_BODY); + put_segment(&src_s, SEGMENT_2, WAL_BODY_2); + + let cfg = config(); + let outcome = sync_archive_to_repo(&cfg, STANZA, &src_s, 1, &dst_s, 2).expect("sync"); + + assert_eq!(outcome.kind, SyncKind::Wal); + assert_eq!(outcome.items, 2); + assert_eq!(outcome.bytes, (WAL_BODY.len() + WAL_BODY_2.len()) as u64); + assert_eq!(read_segment(&dst_s, SEGMENT).as_deref(), Some(WAL_BODY)); + assert_eq!(read_segment(&dst_s, SEGMENT_2).as_deref(), Some(WAL_BODY_2)); + } + + #[test] + fn rerun_is_idempotent_noop() { + let (_src, _dst, src_s, dst_s) = repo_pair(); + seed_archive_info(&src_s); + put_segment(&src_s, SEGMENT, WAL_BODY); + + let cfg = config(); + let first = sync_archive_to_repo(&cfg, STANZA, &src_s, 1, &dst_s, 2).expect("first sync"); + assert_eq!(first.items, 1); + assert_eq!(first.bytes, WAL_BODY.len() as u64); + + // Second run: everything is already present, so nothing is copied. + let second = sync_archive_to_repo(&cfg, STANZA, &src_s, 1, &dst_s, 2).expect("second sync"); + assert_eq!(second.items, 0); + assert_eq!(second.bytes, 0); + assert_eq!(read_segment(&dst_s, SEGMENT).as_deref(), Some(WAL_BODY)); + } + + #[test] + fn segment_already_in_destination_is_skipped() { + let (_src, _dst, src_s, dst_s) = repo_pair(); + seed_archive_info(&src_s); + put_segment(&src_s, SEGMENT, WAL_BODY); + put_segment(&src_s, SEGMENT_2, WAL_BODY_2); + // Destination already has SEGMENT (same bytes); only SEGMENT_2 should copy. + put_segment(&dst_s, SEGMENT, WAL_BODY); + + let cfg = config(); + let outcome = sync_archive_to_repo(&cfg, STANZA, &src_s, 1, &dst_s, 2).expect("sync"); + + assert_eq!(outcome.items, 1); + assert_eq!(outcome.bytes, WAL_BODY_2.len() as u64); + assert_eq!(read_segment(&dst_s, SEGMENT_2).as_deref(), Some(WAL_BODY_2)); + } + + #[test] + fn segment_present_under_compression_suffix_is_skipped() { + let (_src, _dst, src_s, dst_s) = repo_pair(); + seed_archive_info(&src_s); + put_segment(&src_s, SEGMENT, WAL_BODY); + // Destination holds the segment under a .gz suffix — repo_has_segment + // probes every suffix, so the plaintext source form must be skipped. + put_segment(&dst_s, &format!("{SEGMENT}.gz"), b"already-compressed"); + + let cfg = config(); + let outcome = sync_archive_to_repo(&cfg, STANZA, &src_s, 1, &dst_s, 2).expect("sync"); + + assert_eq!(outcome.items, 0); + assert_eq!(outcome.bytes, 0); + // The plaintext form was NOT written (the .gz form already counts). + assert_eq!(read_segment(&dst_s, SEGMENT), None); + } + + #[test] + fn history_and_backup_files_are_mirrored() { + let (_src, _dst, src_s, dst_s) = repo_pair(); + seed_archive_info(&src_s); + put_segment(&src_s, SEGMENT, WAL_BODY); + let history = "00000002.history"; + let backup_history = "000000010000000000000001.00000028.backup"; + put_segment(&src_s, history, b"timeline-history-body"); + put_segment(&src_s, backup_history, b"backup-history-body"); + + let cfg = config(); + let outcome = sync_archive_to_repo(&cfg, STANZA, &src_s, 1, &dst_s, 2).expect("sync"); + + assert_eq!(outcome.items, 3); + assert_eq!(read_segment(&dst_s, history).as_deref(), Some(&b"timeline-history-body"[..])); + assert_eq!( + read_segment(&dst_s, backup_history).as_deref(), + Some(&b"backup-history-body"[..]) + ); + } + + #[test] + fn no_archive_info_yields_empty_outcome() { + let (_src, _dst, src_s, dst_s) = repo_pair(); + // No archive.info seeded on the source: nothing has been archived. + let cfg = config(); + let outcome = sync_archive_to_repo(&cfg, STANZA, &src_s, 1, &dst_s, 2).expect("sync"); + assert_eq!(outcome.kind, SyncKind::Wal); + assert_eq!(outcome.items, 0); + assert_eq!(outcome.bytes, 0); + } + + #[test] + fn sync_one_segment_copies_then_skips() { + let (_src, _dst, src_s, dst_s) = repo_pair(); + put_segment(&src_s, SEGMENT, WAL_BODY); + + let copied = sync_one_segment(&src_s, &dst_s, STANZA, ARCHIVE_ID, SEGMENT).expect("first copy"); + assert_eq!(copied, WAL_BODY.len() as u64); + assert_eq!(read_segment(&dst_s, SEGMENT).as_deref(), Some(WAL_BODY)); + + // Already present → skip, zero bytes. + let again = sync_one_segment(&src_s, &dst_s, STANZA, ARCHIVE_ID, SEGMENT).expect("second copy"); + assert_eq!(again, 0); + } +} diff --git a/crates/pgbr-command/src/sync/backup.rs b/crates/pgbr-command/src/sync/backup.rs new file mode 100644 index 000000000..e1ba8c7a7 --- /dev/null +++ b/crates/pgbr-command/src/sync/backup.rs @@ -0,0 +1,785 @@ +//! repo-sync (Layer 1): byte-identical mirroring of a single backup (and any +//! ancestors it depends on) from the active repository to one destination. +//! +//! Repositories in a multi-repo set are byte-identical (compression global; +//! bundling / block-incremental / cipher-type validated identical; cipher +//! sub-key shared). Mirroring a backup is therefore a PURE RAW BYTE COPY of +//! every stored object under `backup//