diff --git a/CLAUDE.md b/CLAUDE.md index 5ad205459..c8468edf8 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -63,7 +63,7 @@ docker compose run --rm dev cargo test --workspace - `pgbr-protocol` — JSON-line `Request` / `Response` message types + `read_message`/`write_message` codec - `pgbr-postgres` — `crc32c_one`, `version` registry (PG 9.6 .. 18), `control` (`pg_control` header + per-version field offsets), `page` (`pg_checksum_page`) - `pgbr-info` — on-disk info files: `InfoArchive`, `InfoBackup`, `Manifest`, shared INI+SHA-1 `format` -- `pgbr-command` — every command + the dispatch entry (`dispatch_multi` is the real router, taking all configured repositories; `dispatch` is the single-repo convenience wrapper that calls it with `[(1, repo)]`): backup (full/diff/incr), restore (+ delta + reference resolution), archive-push/get, expire (backup + WAL retention), verify, check, info, stanza-create/delete/upgrade, repo-ls/get/put/rm, annotate, manifest, start/stop, server/server-ping (TCP + TLS), help, version; plus the shared `pipeline::RepoTransform` (compress + encrypt) +- `pgbr-command` — every command + the dispatch entry (`dispatch_multi` is the real router, taking all configured repositories; `dispatch` is the single-repo convenience wrapper that calls it with `[(1, repo)]`): backup (full/diff/incr), restore (+ delta + reference resolution), archive-push/get, expire (backup + WAL retention), verify, check, info, stanza-create/delete/upgrade, repo-ls/get/put/rm, annotate, manifest, start/stop, server/server-ping (TCP + TLS), repo-sync (repo-to-repo mirroring of backups + WAL), help, version; plus the shared `pipeline::RepoTransform` (compress + encrypt) - `pgbr-cli` — the `pgbackrust` binary: parse argv → load `config.yaml` + `pgbackrust.conf` → resolve → `pgbr_command::dispatch` ## Adding a configuration option @@ -102,7 +102,7 @@ Beyond the in-crate `cargo test` suite, `integration/` drives the **release bina - **Build the artifact first:** `./integration/build-binary.sh` compiles `pgbr-cli` in the **`build`** docker-compose service (Debian 12 bookworm, glibc 2.36 — *not* the trixie `dev`/`cargo` image) and drops a portable ELF at `integration/artifacts/pgbackrust`. glibc is backward- but not forward-compatible, so building on the lowest supported glibc yields **one** binary that runs on both the bookworm Vagrant VMs and the trixie Docker nodes. A trixie-built binary dies on the VMs with `GLIBC_2.39 not found`, so don't build the integration artifact with `dev`/`cargo`. - **Docker topology** (`integration/docker/docker-compose.yml`) — a 3-node cluster (`principal` primary / `secondaire` standby / `depot` repo) plus a `minio` S3 endpoint, sharing an SSH key so the `postgres` user can hop between nodes. The binary is **bind-mounted** into each node (not COPYed — under Docker Desktop's WSL2 image store the COPYed layer is intermittently not materialised). This is what local/CI validation uses. -- **Scenarios** (`integration/scenarios/`) — 11 numbered scripts (`01-local-minimal` … `11-s3`: local, remote-pull SSH, PITR, encryption, standby, tablespaces, async queuing, multi-repo, TLS, bundling/block, S3) over shared helpers in `_lib.sh`. Run them with `./integration/scenarios/run-all.sh` (optionally filter: `run-all.sh 01 03`). Each scenario runs against a freshly `down -v`'d topology for isolation; a failing scenario is recorded and the loop continues rather than aborting. +- **Scenarios** (`integration/scenarios/`) — 12 numbered scripts (`01-local-minimal` … `12-repo-sync`: local, remote-pull SSH, PITR, encryption, standby, tablespaces, async queuing, multi-repo, TLS, bundling/block, S3, repo-sync) over shared helpers in `_lib.sh`. Run them with `./integration/scenarios/run-all.sh` (optionally filter: `run-all.sh 01 03`). Each scenario runs against a freshly `down -v`'d topology for isolation; a failing scenario is recorded and the loop continues rather than aborting. - **Vagrant topology** (`integration/vagrant/`) — the same KB scenario matrix against VirtualBox VMs via `run-validation.sh` (run after `vagrant up`); an alternative to Docker that exercises real systemd/SSH. Provisioning scripts live in `integration/vagrant/provision/`. ## CI gating 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..e7f0f1d1e 100644 --- a/crates/pgbr-cli/src/lib.rs +++ b/crates/pgbr-cli/src/lib.rs @@ -328,11 +328,40 @@ fn worker_loaded_config(resolved: &ResolvedCli) -> LoadedConfig { const MULTI_REPO_COMMANDS: &[&str] = &[ "archive-push", "archive-get", + "repo-sync", "stanza-create", "stanza-delete", "stanza-upgrade", ]; +/// Whether `loaded` must build one backend per configured repository. +/// +/// The commands in [`MULTI_REPO_COMMANDS`] are unconditionally multi-repo. +/// `backup` is the one exception: it only needs every backend when the resolved +/// `repo-sync` boolean is on, because in that case a successful backup is +/// mirrored to the other repositories immediately afterwards (see +/// `pgbr_command::backup`'s `repo_sync_enabled`). With `repo-sync` off — the +/// default — building all backends is wasted and actively harmful: it would +/// connect every SFTP repo, spawn every `repoN-host` SSH worker and fail +/// construction on any partially configured cloud repo, so a plain `backup` +/// against a config with an extra unreachable repo would error before the +/// backup even starts. Constructing all backends is only justified when the +/// backup will mirror, so gate it on `repo-sync`. +fn needs_all_repo_storages(loaded: &LoadedConfig) -> bool { + MULTI_REPO_COMMANDS.contains(&loaded.command.as_str()) || (loaded.command == "backup" && repo_sync_enabled(loaded)) +} + +/// Whether the resolved `--repo-sync` boolean is enabled (default **false**). +/// +/// Mirrors `pgbr_command::backup`'s reader so the CLI and the command agree on +/// when a backup fans out to every repository. +fn repo_sync_enabled(loaded: &LoadedConfig) -> bool { + matches!( + loaded.options.get(&("repo-sync".to_owned(), None)), + Some(OptionValue::Boolean(true)) + ) +} + /// Build storage (as needed) and dispatch `loaded` to its command, mapping the /// outcome to a process exit code. /// @@ -360,7 +389,7 @@ fn dispatch_loaded(loaded: &LoadedConfig) -> Result { // `--repo`) is the one whose index matches `repo` — or the first configured // when the active index has no backend in the set (it always does, since // `configured_repo_indexes` includes the active index). - if MULTI_REPO_COMMANDS.contains(&loaded.command.as_str()) { + if needs_all_repo_storages(loaded) { let all = storage_helper::build_all_repo_storages(loaded)?; let active = storage_helper::active_repo_index(loaded); let active_storage = all @@ -1612,6 +1641,44 @@ option: assert_eq!(log_level_from_id("nonsense"), None); } + #[test] + fn backup_needs_all_repos_only_when_repo_sync_on() { + use super::needs_all_repo_storages; + + // Plain backup with `repo-sync` unset takes the single-repo path: it + // must NOT build every configured backend, so a config with an extra + // unreachable repo does not fail before the backup starts. + let plain = loaded("backup", Some("demo"), &[]); + assert!(!needs_all_repo_storages(&plain)); + + // Backup with `repo-sync` explicitly off is likewise single-repo. + let off = loaded("backup", Some("demo"), &[("repo-sync", None, OptionValue::Boolean(false))]); + assert!(!needs_all_repo_storages(&off)); + + // Backup with `repo-sync` on must fan out to every repository, since it + // mirrors the completed backup to the others. + let on = loaded("backup", Some("demo"), &[("repo-sync", None, OptionValue::Boolean(true))]); + assert!(needs_all_repo_storages(&on)); + } + + #[test] + fn declared_multi_repo_commands_are_always_multi_repo() { + use super::{MULTI_REPO_COMMANDS, needs_all_repo_storages}; + + // Every command in the unconditional list fans out regardless of + // `repo-sync`. `repo-sync` (the command) in particular is always + // multi-repo. + for command in MULTI_REPO_COMMANDS { + let loaded_cfg = loaded(command, Some("demo"), &[]); + assert!(needs_all_repo_storages(&loaded_cfg), "{command} must be multi-repo"); + } + + // A non-multi command with `repo-sync` on stays single-repo: the gate is + // scoped to `backup` alone, not "any command with repo-sync set". + let restore = loaded("restore", Some("demo"), &[("repo-sync", None, OptionValue::Boolean(true))]); + assert!(!needs_all_repo_storages(&restore)); + } + #[test] fn resolve_log_level_uses_value_then_fallback() { use pgbr_core::log::{LOG_LEVEL_DEBUG, LOG_LEVEL_INFO, LOG_LEVEL_WARN}; diff --git a/crates/pgbr-command/src/archive.rs b/crates/pgbr-command/src/archive.rs index af3fe7156..28fd5de7a 100644 --- a/crates/pgbr-command/src/archive.rs +++ b/crates/pgbr-command/src/archive.rs @@ -75,7 +75,11 @@ use crate::pipeline::{CompressType, RepoTransform}; /// once the plaintext form is found absent. Each maps to the compress codec /// that produced it. `pgbackrust` names archived WAL with the codec's /// extension, so a recovering client must try every suffix. -const COMPRESS_SUFFIXES: &[&str] = &[".gz", ".zst", ".bz2", ".lz4"]; +/// +/// Shared with the sync engine (`sync::archive`), which probes the same +/// suffix set when mirroring WAL between repositories; this is the single +/// source of truth so both paths stay in lockstep. +pub(crate) const COMPRESS_SUFFIXES: &[&str] = &[".gz", ".zst", ".bz2", ".lz4"]; /// Resolve the `compress-type` option to its file-name suffix. Returns /// `""` for `none` (or when the option is unset), and `.gz`/`.bz2`/`.lz4`/ @@ -288,7 +292,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 +300,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 +621,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 +1594,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..d15ae4b5c 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,10 +892,28 @@ 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. + // + // The mirror is a *post-success* side effect: the backup itself has already + // completed, so a mirror failure must not abort the command before retention + // runs. Capture the error WITHOUT `?`, always fall through to expire-auto, + // then surface the mirror error at the very end so monitoring still sees a + // non-zero exit while retention has already been applied. + let sync_error: Option = if repo_sync_enabled(config) && !policy.dry_run { + sync_to_other_repos(config, stanza, &outcome.label, repo_storage, repo_storages).err() + } else { + None + }; + // 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 // cmdExpire() at the end of a successful backup when expire-auto is set. + // + // Runs regardless of a repo-sync failure above so retention never silently + // stops while a mirror target is down. if expire_auto_enabled(config) && !policy.dry_run { log_info("expire-auto: applying retention"); let summary = crate::expire::expire_inner(config, repo_storage)?; @@ -900,7 +923,59 @@ pub fn backup(config: &LoadedConfig, repo_storage: &dyn Storage, pg_storage: &dy summary.kept_labels.len() )); } - Ok(()) + + // Surface a deferred repo-sync failure only after retention has been applied. + sync_error.map_or(Ok(()), Err) +} + +/// 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. +/// +/// Error policy: best-effort per target, mirroring the standalone `repo-sync` +/// command ([`crate::sync::run`]). A failure against one target is logged at +/// `WARN`, the first such error is remembered, and the remaining targets are +/// still attempted so one unreachable repository does not strand the others. +/// The first error is returned only after every target has been attempted; +/// `Ok(())` when all targets succeeded (or none were configured). +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() + )); + + // Best-effort: keep going past a single failure so one unreachable target + // does not strand the others, but remember the first error to fail the + // command afterwards. + let mut first_error: Option = None; + + for (dst_index, dst) in targets { + match crate::sync::backup::sync_backup_to_repo(config, stanza, label, repo_storage, src_index, dst, dst_index) { + Ok(synced) => log_info(&format!( + "repo-sync: repo{dst_index} <- backup {label}: {} object(s), {} byte(s)", + synced.items, synced.bytes + )), + Err(err) => { + log_warn(&format!("repo-sync: repo{dst_index} backup {label} sync failed: {err}")); + first_error.get_or_insert(err); + } + } + } + + first_error.map_or(Ok(()), Err) } /// The backup-control connections resolved per the `backup-standby` policy. @@ -1177,6 +1252,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 +5116,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 +5305,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 +5345,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 +5389,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 +5431,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 +5582,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 +5591,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 +5614,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 +5954,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 +6354,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 +6384,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 +6410,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 +6432,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 +6452,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 +6600,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 +6648,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 +7589,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 +7620,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 +7648,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 +7726,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 +7751,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 +8257,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"); @@ -9365,4 +9450,129 @@ mod tests { "two parallel runs must produce byte-identical bundle objects" ); } + + /// Build a `LoadedConfig` for the `backup` command from `(name, group-index)` + /// keyed option overrides (mirrors the sync module's `cfg` helper). + fn cfg(options: Vec<((&str, Option), OptionValue)>) -> LoadedConfig { + let mut map: BTreeMap<(String, Option), OptionValue> = BTreeMap::new(); + for ((name, idx), value) in options { + map.insert((name.to_owned(), idx), value); + } + LoadedConfig { + command: "backup".to_owned(), + command_role: pgbr_config::ConfigCommandRole::Main, + stanza: Some("demo".to_owned()), + options: map, + params: Vec::new(), + } + } + + /// Seed a source repository with an initialised stanza plus a single full + /// backup `label` (its `backup.info` entry and a `backup.manifest` with no + /// ancestor references) so `sync_to_other_repos` can mirror it verbatim. + fn seed_source_backup(repo: &Posix, stanza: &str, label: &str) { + init_stanza(repo, stanza); + let (mut info, _) = InfoBackup::load_keyed(repo, &backup_info_path(stanza), None).unwrap(); + info.current.insert( + label.to_owned(), + json!({ + "backup-info-size": 123, + "backup-label": label, + "backup-type": "full", + }), + ); + info.save_keyed(repo, &backup_info_path(stanza), None, None).unwrap(); + + let manifest = Manifest { + backup_label: label.to_owned(), + backup_type: "full".to_owned(), + timestamp_start: 1_704_110_400, + timestamp_stop: 1_704_110_410, + db_version: "14".to_owned(), + db_system_id: 6_873_049_345_984_568_091, + files: Vec::new(), + option_checksum_page: None, + paths: vec![ManifestPath { + path: "pg_data".to_owned(), + }], + links: Vec::new(), + }; + let manifest_rel = PathBuf::from(format!("backup/{stanza}/{label}/backup.manifest")); + repo.create_path(manifest_rel.parent().unwrap(), true).unwrap(); + manifest.save_keyed(repo, &manifest_rel, None).unwrap(); + } + + /// `sync_to_other_repos` is best-effort per target: a failing destination is + /// logged and skipped, the remaining destinations are still mirrored, and the + /// first error is returned only after every target has been attempted. This + /// guards the post-backup mirror against the fail-fast `?` regression where an + /// early target failure stranded the later ones (and pre-empted expire-auto). + #[test] + fn sync_to_other_repos_is_best_effort_across_targets() { + let stanza = "demo"; + let label = "20260101-100000F"; + + // Source (repo1) holds the backup to mirror. + let src_dir = tempfile::tempdir().unwrap(); + let src = Posix::new(src_dir.path()); + seed_source_backup(&src, stanza, label); + + // repo2 fails first (uninitialised stanza → no backup.info); repo3 is + // initialised and must still receive the backup despite repo2's failure. + let bad_dir = tempfile::tempdir().unwrap(); + let bad = Posix::new(bad_dir.path()); + let good_dir = tempfile::tempdir().unwrap(); + let good = Posix::new(good_dir.path()); + init_stanza(&good, stanza); + + let config = cfg(Vec::new()); + let storages: Vec<(u32, &dyn Storage)> = vec![(1, &src), (2, &bad), (3, &good)]; + + let err = sync_to_other_repos(&config, stanza, label, &src, &storages) + .expect_err("a failing target must surface after all targets are attempted"); + match err { + CommandError::Other(msg) => { + assert!(msg.contains("not initialized"), "first error should be repo2's: {msg}"); + } + other => panic!("expected Other(not initialized), got {other:?}"), + } + + // The good target (repo3) was still mirrored: its manifest sentinel exists. + assert!( + good.exists(Path::new(&format!("backup/{stanza}/{label}/backup.manifest"))) + .unwrap(), + "repo3 must be mirrored even though repo2 failed first" + ); + // repo3's backup.info now advertises the synced backup. + let (good_info, _) = InfoBackup::load_keyed(&good, &backup_info_path(stanza), None).unwrap(); + assert!( + good_info.current.contains_key(label), + "repo3 backup.info must list the synced label" + ); + } + + /// All targets healthy → `sync_to_other_repos` mirrors each and returns + /// `Ok(())`. + #[test] + fn sync_to_other_repos_ok_when_all_targets_succeed() { + let stanza = "demo"; + let label = "20260101-100000F"; + + let src_dir = tempfile::tempdir().unwrap(); + let src = Posix::new(src_dir.path()); + seed_source_backup(&src, stanza, label); + + let dst_dir = tempfile::tempdir().unwrap(); + let dst = Posix::new(dst_dir.path()); + init_stanza(&dst, stanza); + + let config = cfg(Vec::new()); + let storages: Vec<(u32, &dyn Storage)> = vec![(1, &src), (2, &dst)]; + + sync_to_other_repos(&config, stanza, label, &src, &storages).expect("all-healthy mirror is Ok"); + assert!( + dst.exists(Path::new(&format!("backup/{stanza}/{label}/backup.manifest"))) + .unwrap() + ); + } } 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..98f0b447e --- /dev/null +++ b/crates/pgbr-command/src/sync/archive.rs @@ -0,0 +1,489 @@ +//! 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`, loaded +//! and decrypted under the SOURCE repository's own cipher passphrase (keyed by +//! the active `--repo` index, not the lowest configured one). +//! +//! Idempotent: a WAL segment already present in the destination in any stored +//! form (plaintext or a compression suffix) is skipped; non-segment objects are +//! re-copied only when the destination copy is absent or torn (size mismatch). +//! `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. +//! +//! The destination archive-id directory is listed ONCE up front into a +//! basename → size map (and a set of stored segment bases). Every idempotency +//! decision is then answered from that snapshot rather than per-object +//! [`Storage::exists`] probes, so re-syncing a large archive that copies nothing +//! costs one destination listing instead of one signed HEAD (or several, across +//! compression suffixes) per source segment — decisive on S3 / Azure / GCS. + +use std::collections::BTreeSet; +use std::path::PathBuf; + +use pgbr_config::LoadedConfig; +use pgbr_info::InfoArchive; +use pgbr_postgres::lsn::parse_wal_segment; +use pgbr_storage::Storage; + +use super::{SyncKind, SyncOutcome}; +use crate::CommandError; + +/// 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`, loaded under the SOURCE repository's own cipher passphrase. +/// Idempotent: a segment already present in the destination (in any stored form) +/// is skipped; any other object is re-copied only when the destination copy is +/// absent or a torn (wrong-size) leftover. +/// +/// `src_index` is the configured repository group index of the source; it +/// selects the cipher passphrase the source `archive.info` is decrypted under. +/// This must be the ACTIVE `--repo`, not the lowest configured index — with two +/// encrypted repositories under different `repo-cipher-pass` values (a supported +/// configuration: only `cipher-type` and the shared sub-key are validated equal) +/// the wrong passphrase would fail to decrypt the source `archive.info`. +/// +/// # 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, +) -> Result { + // Resolve the archive-id directory from the SOURCE archive.info, decrypted + // under the source repository's own cipher passphrase (`None` when the source + // is unencrypted). Loading it via the source index — not a slice whose + // position 0 maps to the lowest configured index — is what makes this correct + // when repo1 and repo2 are encrypted under different user passphrases. No + // archive.info → nothing has been archived yet, so there is nothing to mirror. + let Some(info) = load_source_archive_info(config, src, src_index, 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)?; + + // List the DESTINATION archive-id directory ONCE and answer the segment + // idempotency decision from the snapshot: `dst_bases` holds every stored + // WAL-segment base (suffix stripped) so a segment present in ANY compressed + // form counts as already mirrored. This replaces the per-segment + // `repo_has_segment` HEAD storm — one listing versus up to five signed HEADs + // per source segment on the cloud backends. Non-segment objects fall through + // to `copy_object`, whose own size-checked probe both skips a complete copy + // and repairs a torn one. + let mut dst_files = Vec::new(); + super::list_recursive(dst, &aid_dir, &mut dst_files)?; + let mut dst_bases: BTreeSet = BTreeSet::new(); + for entry in &dst_files { + let Some(name) = entry.path.file_name().and_then(|n| n.to_str()) else { + continue; + }; + if let Some(base) = segment_base(name) { + dst_bases.insert(base.to_owned()); + } + } + + // Create the destination archive-id directory once, before the copy loop, + // rather than per object. `copy_object` re-creates parents defensively, but + // doing it here keeps that cost off every segment. + dst.create_path(&aid_dir, true)?; + + 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 the destination may already hold a + // differently-suffixed form, so any stored base counts as present — + // matching the old `repo_has_segment` semantics from the pre-listed set. + // Non-segment objects (.history, .backup) fall through to the size-checked + // copy below. + if let Some(base) = segment_base(name) + && dst_bases.contains(base) + { + continue; + } + + // Raw byte copy with torn-write repair: `copy_object` skips when the + // destination already holds the exact name at the source's size, and + // re-copies over a wrong-size (partial) leftover. The source size comes + // from the listing that enumerated the object. + if let Some(n) = super::copy_object(src, dst, &entry.path, entry.size)? { + items += 1; + bytes += n; + } + } + + Ok(SyncOutcome { + kind: SyncKind::Wal, + items, + bytes, + }) +} + +/// Load the SOURCE repository's `archive.info` for `stanza`, decrypted under its +/// own cipher passphrase (resolved from `src_index`; `None` for an unencrypted +/// repo). Returns `Ok(None)` when the source has no `archive.info` yet — an +/// uninitialised stanza with nothing to mirror. +/// +/// Mirrors [`crate::archive::load_archive_info`]'s per-repo cipher resolution but +/// keys off the explicit `src_index` (the active `--repo`) rather than slice +/// position, and returns `Option` instead of erroring when the file is absent. +/// +/// # Errors +/// +/// [`CommandError::MissingOption`] when the source is encrypted but has no +/// `repo-cipher-pass`; [`CommandError::Other`] when the file cannot be loaded / +/// decrypted; [`CommandError::Storage`] on an underlying storage error. +fn load_source_archive_info( + config: &LoadedConfig, + src: &dyn Storage, + src_index: u32, + stanza: &str, +) -> Result, CommandError> { + let info_path = PathBuf::from(format!("archive/{stanza}/archive.info")); + if !src.exists(&info_path)? { + return Ok(None); + } + let user_pass = crate::cipher::repo_user_pass(config, src_index)?; + let (info, _) = + InfoArchive::load_keyed(src, &info_path, user_pass.as_deref()).map_err(|err| CommandError::Other(err.to_string()))?; + Ok(Some(info)) +} + +/// 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 = crate::archive::COMPRESS_SUFFIXES + .iter() + .find_map(|suffix| name.strip_suffix(suffix)) + .unwrap_or(name); + if parse_wal_segment(base).is_some() { Some(base) } else { None } +} + +#[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}; + + 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 = crate::archive::repo_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 = crate::archive::repo_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).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).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).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).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 — the pre-listed base + // set records the stripped base, so the plaintext source form is 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).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).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).expect("sync"); + assert_eq!(outcome.kind, SyncKind::Wal); + assert_eq!(outcome.items, 0); + assert_eq!(outcome.bytes, 0); + } + + #[test] + fn truncated_destination_segment_is_recopied() { + // A prior sync was killed mid-write, leaving a short (torn) copy of a + // non-segment object at the exact destination path. `copy_object`'s + // size-checked probe must detect the wrong length and re-copy so the + // destination bytes equal the source. Use a non-segment name so the copy + // path (not the base-set skip) is exercised. + let (_src, _dst, src_s, dst_s) = repo_pair(); + seed_archive_info(&src_s); + let history = "00000002.history"; + put_segment(&src_s, history, b"full-history-body-contents"); + // Seed a short leftover at the same path in the destination. + put_segment(&dst_s, history, b"trunc"); + + let cfg = config(); + let outcome = sync_archive_to_repo(&cfg, STANZA, &src_s, 1, &dst_s).expect("sync"); + + // The torn object was repaired: exactly one object copied, dst == src. + assert_eq!(outcome.items, 1); + assert_eq!(outcome.bytes, b"full-history-body-contents".len() as u64); + assert_eq!( + read_segment(&dst_s, history).as_deref(), + Some(&b"full-history-body-contents"[..]) + ); + } + + /// A `LoadedConfig` whose repo1 and repo2 are BOTH `aes-256-cbc` but under + /// DIFFERENT user passphrases — a supported mirror config (only cipher-type + /// and the shared sub-key are validated equal). Used to prove the source + /// `archive.info` is decrypted under the active `--repo`'s own passphrase. + fn encrypted_config(repo1_pass: &str, repo2_pass: &str) -> LoadedConfig { + use pgbr_config::OptionValue; + + let mut options: BTreeMap<(String, Option), OptionValue> = BTreeMap::new(); + options.insert( + ("repo-cipher-type".to_owned(), Some(1)), + OptionValue::StringId("aes-256-cbc".to_owned()), + ); + options.insert( + ("repo-cipher-pass".to_owned(), Some(1)), + OptionValue::String(repo1_pass.to_owned()), + ); + options.insert( + ("repo-cipher-type".to_owned(), Some(2)), + OptionValue::StringId("aes-256-cbc".to_owned()), + ); + options.insert( + ("repo-cipher-pass".to_owned(), Some(2)), + OptionValue::String(repo2_pass.to_owned()), + ); + LoadedConfig { + command: "repo-sync".to_owned(), + command_role: ConfigCommandRole::Main, + stanza: Some(STANZA.to_owned()), + options, + params: Vec::new(), + } + } + + #[test] + fn encrypted_source_at_index_two_decrypts_under_its_own_pass() { + // Regression for the wrong-passphrase-index bug: with repo1 and repo2 + // encrypted under DIFFERENT user passphrases, a `repo-sync --repo=2` must + // decrypt the SOURCE archive.info under repo2's passphrase (src_index=2), + // not repo1's. Seeding archive.info encrypted under repo2's pass and + // syncing with src_index=2 must resolve and copy; a wrong-index load would + // fail to decrypt and error. + let repo1_pass = "1111111111111111111111111111111111111111111111111111111111111111"; + let repo2_pass = "2222222222222222222222222222222222222222222222222222222222222222"; + let sub_key = "3333333333333333333333333333333333333333333333333333333333333333"; + + let (_src, _dst, src_s, dst_s) = repo_pair(); + + // Seed the SOURCE archive.info encrypted under repo2's user passphrase. + src_s + .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_keyed( + &src_s, + Path::new(&format!("archive/{STANZA}/archive.info")), + Some(repo2_pass), + Some(sub_key), + ) + .expect("save encrypted archive.info"); + + // A stored (already repo-encrypted) WAL object; repo-sync copies its bytes + // verbatim, so the body need not be valid ciphertext for this test. + put_segment(&src_s, SEGMENT, WAL_BODY); + + let cfg = encrypted_config(repo1_pass, repo2_pass); + // src_index = 2 → the active --repo=2; archive.info must load under repo2's + // pass. Passing 1 here would decrypt with repo1's pass and error. + let outcome = sync_archive_to_repo(&cfg, STANZA, &src_s, 2, &dst_s).expect("sync must resolve under repo2's pass"); + + assert_eq!(outcome.items, 1); + assert_eq!(outcome.bytes, WAL_BODY.len() as u64); + assert_eq!(read_segment(&dst_s, SEGMENT).as_deref(), Some(WAL_BODY)); + } +} diff --git a/crates/pgbr-command/src/sync/backup.rs b/crates/pgbr-command/src/sync/backup.rs new file mode 100644 index 000000000..31cc29359 --- /dev/null +++ b/crates/pgbr-command/src/sync/backup.rs @@ -0,0 +1,947 @@ +//! 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//