diff --git a/.gitguardian.yaml b/.gitguardian.yaml index 7738362de..e1e4341d6 100644 --- a/.gitguardian.yaml +++ b/.gitguardian.yaml @@ -17,6 +17,14 @@ # (removed from the tree; still in git history) # # IMPORTANT: never add a *real* secret under these paths — it would be ignored. +# +# ignored-matches (below) target INLINE test fixtures that live in crate source +# files (crates/**/src/*.rs) alongside real code — those paths CANNOT be +# path-ignored without blinding the scanner to genuine secrets, so the exact +# synthetic values are ignored instead. Added 2026-07-04: fake cloud pagination +# cursors (S3 NextContinuationToken / Azure NextMarker) in the list() +# pagination unit tests; the live fixtures now use obviously-fake placeholders, +# these entries cover the original values still present in git history. version: 2 secret: @@ -25,3 +33,8 @@ secret: - "integration/**" - "test/certificate/**" - "doc/resource/fake-cert/**" + ignored-matches: + - name: "s3 list() test fixture: fake NextContinuationToken (pagination cursor, not a credential)" + match: "1ueGcxLPRx1Tr/XY7zn+1n" + - name: "azure list() test fixture: fake NextMarker (pagination cursor, not a credential)" + match: "2!72!MDAwMDI1IQ==" diff --git a/crates/pgbr-cli/src/remote_storage.rs b/crates/pgbr-cli/src/remote_storage.rs index f6cae2d3c..612706f9f 100644 --- a/crates/pgbr-cli/src/remote_storage.rs +++ b/crates/pgbr-cli/src/remote_storage.rs @@ -26,7 +26,7 @@ use std::io::{Read, Write}; use std::net::{Shutdown, TcpStream}; -use std::path::Path; +use std::path::{Path, PathBuf}; use std::process::{Child, ChildStdin, ChildStdout}; use std::sync::{Arc, Mutex}; @@ -163,6 +163,10 @@ impl Storage for RemoteProcessStorage { fn create_symlink(&self, link_path: &Path, target: &Path) -> Result<(), StorageError> { self.inner.create_symlink(link_path, target) } + + fn read_link(&self, path: &Path) -> Result { + self.inner.read_link(path) + } } /// A `Send`-able, cloneable handle to one client-side rustls stream. @@ -460,6 +464,10 @@ impl Storage for Remote fn create_symlink(&self, link_path: &Path, target: &Path) -> Result<(), StorageError> { self.inner().create_symlink(link_path, target) } + + fn read_link(&self, path: &Path) -> Result { + self.inner().read_link(path) + } } #[cfg(test)] diff --git a/crates/pgbr-command/src/archive.rs b/crates/pgbr-command/src/archive.rs index 28fd5de7a..1620a7c45 100644 --- a/crates/pgbr-command/src/archive.rs +++ b/crates/pgbr-command/src/archive.rs @@ -221,6 +221,24 @@ fn transform_segment(transform: &RepoTransform, bytes: &[u8]) -> Result, transform.apply_forward_keyed(bytes).map_err(CommandError::from) } +/// Crash-safe local write for the parallel drain/prefetch workers, which run on +/// absolute host paths outside the [`Storage`] trait. +/// +/// A thin adapter over the shared [`pgbr_storage::atomic_write_file`] primitive +/// (temp + fsync + atomic rename + parent-directory fsync) — the same guarantee +/// [`Storage::write_atomic_path`] gives the serial path, so a crashed worker +/// never leaves a torn WAL segment in the repository. The temp/rename mechanics +/// live in one place (`pgbr-storage`) so this path and the `Posix` backend can +/// never drift. +/// +/// # Errors +/// +/// Surfaces any [`pgbr_storage::StorageError`] from the underlying write (create, +/// `write_all`, `sync_all`, rename, or parent-directory fsync). +fn write_atomic_std(path: &Path, bytes: &[u8]) -> Result<(), pgbr_storage::StorageError> { + pgbr_storage::atomic_write_file(path, bytes) +} + /// Run `bytes` through `filter` (process + finish) and return the transformed /// output. fn run_filter(filter: &mut dyn Filter, bytes: &[u8]) -> Result, CommandError> { @@ -231,19 +249,97 @@ fn run_filter(filter: &mut dyn Filter, bytes: &[u8]) -> Result, CommandE } /// Write `bytes` to `dst_path` in `dst`, creating the destination's parent -/// directory first and flushing/closing the writer so the file is durable. +/// directory first and writing crash-safely. +/// +/// The write is routed through [`Storage::write_atomic_path`] rather than a raw +/// `open_write` + `write` + `close`: on a local (`Posix`/`Cifs`) backend that is +/// a temp-file + fsync + atomic-rename, so a crash or a concurrent reader can +/// never observe a torn / half-written WAL segment — the destination flips +/// atomically from absent to complete. On remote/object backends +/// `write_atomic_path` degrades to the strongest single-call guarantee the +/// backend offers. C ref: `archivePushFile()` writes WAL through a +/// temp-then-rename write in `src/command/archive/push/push.c`. fn write_segment(bytes: &[u8], dst: &dyn Storage, dst_path: &Path) -> Result<(), CommandError> { if let Some(parent) = dst_path.parent() { dst.create_path(parent, true)?; } - let mut writer = dst.open_write(dst_path)?; - writer.write(bytes)?; - writer.flush()?; - writer.close()?; + dst.write_atomic_path(dst_path, bytes)?; Ok(()) } +/// Outcome of the pre-write idempotency probe: whether a WAL segment already +/// present in the repository is byte-identical to the one about to be pushed. +enum PushIdempotency { + /// No file exists at the destination — write it. + Absent, + /// A file already exists and its plaintext content matches the segment + /// being pushed — nothing to do (an idempotent re-push). + Identical, + /// A file already exists but could NOT be decoded (torn / truncated / + /// corrupt — the residue of a crash mid-write). It is self-healed by + /// overwriting: an undecodable object carries no trustworthy content to + /// preserve, so re-writing the fresh segment is strictly safer than failing + /// the push forever. Distinct from a *decodable-but-different* object, which + /// is a genuine name collision and still a hard error. + Overwrite, +} + +/// Decide whether pushing the plaintext `plain` segment to `dst_path` in `dst` +/// is a no-op (the segment is already archived identically), a fresh write, or a +/// hard conflict. +/// +/// pgBackRust's `archive-push` is idempotent: `PostgreSQL` re-invokes the +/// archiver for the same segment after a crash / retry, so a segment that is +/// already present must succeed silently rather than be blindly overwritten — +/// but only when the stored content is the *same* WAL. A segment already present +/// with **different** (decodable) content is corruption (two different WAL +/// records claiming the same segment name) and must fail loudly, never be +/// clobbered. Because an encrypted repo stores non-deterministic ciphertext +/// (`Salted__` + random salt), the comparison is on the reversed *plaintext*, +/// not the raw stored bytes: the existing file is decoded through `transform` +/// before comparing to `plain`. C ref: `archivePushFile()` in +/// `src/command/archive/push/push.c`, which compares checksums and treats a +/// mismatch as an error. +/// +/// Self-heal: if the existing object cannot be *decoded* at all (torn / +/// truncated / corrupt ciphertext left by a crash mid-write, or a bad +/// decompress stream), it carries no trustworthy content to preserve. Rather +/// than fail the push forever against that undecodable residue, treat it as +/// [`PushIdempotency::Overwrite`] and re-write the fresh segment. The +/// decodable-but-different case is still a hard error — the distinction is +/// exactly *undecodable ⇒ overwrite* vs *decodable-but-different ⇒ error*. +/// +/// # Errors +/// +/// [`CommandError::Other`] when a file already exists whose plaintext decodes +/// but differs from `plain`. [`CommandError::Storage`] / [`CommandError::Io`] on +/// a storage read failure. +fn probe_push_idempotency( + dst: &dyn Storage, + dst_path: &Path, + plain: &[u8], + transform: &RepoTransform, + segment: &str, +) -> Result { + if !dst.exists(dst_path)? { + return Ok(PushIdempotency::Absent); + } + let stored = read_segment(dst, dst_path)?; + // Decode failure ⇒ the existing object is torn/corrupt (crash residue). + // Self-heal by overwriting rather than failing durably. + let Ok(existing) = transform.apply_reverse_keyed(&stored) else { + return Ok(PushIdempotency::Overwrite); + }; + if existing == plain { + Ok(PushIdempotency::Identical) + } else { + Err(CommandError::Other(format!( + "WAL segment already exists with different content: {segment}" + ))) + } +} + /// Write `bytes` to `dest_arg` exactly the way `PostgreSQL`'s `restore_command` /// contract expects: as a plain CLI path — absolute as-is, relative against /// the invoker's current working directory — NOT joined onto `pg1-path` or @@ -265,7 +361,8 @@ fn write_segment_to_dest_arg(bytes: &[u8], dest_arg: &Path) -> Result<(), Comman })? .join(dest_arg) }; - if let Some(parent) = resolved.parent() { + let parent = resolved.parent().filter(|p| !p.as_os_str().is_empty()); + if let Some(parent) = parent { // An empty parent (relative bare filename) is a no-op for create_dir_all, // which is the correct behaviour: the caller's cwd already exists. std::fs::create_dir_all(parent).map_err(|err| { @@ -275,12 +372,63 @@ fn write_segment_to_dest_arg(bytes: &[u8], dest_arg: &Path) -> Result<(), Comman )) })?; } - std::fs::write(&resolved, bytes).map_err(|err| { + + // Write crash-safely: bytes go to a sibling `.pgbackrust.tmp`, are + // fsync'd, then atomically renamed onto the destination, and finally the + // containing directory is fsync'd so the rename itself is durable. This + // mirrors the C `archive-get` writing the destination through a + // temp-then-rename write, so PostgreSQL's `restore_command` never sees a + // torn WAL segment after a crash. `std::fs::write` on its own leaves a + // half-written recovery segment observable to a restarting cluster. + let dir_for_sync = parent.map_or_else(|| PathBuf::from("."), Path::to_path_buf); + let tmp = { + let mut name = resolved + .file_name() + .map_or_else(|| std::ffi::OsString::from("pgbackrust"), std::ffi::OsString::from); + name.push(".pgbackrust.tmp"); + dir_for_sync.join(name) + }; + + // Write + fsync the temp file (scope the handle so it is closed before the + // rename on platforms that require it). + { + use std::io::Write as _; + let mut file = std::fs::File::create(&tmp).map_err(|err| { + CommandError::Other(format!( + "archive-get could not create temp destination '{}': {err}", + tmp.display() + )) + })?; + file.write_all(bytes).map_err(|err| { + CommandError::Other(format!( + "archive-get could not write temp destination '{}': {err}", + tmp.display() + )) + })?; + file.sync_all().map_err(|err| { + CommandError::Other(format!( + "archive-get could not fsync temp destination '{}': {err}", + tmp.display() + )) + })?; + } + + std::fs::rename(&tmp, &resolved).map_err(|err| { + // Best-effort cleanup so a failed rename does not litter a stray temp. + let _ = std::fs::remove_file(&tmp); CommandError::Other(format!( - "archive-get could not write destination '{}': {err}", + "archive-get could not rename temp '{}' onto destination '{}': {err}", + tmp.display(), resolved.display() )) })?; + + // fsync the destination directory so the rename survives a crash. A failure + // here is non-fatal on filesystems that reject directory fsync — the rename + // itself already succeeded. + if let Ok(dir) = std::fs::File::open(&dir_for_sync) { + let _ = dir.sync_all(); + } Ok(()) } @@ -604,6 +752,72 @@ fn is_checkable_wal_segment(name: &str) -> bool { parse_wal_segment(base).is_some() } +/// Validate the integrity of a WAL segment fetched from the repository *before* +/// it is written back toward `PostgreSQL`. +/// +/// `archive-get` hands the recovered bytes straight to a restoring cluster, so a +/// truncated / torn / wrong-length segment silently corrupts recovery. When the +/// recovered plaintext parses as a genuine WAL long-header page this re-parses it +/// and cross-checks: +/// +/// - a **full** segment's byte length must equal the header's `xlp_seg_size` +/// (the cluster's configured WAL segment size) — a short read / torn store is +/// rejected. Partial segments (`.partial`) are intentionally shorter, so their +/// length is not enforced; +/// - if the file name encodes a timeline it must match the header timeline (a +/// name/header timeline disagreement is corruption). +/// +/// The header re-parse also confirms the decrypt+decompress round-trip produced +/// coherent WAL rather than silently writing garbage from a mis-keyed decrypt. +/// Bytes that do not carry a WAL long header at all — non-segment companion +/// files (`.history`, `..backup`) `PostgreSQL` also archives — +/// are served verbatim (matching pgBackRust, which only header-checks real +/// segments); `PostgreSQL`'s own per-record WAL CRC is the last line of defence +/// on replay. C ref: `archiveGetFile()` re-reads the segment header in +/// `src/command/archive/get/get.c`. +/// +/// # Errors +/// +/// [`CommandError::Other`] on a length or timeline mismatch of a segment whose +/// header parsed. +fn check_fetched_segment(bytes: &[u8], segment: &str) -> Result<(), CommandError> { + // Only real WAL segments carry a long page header worth cross-checking; + // `.history` / `.backup` companion files are served verbatim. + if !is_checkable_wal_segment(segment) { + return Ok(()); + } + + // Cross-check whenever the bytes parse as a WAL long-header page. A payload + // that does not (e.g. a non-WAL fixture) is left to the caller / PostgreSQL + // CRC — this guard's job is to reject a *genuine* WAL segment that came back + // truncated or on the wrong timeline. + let Some(header) = pgbr_postgres::lsn::parse_wal_header(bytes) else { + return Ok(()); + }; + + // A full segment must be exactly `xlp_seg_size` bytes; a `.partial` segment + // is legitimately shorter and only timeline-checked. + let is_partial = segment.ends_with(".partial"); + if !is_partial && bytes.len() as u64 != u64::from(header.segment_size) { + return Err(CommandError::Other(format!( + "archive-get: fetched WAL segment {segment} is {} bytes but the header declares {} (truncated or torn store)", + bytes.len(), + header.segment_size + ))); + } + + if let Some((name_timeline, _, _)) = parse_wal_segment(segment.strip_suffix(".partial").unwrap_or(segment)) + && name_timeline != header.timeline + { + return Err(CommandError::Other(format!( + "archive-get: fetched WAL segment {segment} name timeline {name_timeline} does not match header timeline {}", + header.timeline + ))); + } + + Ok(()) +} + /// Load the stanza's `archive.info` from the first repository that has it, /// decrypting it with that repository's cipher passphrase when the repo is /// encrypted. Used to resolve the archive-id directory (`-`) @@ -838,6 +1052,15 @@ pub fn push(config: &LoadedConfig, repo_storages: &[&dyn Storage], pg_storage: & let transforms = per_repo_transforms(config, repo_storages, stanza)?; let dest = repo_segment_path(stanza, &archive_id, &format!("{segment}{}", compress_suffix(config))); for (repo, transform) in repo_storages.iter().zip(transforms.iter()) { + // Idempotency guard: a crash/retry may re-push a segment already in the + // repo. Identical content ⇒ success without re-writing; decodable-but- + // different content ⇒ hard error (never clobber a distinct WAL under the + // same name); undecodable (torn/corrupt) content ⇒ overwrite to + // self-heal a crash-time residue. Mirrors upstream archivePushFile. + match probe_push_idempotency(*repo, &dest, &bytes, transform, segment)? { + PushIdempotency::Identical => continue, + PushIdempotency::Absent | PushIdempotency::Overwrite => {} + } let stored = transform_segment(transform, &bytes)?; write_segment(&stored, *repo, &dest)?; } @@ -979,8 +1202,11 @@ pub fn drain_push_spool( match drain_one(spool, repo_storage, stanza, &archive_id, &segment, suffix, &staged, transform) { Ok(()) => { - spool.remove(&staged, false)?; + // `.ok` before staged removal — the segment is already in the + // repo, so a crash between the two must not drop the handshake. + // See the ordering rationale in `drain_out_spool`. write_segment(b"", spool, &status_ok_path(stanza, &segment))?; + spool.remove(&staged, false)?; drained += 1; } Err(err) => { @@ -1011,6 +1237,22 @@ fn drain_one( None => bytes, }; let dest = repo_segment_path(stanza, archive_id, &format!("{segment}{suffix}")); + // Idempotency guard: a crash between the repo write and the staged-copy + // removal re-runs this segment on the next drain. This legacy drain applies + // a deterministic compress-only transform (no cipher), so the re-produced + // `stored` bytes are byte-identical for the same input — compare them to + // whatever is already at `dest`: equal ⇒ nothing to do; different ⇒ a + // distinct WAL under the same name, which is a hard error. Mirrors upstream + // archivePushFile. + if repo_storage.exists(&dest)? { + let existing = read_segment(repo_storage, &dest)?; + if existing == stored { + return Ok(()); + } + return Err(CommandError::Other(format!( + "WAL segment already exists with different content: {segment}" + ))); + } write_segment(&stored, repo_storage, &dest) } @@ -1178,8 +1420,18 @@ fn drain_out_spool(spool: &dyn Storage, stanza: &str, targets: &[DrainTarget], p match drain_one_to_targets(spool, stanza, &segment, &staged, targets, process_max) { Ok(()) => { - spool.remove(&staged, false)?; + // Write the `.ok` status BEFORE removing the staged copy. At + // this point the segment is durably in every repository, so a + // crash between the two steps must not leave a state with + // neither the staged copy nor the `.ok` — the foreground call + // would then re-stage and re-drain (harmless now that the drain + // is idempotent) but, worse, a crash after removing the staged + // copy but before the `.ok` would lose the handshake entirely. + // Ordering `.ok` first closes that window: the segment is + // already archived, so the `.ok` is truthful, and the staged + // copy is then cleaned up. write_segment(b"", spool, &status_ok_path(stanza, &segment))?; + spool.remove(&staged, false)?; drained += 1; } Err(err) => { @@ -1264,9 +1516,27 @@ fn drain_out_spool_parallel( .ok_or_else(|| "drain-segment: missing segment name".to_owned())?; let bytes = std::fs::read(staged_abs).map_err(|err| format!("read {staged_abs}: {err}"))?; for (transform, dir) in closure_transforms.iter().zip(closure_target_dirs.iter()) { - let stored = transform.apply_forward_keyed(&bytes).map_err(|err| err.to_string())?; let dest = dir.join(format!("{segment}{}", transform.repo_suffix())); - std::fs::write(&dest, &stored).map_err(|err| format!("write {}: {err}", dest.display()))?; + // Idempotency guard (parallel path): if the segment is already at + // `dest`, compare on the reversed plaintext — identical ⇒ skip the + // re-write, decodable-but-different ⇒ a hard error, undecodable + // (torn/corrupt crash residue) ⇒ fall through and overwrite to + // self-heal. Then write crash-safely. + if let Ok(existing_stored) = std::fs::read(&dest) { + match transform.apply_reverse_keyed(&existing_stored) { + Ok(existing) if existing == bytes => continue, + Ok(_) => { + return Err(format!( + "WAL segment already exists with different content: {}", + dest.display() + )); + } + // Undecodable existing object: self-heal by overwriting. + Err(_) => {} + } + } + let stored = transform.apply_forward_keyed(&bytes).map_err(|err| err.to_string())?; + write_atomic_std(&dest, &stored).map_err(|err| format!("write {}: {err}", dest.display()))?; } Ok(Response::Ok(OkResponse { out: None })) }); @@ -1281,8 +1551,11 @@ fn drain_out_spool_parallel( let staged = out_dir.join(&jr.key); match jr.result { Ok(_) => { - spool.remove(&staged, false)?; + // `.ok` before staged removal — the segment is already in every + // repo, so a crash between the two must not drop the handshake. + // See the ordering rationale in `drain_out_spool`. write_segment(b"", spool, &status_ok_path(stanza, &jr.key))?; + spool.remove(&staged, false)?; drained += 1; } Err(message) => { @@ -1331,12 +1604,21 @@ fn drain_one_to_targets( } for target in targets { - let stored = transform_segment(target.transform, &bytes)?; let dest = repo_segment_path( stanza, target.archive_id, &format!("{segment}{}", target.transform.repo_suffix()), ); + // Idempotency guard: a crash mid-drain re-runs this fan-out. If the + // segment is already in this target with identical plaintext, skip it; + // if it decodes but differs, fail loudly rather than clobber; if it is + // undecodable (torn/corrupt crash residue), overwrite to self-heal. + // Mirrors upstream archivePushFile. + match probe_push_idempotency(target.repo, &dest, &bytes, target.transform, segment)? { + PushIdempotency::Identical => continue, + PushIdempotency::Absent | PushIdempotency::Overwrite => {} + } + let stored = transform_segment(target.transform, &bytes)?; write_segment(&stored, target.repo, &dest)?; } Ok(()) @@ -1404,8 +1686,26 @@ fn drain_one_to_targets_parallel( let transform = closure_transforms .get(idx_usize) .ok_or_else(|| format!("drain-target: no transform for index {idx_usize}"))?; + // Idempotency guard (parallel path): a crash mid-drain re-runs this + // fan-out. If the destination already holds this segment, compare on + // the reversed plaintext — identical ⇒ skip the re-write, decodable-but- + // different ⇒ a distinct WAL under the same name, a hard error; + // undecodable (torn/corrupt crash residue) ⇒ fall through and overwrite + // to self-heal. Mirrors the serial path's `probe_push_idempotency`. + if let Ok(existing_stored) = std::fs::read(dest_abs) { + match transform.apply_reverse_keyed(&existing_stored) { + Ok(existing) if existing == *closure_bytes => { + return Ok(Response::Ok(OkResponse { out: None })); + } + Ok(_) => { + return Err(format!("WAL segment already exists with different content: {dest_abs}")); + } + // Undecodable existing object: self-heal by overwriting. + Err(_) => {} + } + } let stored = transform.apply_forward_keyed(&closure_bytes).map_err(|err| err.to_string())?; - std::fs::write(dest_abs, &stored).map_err(|err| format!("write {dest_abs}: {err}"))?; + write_atomic_std(Path::new(dest_abs), &stored).map_err(|err| format!("write {dest_abs}: {err}"))?; Ok(Response::Ok(OkResponse { out: None })) }); @@ -1650,6 +1950,11 @@ fn fetch_from_repo( let stored = read_segment(repo_storage, &source)?; let bytes = decode_stored_segment(&stored, suffix, sub_key)?; + // Integrity gate: re-parse the WAL header and verify the segment length + // before handing the bytes to a restoring PostgreSQL — a torn store or a + // mis-decrypt must fail here rather than silently corrupt recovery. + check_fetched_segment(&bytes, segment)?; + write_segment_to_dest_arg(&bytes, dest) } @@ -1931,7 +2236,7 @@ fn prefetch_get_spool_parallel( // exists (the main thread `create_path`d it) so `std::fs::write` is // safe with no further mkdir. let dest = spool_dir.join(segment); - std::fs::write(&dest, &bytes).map_err(|err| format!("write {}: {err}", dest.display()))?; + write_atomic_std(&dest, &bytes).map_err(|err| format!("write {}: {err}", dest.display()))?; Ok(Response::Ok(OkResponse { out: Some(json!({ "prefetched": true })), @@ -2038,7 +2343,31 @@ pub(crate) fn read_archived_segment( }; let stored = read_segment(repo, &source)?; - Ok(Some(decode_stored_segment(&stored, suffix, sub_key)?)) + let bytes = decode_stored_segment(&stored, suffix, sub_key)?; + + // Lenient integrity gate: when the recovered bytes parse as a genuine WAL + // long-header page, cross-check the declared segment size so a + // torn store or a mis-decrypt of real WAL is caught before the caller + // (backup archive-copy) trusts it. Bytes that do not carry a WAL header + // (non-segment companion files, or the transform-round-trip test fixtures) + // are returned unchecked — the caller verifies those against the manifest + // checksum. This mirrors the stricter `check_fetched_segment` on the + // archive-get write path, minus the "must be WAL" requirement this + // read-only sibling cannot assume. + if is_checkable_wal_segment(segment) + && let Some(header) = pgbr_postgres::lsn::parse_wal_header(&bytes) + { + let is_partial = segment.ends_with(".partial"); + if !is_partial && bytes.len() as u64 != u64::from(header.segment_size) { + return Err(CommandError::Other(format!( + "archived WAL segment {segment} is {} bytes but its header declares {} (truncated or torn store)", + bytes.len(), + header.segment_size + ))); + } + } + + Ok(Some(bytes)) } #[cfg(test)] @@ -2054,9 +2383,9 @@ mod tests { use tempfile::TempDir; use super::{ - CommandError, check_wal_header, drain_push_spool, drain_push_spool_keyed, drain_push_spool_multi, fetch_segment_with_retry, - get, get_in_dir, per_repo_transforms, prefetch_get_spool, push, push_out_dir, push_queue_exceeded, read_archived_segment, - status_error_path, status_ok_path, wal_backlog_bytes, + CommandError, check_fetched_segment, check_wal_header, drain_push_spool, drain_push_spool_keyed, drain_push_spool_multi, + fetch_segment_with_retry, get, get_in_dir, per_repo_transforms, prefetch_get_spool, push, push_out_dir, + push_queue_exceeded, read_archived_segment, status_error_path, status_ok_path, wal_backlog_bytes, }; use crate::pipeline::{CompressType, RepoTransform}; use pgbr_info::InfoArchive; @@ -3374,6 +3703,128 @@ mod tests { assert_eq!(bytes, None, "a segment not in the archive yields None"); } + // ----------------------------------------------------------------------- + // archive-get integrity gate (check_fetched_segment) + // ----------------------------------------------------------------------- + + #[test] + fn check_fetched_segment_accepts_full_segment_matching_header() { + // 64-byte "segment" whose header declares its own length (64) and + // timeline 1 — matches SEGMENT's name timeline, so it passes. + let bytes = wal_segment_bytes(PG14_WAL_MAGIC, 1, TEST_SYSTEM_ID, 64); + check_fetched_segment(&bytes, SEGMENT).expect("a segment matching its header is accepted"); + } + + #[test] + fn check_fetched_segment_rejects_truncated_full_segment() { + // Header declares a 16 MiB segment but the fetched bytes are only 64 — + // a torn/truncated store must be rejected before it reaches PG. + let bytes = wal_segment_bytes(PG14_WAL_MAGIC, 1, TEST_SYSTEM_ID, 16 * 1024 * 1024); + let err = check_fetched_segment(&bytes, SEGMENT).expect_err("truncated segment must be rejected"); + assert!(matches!(err, CommandError::Other(msg) if msg.contains("truncated or torn store"))); + } + + #[test] + fn check_fetched_segment_rejects_wrong_timeline() { + // Header timeline 9 disagrees with SEGMENT's name timeline (1). + let bytes = wal_segment_bytes(PG14_WAL_MAGIC, 9, TEST_SYSTEM_ID, 64); + let err = check_fetched_segment(&bytes, SEGMENT).expect_err("timeline mismatch must be rejected"); + assert!(matches!(err, CommandError::Other(msg) if msg.contains("timeline"))); + } + + #[test] + fn check_fetched_segment_passes_headerless_payload() { + // Bytes without a WAL long header (a non-WAL fixture) are served + // verbatim — the gate only cross-checks genuine WAL. + check_fetched_segment(WAL_BODY, SEGMENT).expect("headerless payload is not rejected here"); + } + + #[test] + fn check_fetched_segment_skips_history_file() { + // A `.history` companion file carries no WAL header and is served as-is. + check_fetched_segment(b"1\t0/0\tno-header", "00000002.history").expect("history file is not header-checked"); + } + + // ----------------------------------------------------------------------- + // archive-push idempotency (probe_push_idempotency via push) + // ----------------------------------------------------------------------- + + #[test] + fn archive_push_is_idempotent_for_identical_segment() { + let (_repo, _pg, repo_s, pg_s) = posix_pair(); + seed_archive_info_generic(&repo_s, "demo"); + let wal_source = format!("pg_wal/{SEGMENT}"); + put(&pg_s, &wal_source, WAL_BODY); + + let mut cfg = fake_config(Some("demo"), vec![wal_source]); + cfg.options + .insert(("archive-header-check".to_owned(), None), OptionValue::Boolean(false)); + + // First push writes it; a second identical push is an idempotent no-op + // (a crash/retry re-invocation from PostgreSQL), not an error. + push(&cfg, &[&repo_s as &dyn Storage], &pg_s).expect("first push succeeds"); + push(&cfg, &[&repo_s as &dyn Storage], &pg_s).expect("identical re-push is idempotent"); + + let dest = format!("archive/demo/{ARCHIVE_ID}/{SEGMENT}"); + assert_eq!(read(&repo_s, &dest), WAL_BODY, "repo copy unchanged after re-push"); + } + + #[test] + fn archive_push_conflicts_on_different_content() { + let (_repo, _pg, repo_s, pg_s) = posix_pair(); + seed_archive_info_generic(&repo_s, "demo"); + let wal_source = format!("pg_wal/{SEGMENT}"); + put(&pg_s, &wal_source, WAL_BODY); + + let mut cfg = fake_config(Some("demo"), vec![wal_source]); + cfg.options + .insert(("archive-header-check".to_owned(), None), OptionValue::Boolean(false)); + push(&cfg, &[&repo_s as &dyn Storage], &pg_s).expect("first push succeeds"); + + // A DIFFERENT segment already stored under the same name is corruption: + // the re-push must fail loudly rather than clobber it. + let dest = format!("archive/demo/{ARCHIVE_ID}/{SEGMENT}"); + put(&repo_s, &dest, b"a-completely-different-wal-segment"); + let err = push(&cfg, &[&repo_s as &dyn Storage], &pg_s).expect_err("conflicting re-push must error"); + assert!(matches!(err, CommandError::Other(msg) if msg.contains("already exists with different content"))); + } + + #[test] + fn archive_push_self_heals_undecodable_existing_object() { + // A crash mid-write can leave a torn / corrupt object at the + // destination that no longer decodes. The idempotency probe must treat + // an UNDECODABLE existing object as absent and overwrite it (self-heal) + // rather than fail the push forever. A `.gz` repo makes the distinction + // sharp: garbage bytes at `.gz` are not a valid gzip stream, so + // the reverse transform errors — which must resolve to an overwrite, not + // an "already exists with different content" error. + let (_repo, _pg, repo_s, pg_s) = posix_pair(); + seed_archive_info_generic(&repo_s, "demo"); + let wal_source = format!("pg_wal/{SEGMENT}"); + put(&pg_s, &wal_source, WAL_BODY); + + let mut cfg = fake_config_compress(Some("demo"), vec![wal_source], "gz"); + cfg.options + .insert(("archive-header-check".to_owned(), None), OptionValue::Boolean(false)); + + // Seed the destination with undecodable garbage (not a gzip stream), + // simulating a torn object left by a crash mid-write. + let dest = format!("archive/demo/{ARCHIVE_ID}/{SEGMENT}.gz"); + put(&repo_s, &dest, b"\x00\x01\x02-not-a-gzip-stream-torn-residue"); + + // The push must succeed by overwriting the torn object rather than + // erroring on "different content". + push(&cfg, &[&repo_s as &dyn Storage], &pg_s).expect("push self-heals a torn existing object"); + + // The stored object now decodes back to the fresh WAL body. + let stored = read(&repo_s, &dest); + let transform = RepoTransform::with_key(CompressType::Gz, 3, None); + let recovered = transform + .apply_reverse_keyed(&stored) + .expect("healed object decodes as valid gzip"); + assert_eq!(recovered, WAL_BODY, "torn object was overwritten with the fresh segment"); + } + // ----------------------------------------------------------------------- // archive-push-queue-max // ----------------------------------------------------------------------- diff --git a/crates/pgbr-command/src/backup.rs b/crates/pgbr-command/src/backup.rs index d15ae4b5c..3c8bd7e39 100644 --- a/crates/pgbr-command/src/backup.rs +++ b/crates/pgbr-command/src/backup.rs @@ -46,11 +46,10 @@ //! first segment (everything before the first `_`). Its `backup.info` entry //! records `backup-type: "incr"` plus `backup-reference: []`. //! -//! Deliberately out of scope for this slice (follow-ups): -//! -//! - **Symlink target resolution.** The `Storage` trait has no link-target -//! accessor yet, so [`ManifestLink`] entries are recorded with an empty -//! `destination`. See the `// TODO: resolve link target` note in [`walk`]. +//! Symlink targets are resolved via [`Storage::read_link`]; on filesystem +//! backends each [`ManifestLink`] records the link's real `destination`. Backends +//! without link-target support (remote/object stores) still record an empty +//! destination, which remains a follow-up for remote tablespace backups. //! //! The real work lives in [`backup_inner_typed`], which takes the backup type, //! label, and start timestamp as parameters so tests can pin them; @@ -69,7 +68,7 @@ use pgbr_postgres::control::read_pg_control_data; use pgbr_postgres::lsn::{lsn_text_to_wal_segment, parse_lsn, wal_segment_range}; use pgbr_protocol::message::{OkResponse, Request, Response}; use pgbr_protocol::parallel::{Job, ParallelExecutor}; -use pgbr_storage::{Storage, StorageInfo, StorageKind}; +use pgbr_storage::{Storage, StorageError, StorageInfo, StorageKind}; use serde_json::json; use crate::CommandError; @@ -527,6 +526,52 @@ fn is_user_excluded(rel_path: &str, excludes: &[String]) -> bool { /// validated one `PAGE_SIZE` slice at a time. const PAGE_SIZE: usize = pgbr_postgres::page::BLCKSZ; +/// Number of pages in one relation segment file. +/// +/// `PostgreSQL` splits a relation into 1 GiB segments (`RELSEG_SIZE` pages of +/// `BLCKSZ` bytes each — `1 GiB / 8 KiB = 131072` for the default build). The +/// first segment is `` (block numbers `0 .. RELSEG_SIZE`), the second is +/// `.1` (block numbers `RELSEG_SIZE .. 2*RELSEG_SIZE`), and so on. The +/// `pd_checksum` a page carries is computed over the page's **absolute** block +/// number, so validating a `.k` segment requires offsetting the per-file +/// page index by `k * RELSEG_SIZE`. +/// +/// Computed entirely in `u32` arithmetic (no narrowing cast): `1 GiB` +/// (`1 << 30`) divided by the `8192`-byte page size. A `const` assertion below +/// ties this to [`PAGE_SIZE`] so a future non-default `BLCKSZ` fails the build +/// rather than silently mis-numbering segment blocks. +const RELSEG_SIZE: u32 = (1u32 << 30) / 8192; + +/// Compile-time guard: [`RELSEG_SIZE`] is derived from a hardcoded `8192`-byte +/// page, so it is only correct while [`PAGE_SIZE`] is that size. Break the build +/// if the page size ever changes so this constant is revisited. +const _: () = assert!(PAGE_SIZE == 8192, "RELSEG_SIZE assumes an 8192-byte page"); + +/// Absolute first-block number of a relation segment file, from its basename. +/// +/// A segment name is `` (segment 0) or `.k` (segment `k`); the +/// suffix, when present, is the decimal segment index. The returned value is +/// `k * RELSEG_SIZE` — the absolute block number of the segment's first page, +/// used to thread the correct `block_no` into per-page checksum validation. +/// +/// Non-relation paths, or a segment index that overflows `u32`, yield `0` (a +/// safe fallback: segment 0's numbering, which is also correct for the common +/// single-segment case). A fork suffix such as `_vm` / `_fsm` is not a numeric +/// segment and is handled by [`is_relation_segment_name`] rejecting it upstream. +fn relation_segment_base(rel: &str) -> u32 { + // Take the basename (last `/`-separated component); the path prefix carries + // no segment information. + let name = rel.rsplit('/').next().unwrap_or(rel); + match name.split_once('.') { + Some((_node, segment)) => segment + .parse::() + .ok() + .and_then(|k| k.checked_mul(RELSEG_SIZE)) + .unwrap_or(0), + None => 0, + } +} + /// Whether a PG-data-relative path is a *relation file* eligible for /// page-checksum validation. /// @@ -616,13 +661,18 @@ fn is_valid_page(page: &[u8], block_no: u32, check_header: bool) -> bool { /// /// `bytes` must already be confirmed page-aligned (a multiple of `PAGE_SIZE`) /// by the caller. `check_header` enables the per-page header validation -/// (`page-header-check`) in addition to the checksum. Returns the (possibly -/// empty) list of block numbers that did not validate, in ascending order. An -/// all-empty (or empty-`bytes`) file yields an empty list. -fn validate_relation_pages(bytes: &[u8], check_header: bool) -> Vec { +/// (`page-header-check`) in addition to the checksum. `block_base` is the +/// absolute block number of this file's first page — `0` for segment ``, +/// `k * RELSEG_SIZE` for `.k` (see [`relation_segment_base`]) — so the +/// `pd_checksum` of a page in a `>= 1` segment is verified against its true +/// block number rather than a per-file index starting at 0. Returns the +/// (possibly empty) list of **absolute** block numbers that did not validate, +/// in ascending order. An all-empty (or empty-`bytes`) file yields an empty +/// list. +fn validate_relation_pages(bytes: &[u8], check_header: bool, block_base: u32) -> Vec { let mut invalid = Vec::new(); for (idx, page) in bytes.chunks_exact(PAGE_SIZE).enumerate() { - let block_no = u32::try_from(idx).unwrap_or(u32::MAX); + let block_no = u32::try_from(idx).unwrap_or(u32::MAX).saturating_add(block_base); if !is_valid_page(page, block_no, check_header) { invalid.push(block_no); } @@ -2062,7 +2112,7 @@ fn transform_and_validate(job: &CopyJob, bytes: &[u8], transform: &RepoTransform // non-relation file, which is never flagged) is left unvalidated // (`checksum_page == None`). let checksum_page = if job.validate_pages && !bytes.is_empty() && bytes.len().is_multiple_of(PAGE_SIZE) { - let invalid = validate_relation_pages(bytes, job.validate_page_header); + let invalid = validate_relation_pages(bytes, job.validate_page_header, relation_segment_base(&job.rel)); Some(if invalid.is_empty() { ChecksumPage::Validated } else { @@ -2296,7 +2346,7 @@ fn run_bundled_copy_serial(ctx: BundledCopyCtx<'_>) -> Result<(Vec // Page-checksum + page-header validation, identical to the per-file path. let checksum_page = if job.validate_pages && !bytes.is_empty() && bytes.len() % PAGE_SIZE == 0 { - let invalid = validate_relation_pages(&bytes, job.validate_page_header); + let invalid = validate_relation_pages(&bytes, job.validate_page_header, relation_segment_base(&skeleton.path)); Some(if invalid.is_empty() { ChecksumPage::Validated } else { @@ -2701,7 +2751,7 @@ fn run_bundled_copy_parallel(ctx: BundledCopyCtx<'_>) -> Result<(Vec) -> Result<(Vec bool { + entry.get("db-id").and_then(serde_json::Value::as_u64).map_or_else( + || { + log_warn(&format!( + "backup.info entry {label} has a missing or non-integer db-id; \ + treating it as ineligible for prior/resume reuse — a diff/incr may be \ + promoted to full. This is a sign of a corrupt backup.info." + )); + false + }, + |entry_db_id| entry_db_id == u64::from(db_id), + ) +} + +/// Find the label of the latest full backup recorded in `backup.info` for the +/// current `db-id`. /// /// "Latest" is the lexicographically-greatest label whose `backup-type` is -/// `full` — pgBackRust full labels sort chronologically. Returns `None` when no -/// full backup exists. -fn latest_full_label(info: &InfoBackup) -> Option { +/// `full` **and** whose `db-id` matches `db_id` — pgBackRust full labels sort +/// chronologically. A full from an older history id is skipped. Returns `None` +/// when no such full backup exists. +fn latest_full_label(info: &InfoBackup, db_id: u32) -> Option { info.current .iter() .rev() - .find(|(_, entry)| entry.get("backup-type").and_then(serde_json::Value::as_str) == Some(BACKUP_TYPE_FULL)) + .find(|(label, entry)| { + entry.get("backup-type").and_then(serde_json::Value::as_str) == Some(BACKUP_TYPE_FULL) + && entry_matches_db_id(label.as_str(), entry, db_id) + }) .map(|(label, _)| label.clone()) } /// Find the label of the latest backup of **any** type recorded in -/// `backup.info` — the "prior" backup an incremental references. +/// `backup.info` for the current `db-id` — the "prior" backup an incremental +/// references. /// /// `info.current` is a `BTreeMap` keyed by label, so its keys iterate in -/// ascending (chronological) order; the last key is the most recent backup. -/// Returns `None` when no backup exists. -fn latest_any_label(info: &InfoBackup) -> Option { - info.current.keys().next_back().cloned() +/// ascending (chronological) order; the most recent entry whose `db-id` matches +/// `db_id` is the prior. A backup from an older history id is skipped. Returns +/// `None` when no matching backup exists. +fn latest_any_label(info: &InfoBackup, db_id: u32) -> Option { + info.current + .iter() + .rev() + .find(|(label, entry)| entry_matches_db_id(label.as_str(), entry, db_id)) + .map(|(label, _)| label.clone()) } /// Convert a Unix timestamp (seconds, UTC) to `(year, month, day, hour, minute, @@ -3579,10 +3670,9 @@ fn run_copy_jobs( /// re-copy of files whose checksum still matches. struct ResumeContext { /// The aborted prior run's manifest (its file entries are the reuse source). + /// Each candidate's repo object is checked for presence via the copy job's + /// `rel_dest` (the suffixed `/` path) before reuse. manifest: Manifest, - /// Storage-relative backup root, used to check each candidate's repo object - /// is actually present before reusing it. - backup_root: String, } impl ResumeContext { @@ -3600,10 +3690,7 @@ impl ResumeContext { // encrypted repo; load it keyed so resume can read it (`None` == the // plaintext load on an unencrypted repo). let manifest = Manifest::load_keyed(repo_storage, &manifest_path, sub_key).ok()?; - Some(Self { - manifest, - backup_root: backup_root.to_owned(), - }) + Some(Self { manifest }) } /// Move every job whose file the prior backup already holds out of @@ -3617,6 +3704,7 @@ impl ResumeContext { /// just leaves the file on the copy path. fn split_resumable( &self, + repo_storage: &dyn Storage, pg_storage: &dyn Storage, skeletons: &mut Vec, jobs: &mut Vec, @@ -3627,7 +3715,7 @@ impl ResumeContext { // The two vectors are index-parallel: skeleton[i] corresponds to job[i]. for (skeleton, job) in std::mem::take(skeletons).into_iter().zip(std::mem::take(jobs)) { - if let Some(reused_file) = self.try_reuse(pg_storage, &skeleton) { + if let Some(reused_file) = self.try_reuse(repo_storage, pg_storage, &skeleton, &job) { reused.push(reused_file); } else { kept_skeletons.push(skeleton); @@ -3643,20 +3731,37 @@ impl ResumeContext { /// Try to reuse one planned file from the prior partial backup; `None` when /// it cannot be reused (no matching prior entry, size / checksum mismatch, /// repo object missing, or a read error). - fn try_reuse(&self, pg_storage: &dyn Storage, skeleton: &ManifestFile) -> Option { + fn try_reuse( + &self, + repo_storage: &dyn Storage, + pg_storage: &dyn Storage, + skeleton: &ManifestFile, + job: &CopyJob, + ) -> Option { let prior = self.manifest.file(&skeleton.path)?; if prior.size != skeleton.size { return None; } let prior_checksum = prior.checksum.as_deref()?; - // The prior entry's repo object must still be present to reuse it. The - // repo filename carries the compression suffix; a referenced (not copied) - // prior entry has no standalone object and is not reusable here. + // A referenced (not copied) prior entry has no standalone object in this + // backup root and is not reusable here. if prior.reference.is_some() { return None; } + // The prior entry's repo object must actually be present to reuse it — + // the partial manifest can list a file whose bytes never finished being + // written (a crash mid-copy), in which case reusing the manifest entry + // would reference a missing/torn object. `job.rel_dest` is the + // repo-relative path *with* the compression suffix + // (`/`), i.e. exactly the object the prior run + // would have written. A missing object (or an `exists` I/O error) falls + // back to the copy path. + if !repo_storage.exists(&PathBuf::from(&job.rel_dest)).unwrap_or(false) { + return None; + } + // Recompute the current source's plaintext checksum and compare. A read // failure (e.g. the file vanished) just falls back to the copy path. let bytes = pg_storage.open_read(&PathBuf::from(&skeleton.path)).ok()?.read_all().ok()?; @@ -3668,7 +3773,6 @@ impl ResumeContext { // The reused file keeps the prior entry's checksum/page result but is a // standalone copied file in *this* backup (reference stays None): its // bytes already live at `/` from the prior run. - let _ = &self.backup_root; Some(ManifestFile { checksum: Some(checksum), checksum_page: prior.checksum_page.clone(), @@ -3911,11 +4015,23 @@ fn plan_backup( }, StorageKind::Path => plan.paths.push(ManifestPath { path: entry.rel }), StorageKind::Link => { - // TODO: resolve link target once `Storage` exposes a - // link-target accessor; record an empty destination for now. + // Record the link's real target. Filesystem backends resolve it + // via `read_link`; backends without link-target support (the + // remote/object stores) return `StorageError::Unsupported`, in + // which case we fall back to an empty destination — the same + // known-limitation behaviour as before, kept out of the restore + // path's hard error only for those backends. A genuine local I/O + // failure maps to `Backend`/`Io`, NOT `Unsupported`, so it + // propagates and fails the backup loudly instead of silently + // recording an empty (unrestorable) symlink target. + let destination = match pg_storage.read_link(Path::new(&entry.rel)) { + Ok(target) => target.to_string_lossy().into_owned(), + Err(StorageError::Unsupported { .. }) => String::new(), + Err(err) => return Err(err.into()), + }; plan.links.push(ManifestLink { path: entry.rel, - destination: String::new(), + destination, }); } // Sockets / FIFOs / devices are not part of a base backup. @@ -3950,11 +4066,11 @@ fn resolve_prior( BackupType::Full => return Ok((None, None)), // No prior full (diff) / no prior backup at all (incr) → no base to // reference; signal the caller (run_backup) to promote this to a full. - BackupType::Diff => match latest_full_label(info) { + BackupType::Diff => match latest_full_label(info, info.db_id) { Some(label) => label, None => return Ok((None, None)), }, - BackupType::Incr => match latest_any_label(info) { + BackupType::Incr => match latest_any_label(info, info.db_id) { Some(label) => label, None => return Ok((None, None)), }, @@ -3998,7 +4114,8 @@ fn resolve_prior( /// (compress then encrypt), write the transformed bytes to /// `backup//