diff --git a/Cargo.lock b/Cargo.lock index bb25278..9618e08 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -369,10 +369,12 @@ dependencies = [ "criterion", "flate2", "jiff", + "libc", "lz4_flex", "postcard", "proptest", "rusqlite", + "rustix", "serde", "serde_json", "tempfile", diff --git a/Cargo.toml b/Cargo.toml index 4989ea4..d12e4f8 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -16,10 +16,12 @@ flate2 = "1.1.5" jiff = "0.2.34" lz4_flex = "0.14.0" postcard = { version = "1.1.3", features = ["use-std"] } +rustix = { version = "1.1.4", features = ["fs"] } serde = { version = "1.0.229", features = ["derive"] } [dev-dependencies] criterion = "0.8.2" +libc = "0.2.187" proptest = "1.11.0" rusqlite = { version = "0.40.1", features = ["bundled"] } serde_json = "1.0.149" diff --git a/README.md b/README.md index 1903de7..d71b53f 100644 --- a/README.md +++ b/README.md @@ -61,6 +61,7 @@ Verify or snapshot a directory store: ```sh cargo run --release -- check-store ./energy.ftwdb cargo run --release -- backup ./energy.ftwdb ./backups/energy-2026-07-21.ftwdb +cargo run --release -- restore ./backups/energy-2026-07-21.ftwdb ./restored-energy.ftwdb ``` ## Design documents @@ -70,7 +71,7 @@ cargo run --release -- backup ./energy.ftwdb ./backups/energy-2026-07-21.ftwdb - [Storage format v1](docs/format.md) - [Immutable segment format](docs/segment-format.md) - [Persistent rollups and retention](docs/rollups.md) -- [Integrity checks and backup](docs/operations.md) +- [Integrity checks, backup, and restore](docs/operations.md) - [OSS database research](docs/research.md) - [Benchmark protocol](docs/benchmarking.md) - [Deterministic energy workload](docs/workload.md) diff --git a/bench/sd-card-emulator/README.md b/bench/sd-card-emulator/README.md index 6d9a05a..61fbfcb 100644 --- a/bench/sd-card-emulator/README.md +++ b/bench/sd-card-emulator/README.md @@ -67,8 +67,12 @@ docker run --rm --privileged \ The script installs `nbd-client` and `e2fsprogs` in the disposable container. It writes raw results under the ignored `bench-results/linux-nbd-smoke` -directory. It uses the checked-in real-installation fixture and rejects a -changed active-log checksum or recovered FTWDB watermark. +directory, which must be absent or empty when the run starts. It uses the +checked-in real-installation fixture and rejects a +changed active-log checksum or recovered FTWDB watermark. After recovery, it +backs up the checked store outside the emulated card, restores that backup to a +new store on the card, and compares checked counts, snapshot CRCs, and active +log SHA-256 values. ```sh mkdir -p bench-results/sd-emulator diff --git a/bench/sd-card-emulator/linux-smoke.sh b/bench/sd-card-emulator/linux-smoke.sh index d7400fb..5eb4110 100755 --- a/bench/sd-card-emulator/linux-smoke.sh +++ b/bench/sd-card-emulator/linux-smoke.sh @@ -4,6 +4,10 @@ set -euo pipefail source bench/sd-card-emulator/linux-nbd-common.sh out=${FTW_NBD_OUTPUT:-/work/bench-results/linux-nbd-smoke} +if [[ -e "$out" ]] && find "$out" -mindepth 1 -print -quit | grep -q .; then + printf 'output directory must be absent or empty: %s\n' "$out" >&2 + exit 1 +fi linux_nbd_prepare trap linux_nbd_cleanup EXIT linux_nbd_start bench/sd-card-emulator/profiles/healthy.json 42 @@ -58,12 +62,57 @@ test "$before_hash" = "$after_hash" --output "$out/verification.jsonl" \ >"$out/verification.json" +# Keep the backup off the emulated card, then prove that a new store restored +# onto the recovered card has the same checked counts and selected bytes. +"$ftw" backup "$mount_dir/database" "$out/restore-backup" \ + >"$out/restore-backup.json" +"$ftw" restore "$out/restore-backup" "$mount_dir/restored-database" \ + >"$out/restore.json" +"$ftw" check-store "$mount_dir/restored-database" \ + >"$out/restore-check.json" +sha256sum "$out/restore-backup/active.wlog" >"$out/restore-backup-sha.txt" +sha256sum "$mount_dir/restored-database/active.wlog" >"$out/restore-target-sha.txt" + +json_u64() { + local file=$1 + local field=$2 + local value + value=$(sed -n "s/.*\"$field\":\([0-9][0-9]*\).*/\1/p" "$file") + test -n "$value" + printf '%s\n' "$value" +} + +json_string() { + local file=$1 + local field=$2 + local value + value=$(sed -n "s/.*\"$field\":\"\([^\"]*\)\".*/\1/p" "$file") + test -n "$value" + printf '%s\n' "$value" +} + +test "$(json_u64 "$out/check-after.json" raw_points)" = \ + "$(json_u64 "$out/restore-check.json" raw_points)" +test "$(json_u64 "$out/check-after.json" raw_commits)" = \ + "$(json_u64 "$out/restore-check.json" raw_commits)" +test "$(json_u64 "$out/check-after.json" raw_points)" = \ + "$(json_u64 "$out/restore.json" raw_points)" +test "$(json_u64 "$out/check-after.json" raw_commits)" = \ + "$(json_u64 "$out/restore.json" raw_commits)" +test "$(json_string "$out/restore.json" source_snapshot_crc32)" = \ + "$(json_string "$out/restore.json" destination_snapshot_crc32)" +test "$(cut -d' ' -f1 "$out/restore-backup-sha.txt")" = \ + "$(cut -d' ' -f1 "$out/restore-target-sha.txt")" + linux_nbd_finish trap - EXIT printf 'linux_nbd_smoke=passed\n' +printf 'restore_drill=passed\n' printf 'fsck_exit=%d\n' "$fsck_exit" printf 'active_log_sha256=%s\n' "$after_hash" cat "$out/load.json" cat "$out/check-after.json" cat "$out/verification.json" +cat "$out/restore.json" +cat "$out/restore-check.json" diff --git a/docs/operations.md b/docs/operations.md index 4e20f93..d6f0f77 100644 --- a/docs/operations.md +++ b/docs/operations.md @@ -1,8 +1,8 @@ -# Integrity checks and backup +# Integrity checks, backup, and restore ## Integrity check -`ftwdb check-store ` opens the store read-only: the active commit +`ftw check-store ` opens the store read-only: the active commit log is opened without write access under a shared lock, torn-tail recovery is simulated in memory, and no manifest generation is published, pruned, or swept. It loads the highest valid manifest generation and re-opens every active rollup @@ -22,7 +22,7 @@ startup; a read-only open never deletes them. ## Snapshot backup -`ftwdb backup ` opens the source read-only, so a +`ftw backup ` opens the source read-only, so a backup can never alter (or accidentally create) the store it is copying, and uses this publication order: @@ -32,9 +32,18 @@ uses this publication order: 4. hard-link active immutable rollups and the selected manifest when source and destination share a filesystem, otherwise copy and sync them; 5. sync both subdirectories and the temporary root; -6. rename the temporary root to the requested absent destination; +6. publish the temporary root with an atomic no-clobber rename; 7. sync the parent directory; -8. open and integrity-check the published backup. +8. open the published backup read-only and integrity-check it. + +Backup and restore use Linux `renameat2(RENAME_NOREPLACE)` or the matching +Apple exclusive rename through `rustix`. They do not use an `exists()` check +followed by a replacing rename. On an unsupported Unix target, publication +returns an unsupported-operation error instead of weakening this rule. If the +final parent sync or post-check fails, the code verifies the directory identity +and tries to move its own publication back to its hidden name, remove it, and +sync the parent. If that rollback also fails, the error says that rollback +failed; the caller must inspect the named destination before retrying. The active log is always copied, never hard-linked, because later appends to a shared inode would silently mutate the backup. A unit test writes to the source @@ -49,8 +58,58 @@ the link error kind. If both fail, the returned I/O error keeps the copy error kind and its text and source also include the hard-link failure. This is a local consistent snapshot, not yet a remote backup policy. Encryption, -incremental upload, retention, restore drills, and salvage of a corrupted source -remain operational work. +incremental upload, retention, and salvage of a corrupted source remain open. + +## Strict restore + +`ftw restore ` restores only a fully valid snapshot. +The command exits with code 2 for missing or extra arguments and code 1 for a +store, file, or publication error. It has no replace option. An existing target, +including an empty directory or a dangling symlink, causes `AlreadyExists` and +stays unchanged. + +Restore opens the backup read-only under a shared lock, runs the full store +check, and refuses any raw-log recovery. Both recovered tail bytes and a +recovery reason must be zero. A short final header or payload, a full bad frame, +a bad selected manifest, a bad active rollup, or an active rollup whose source +watermark trails the raw log causes an error. Orphan rollups, older manifest +generations, and other files outside the selected snapshot do not affect the +restore and do not appear in the target. + +The selected snapshot contains: + +- `active.wlog`, always copied; +- the selected manifest generation, when one exists; +- each active immutable rollup named by that manifest. + +Every selected path must be a regular file according to `symlink_metadata` and +the opened file identity. Read-only file opens use no-follow and nonblocking +flags, and snapshot checksum traversal uses directory file descriptors, so +restore rejects symlinks and special files without blocking or leaving the +snapshot root. It copies the selected files to a unique hidden sibling +directory, syncs each file and directory, and opens that stage read-only for a +full check. The stage lock stays held through no-clobber publication and the +target's read-only post-check, so a writer cannot append between publication +and verification. + +The snapshot checksum uses the CRC32 domain `FTWDB snapshot CRC32 v1\0` and the +selected relative paths sorted by UTF-8 bytes. Each file adds, in order, the +path byte length as little-endian `u64`, the path bytes, the file length as +little-endian `u64`, and the exact file bytes. The byte count is the sum of +those file lengths. Permissions, times, directories, and unselected files do +not enter the checksum. Restore compares source and stage before publication, +then source and target after publication. + +On success, `ftwdb-restore-v1` reports `files`, `bytes`, +`manifest_generation`, `raw_commits`, `raw_points`, +`source_snapshot_crc32`, and `destination_snapshot_crc32`. The CRC values are +eight lower-case hex digits. Equal CRCs are a deterministic corruption check, +not proof that the selected bytes match, and they do not turn CRC32 into a +cryptographic authenticity check. + +Restore does not repair a damaged backup and does not infer data past a corrupt +frame. Use a separate target, keep the source backup, and run `ftw check-store` +on the result. Salvage will use a separate command and safety review. ## Sync and full-disk checks @@ -73,12 +132,14 @@ the M4 physical SD-card power-cut release gate. `tests/cli.rs` runs the built `ftw` binary as a subprocess. It fixes the usage contract at exit code 2 with usage text on standard error, while data, file, and store errors use exit code 1. The test covers the generated workload, -sanitized real fixture, TSBS IoT, integrity check, log inspection, and backup -paths. It parses each promised JSON record and checks its main counts and -status fields. +sanitized real fixture, TSBS IoT, integrity check, log inspection, backup, and +strict restore paths. It parses each promised JSON record and checks its main +counts and status fields. The test also snapshots every path and every file byte in a store before and after both `check-store` and `inspect`. Separate missing-path checks prove that neither command creates its input. The backup check covers linked and copied file counts plus the hard-link fallback count on a normal local filesystem. -Restore drills, salvage, and physical SD-card tests remain M4 work. +The restore check covers JSON, exact bytes, counts, checksum equality, +corruption refusal, and no-clobber behavior. Salvage and physical SD-card tests +remain M4 work. diff --git a/docs/roadmap.md b/docs/roadmap.md index 1f2639b..4852a24 100644 --- a/docs/roadmap.md +++ b/docs/roadmap.md @@ -80,12 +80,17 @@ workload and failure surface. A deterministic NBD SD-card emulator and one ARM64 Linux/ext4 smoke run add repeatable media-cache and power-loss checks, but the recorded cut happened after the load and host sync. A repeatable 64 MiB Linux/NBD/ext4 run reached `ENOSPC` after 770,000 durable points in 77 commits; -`check-store` and `inspect` passed on that prefix. Local snapshot backup plus -post-publication integrity verification is implemented. Process-level CLI -tests now cover usage errors, generated and sanitized input paths, integrity -checks, inspection, backup JSON, and byte-for-byte read-only behavior. Together -with the process-lock, sync-failure, `/dev/full`, and NBD tests, this closes the +`check-store` and `inspect` passed on that prefix. Local snapshot backup and +strict restore now use atomic no-clobber publication, read-only checks, selected +file CRCs, and tested rollback after final sync or post-check errors. +Process-level CLI tests cover usage errors, generated and sanitized input paths, +integrity checks, inspection, backup and restore JSON, exact restored bytes, +and byte-for-byte read-only behavior. The sanitized fixture restore fixes the +expected result at 889,978 points, 89 commits, and snapshot CRC32 `9ff1a95b`. +The Linux NBD power-loss run now follows recovery with an off-card backup and a +verified restore to a new target on the emulated card. Together with the +process-lock, sync-failure, `/dev/full`, and NBD tests, this closes the robustness gaps tracked in issue #17. Tests on the target board and SD cards, -power cuts during commits, soak runs, remote backup policy, restore drills, -corruption salvage, and the remaining result-verified adapters stay open and -are required for the M4 exit. +power cuts during commits, soak runs, remote backup policy, corruption salvage, +and the remaining result-verified adapters stay open and are required for the +M4 exit. diff --git a/src/bin/ftw.rs b/src/bin/ftw.rs index a56367c..87a8bef 100644 --- a/src/bin/ftw.rs +++ b/src/bin/ftw.rs @@ -1,7 +1,8 @@ use flate2::read::GzDecoder; use ftwdb::{ - BackupReport, Config, Database, Durability, EnergyWorkload, Error, RollupResolution, Store, - Transaction, WorkloadConfig, gauge_bucket_checksum, load_real_fixture, load_tsbs_iot, + BackupReport, Config, Database, Durability, EnergyWorkload, Error, RestoreReport, + RollupResolution, Store, Transaction, WorkloadConfig, gauge_bucket_checksum, load_real_fixture, + load_tsbs_iot, }; use std::env; use std::fs::File; @@ -40,6 +41,7 @@ fn main() -> ExitCode { Some("inspect") => inspect(&arguments[2..]), Some("check-store") => check_store(&arguments[2..]), Some("backup") => backup(&arguments[2..]), + Some("restore") => restore(&arguments[2..]), Some("generate") => generate(&arguments[2..]), Some("bench-ftwdb") => bench_ftwdb(&arguments[2..]), Some("bench-real-fixture") => bench_real_fixture(&arguments[2..]), @@ -292,6 +294,30 @@ fn backup(arguments: &[String]) -> CliResult<()> { Ok(()) } +fn restore(arguments: &[String]) -> CliResult<()> { + if arguments.len() != 2 { + return Err(usage_error( + "restore requires a backup and absent target directory", + )); + } + let report = Store::restore_from(&arguments[0], &arguments[1])?; + println!("{}", restore_report_json(&report)); + Ok(()) +} + +fn restore_report_json(report: &RestoreReport) -> String { + format!( + "{{\"format\":\"ftwdb-restore-v1\",\"files\":{},\"bytes\":{},\"manifest_generation\":{},\"raw_commits\":{},\"raw_points\":{},\"source_snapshot_crc32\":\"{:08x}\",\"destination_snapshot_crc32\":\"{:08x}\"}}", + report.files, + report.bytes, + report.manifest_generation, + report.raw_commits, + report.raw_points, + report.source_snapshot_crc32, + report.destination_snapshot_crc32 + ) +} + fn backup_report_json(report: &BackupReport) -> String { let fallback_error_kinds = report .hard_link_fallback_error_kinds @@ -548,14 +574,14 @@ fn runtime_invalid(reason: impl Into) -> CliError { fn usage(program: &str) { eprintln!( - "usage:\n {program} inspect \n {program} check-store \n {program} backup \n {program} generate [--seed N] [--sites N] [--days N] [--cadence-seconds N] [--start-micros N]\n {program} bench-ftwdb [--durability always|manual] [--batch-points N]\n {program} bench-real-fixture [--durability always|manual|every-bytes:N] [--batch-points N]\n {program} bench-tsbs-iot [--durability always|manual|every-bytes:N] [--batch-rows N]" + "usage:\n {program} inspect \n {program} check-store \n {program} backup \n {program} restore \n {program} generate [--seed N] [--sites N] [--days N] [--cadence-seconds N] [--start-micros N]\n {program} bench-ftwdb [--durability always|manual] [--batch-points N]\n {program} bench-real-fixture [--durability always|manual|every-bytes:N] [--batch-points N]\n {program} bench-tsbs-iot [--durability always|manual|every-bytes:N] [--batch-rows N]" ); } #[cfg(test)] mod tests { - use super::backup_report_json; - use ftwdb::BackupReport; + use super::{backup_report_json, restore_report_json}; + use ftwdb::{BackupReport, RestoreReport}; #[test] fn backup_json_uses_fixed_error_kind_names() { @@ -574,4 +600,22 @@ mod tests { "{\"format\":\"ftwdb-backup-v1\",\"files\":4,\"bytes\":123,\"manifest_generation\":7,\"linked_files\":2,\"copied_files\":2,\"hard_link_fallbacks\":1,\"hard_link_fallback_error_kinds\":[\"crosses-devices\"]}" ); } + + #[test] + fn restore_json_has_versioned_checksums_and_counts() { + let report = RestoreReport { + files: 4, + bytes: 123, + manifest_generation: 7, + raw_commits: 8, + raw_points: 9, + source_snapshot_crc32: 0x0123_abcd, + destination_snapshot_crc32: 0x0123_abcd, + }; + + assert_eq!( + restore_report_json(&report), + "{\"format\":\"ftwdb-restore-v1\",\"files\":4,\"bytes\":123,\"manifest_generation\":7,\"raw_commits\":8,\"raw_points\":9,\"source_snapshot_crc32\":\"0123abcd\",\"destination_snapshot_crc32\":\"0123abcd\"}" + ); + } } diff --git a/src/error.rs b/src/error.rs index 1c7e051..2e49680 100644 --- a/src/error.rs +++ b/src/error.rs @@ -31,6 +31,12 @@ pub enum Error { path: PathBuf, }, ReadOnly, + /// A no-clobber rename succeeded, but a later durability or verification + /// step failed and the process could not fully roll its own directory back. + SnapshotPublication { + path: PathBuf, + reason: String, + }, } impl fmt::Display for Error { @@ -64,6 +70,11 @@ impl fmt::Display for Error { f, "database is open read-only; reopen it writable to modify it" ), + Self::SnapshotPublication { path, reason } => write!( + f, + "snapshot publication at {} could not be rolled back cleanly: {reason}", + path.display() + ), } } } diff --git a/src/lib.rs b/src/lib.rs index 02d4468..131fc33 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -17,6 +17,7 @@ mod real_fixture; mod rollup; mod rollup_segment; mod segment; +mod snapshot; mod storage; mod store; mod transaction; @@ -38,8 +39,8 @@ pub use rollup_segment::{RollupSegment, RollupSegmentStats}; pub use segment::{Segment, SegmentStats}; pub use storage::{Commit, Config, Database, Durability, PlanOutcome, Point, RecoveredTail, Stats}; pub use store::{ - BackupReport, IntegrityReport, MaintenanceReport, RetentionGate, RollupQuery, RollupSource, - Store, + BackupReport, IntegrityReport, MaintenanceReport, RestoreReport, RetentionGate, RollupQuery, + RollupSource, Store, }; pub use transaction::Transaction; pub use tsbs::{TsbsIotLoadReport, load_tsbs_iot}; diff --git a/src/manifest.rs b/src/manifest.rs index e663d36..75348a2 100644 --- a/src/manifest.rs +++ b/src/manifest.rs @@ -1,4 +1,4 @@ -use crate::storage::sync_directory; +use crate::storage::{open_regular_file_read_only, sync_directory}; use crate::{Error, Result, RollupResolution}; use crc32fast::hash; use serde::{Deserialize, Serialize}; @@ -178,7 +178,7 @@ pub(crate) fn referenced_rollup_files( } fn read_manifest(path: &Path, expected_generation: u64) -> Result { - let mut file = OpenOptions::new().read(true).open(path)?; + let mut file = open_regular_file_read_only(path)?; let file_len = file.metadata()?.len(); if file_len < HEADER_BYTES as u64 { return corruption("manifest is too small"); diff --git a/src/rollup_segment.rs b/src/rollup_segment.rs index 4e868f4..36554d7 100644 --- a/src/rollup_segment.rs +++ b/src/rollup_segment.rs @@ -1,4 +1,4 @@ -use crate::storage::sync_parent_directory; +use crate::storage::{open_regular_file_read_only, sync_parent_directory}; use crate::{Error, GaugeBucket, Result, Sample}; use crc32fast::hash; use lz4_flex::block::{compress_prepend_size, decompress}; @@ -70,7 +70,7 @@ impl RollupSegment { } pub fn open(path: impl AsRef) -> Result { - let mut file = OpenOptions::new().read(true).open(path)?; + let mut file = open_regular_file_read_only(path.as_ref())?; let file_len = file.metadata()?.len(); if file_len < HEADER_BYTES as u64 { return corruption(0, "rollup segment is too small"); diff --git a/src/snapshot.rs b/src/snapshot.rs new file mode 100644 index 0000000..1da71b3 --- /dev/null +++ b/src/snapshot.rs @@ -0,0 +1,551 @@ +use crate::storage::sync_directory; +use crate::{Error, Result}; +use crc32fast::Hasher; +use std::fs::File; +use std::io::Read; +use std::os::unix::fs::MetadataExt; +use std::path::{Component, Path, PathBuf}; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::time::{SystemTime, UNIX_EPOCH}; + +const SNAPSHOT_CRC32_DOMAIN: &[u8] = b"FTWDB snapshot CRC32 v1\0"; +const CHECKSUM_BUFFER_BYTES: usize = 64 * 1024; +const STAGE_ATTEMPTS: u64 = 64; + +static STAGE_COUNTER: AtomicU64 = AtomicU64::new(0); + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub(crate) struct SnapshotDigest { + pub(crate) files: usize, + pub(crate) bytes: u64, + pub(crate) crc32: u32, +} + +/// Computes the FTWDB snapshot CRC32 over the selected files only. +/// +/// Paths must be unique, safe, UTF-8 relative paths sorted by their UTF-8 +/// bytes. The checksum domain is `FTWDB snapshot CRC32 v1\0`. Each file then +/// contributes its path length as little-endian `u64`, path bytes, file length +/// as little-endian `u64`, and exact file bytes. Directory entries, mtimes, +/// permissions, inactive manifests, and unselected rollups are excluded. +pub(crate) fn snapshot_digest(root: &Path, relative_paths: &[String]) -> Result { + let root_metadata = std::fs::symlink_metadata(root)?; + if !root_metadata.file_type().is_dir() { + return Err(Error::Io(std::io::Error::new( + std::io::ErrorKind::InvalidData, + "snapshot root is not a directory", + ))); + } + let mut previous: Option<&str> = None; + let mut hasher = Hasher::new(); + hasher.update(SNAPSHOT_CRC32_DOMAIN); + let mut bytes = 0_u64; + let mut buffer = vec![0_u8; CHECKSUM_BUFFER_BYTES]; + + for relative in relative_paths { + validate_relative_path(relative)?; + if previous.is_some_and(|previous| previous >= relative.as_str()) { + return Err(Error::InvalidArgument( + "snapshot paths must be unique and sorted", + )); + } + previous = Some(relative); + + let path_bytes = relative.as_bytes(); + let path_len = u64::try_from(path_bytes.len()) + .map_err(|_| Error::Serialization("snapshot path is too long".to_owned()))?; + let mut file = open_snapshot_file(root, relative)?; + let file_metadata = file.metadata()?; + let file_len = file_metadata.len(); + bytes = bytes + .checked_add(file_len) + .ok_or_else(|| Error::Serialization("snapshot byte count exceeds u64".to_owned()))?; + + hasher.update(&path_len.to_le_bytes()); + hasher.update(path_bytes); + hasher.update(&file_len.to_le_bytes()); + + let mut remaining = file_len; + while remaining > 0 { + let requested = usize::try_from(remaining.min(buffer.len() as u64)).unwrap(); + let read = file.read(&mut buffer[..requested])?; + if read == 0 { + return Err(Error::Io(std::io::Error::new( + std::io::ErrorKind::UnexpectedEof, + format!("snapshot file changed while reading {relative}"), + ))); + } + hasher.update(&buffer[..read]); + remaining -= read as u64; + } + if file.read(&mut buffer[..1])? != 0 { + return Err(Error::Io(std::io::Error::new( + std::io::ErrorKind::InvalidData, + format!("snapshot file grew while reading {relative}"), + ))); + } + } + + Ok(SnapshotDigest { + files: relative_paths.len(), + bytes, + crc32: hasher.finalize(), + }) +} + +fn open_snapshot_file(root: &Path, relative: &str) -> Result { + use rustix::fs::{Mode, OFlags, open, openat}; + + let components: Vec<_> = Path::new(relative).components().collect(); + let mut directory = open( + root, + OFlags::RDONLY | OFlags::CLOEXEC | OFlags::NOFOLLOW | OFlags::DIRECTORY, + Mode::empty(), + ) + .map_err(|error| Error::Io(error.into()))?; + for (index, component) in components.iter().enumerate() { + let Component::Normal(component) = component else { + return Err(Error::InvalidArgument( + "snapshot path must be a safe relative path", + )); + }; + let last = index + 1 == components.len(); + let flags = if last { + OFlags::RDONLY | OFlags::CLOEXEC | OFlags::NOFOLLOW | OFlags::NONBLOCK + } else { + OFlags::RDONLY | OFlags::CLOEXEC | OFlags::NOFOLLOW | OFlags::DIRECTORY + }; + let opened = openat(&directory, *component, flags, Mode::empty()) + .map_err(|error| Error::Io(error.into()))?; + if last { + let file = File::from(opened); + if file.metadata()?.file_type().is_file() { + return Ok(file); + } + return Err(Error::Io(std::io::Error::new( + std::io::ErrorKind::InvalidData, + format!("selected snapshot path is not a regular file: {relative}"), + ))); + } + directory = opened; + } + Err(Error::InvalidArgument( + "snapshot path must name a regular file", + )) +} + +fn validate_relative_path(relative: &str) -> Result<()> { + let path = Path::new(relative); + if relative.is_empty() + || path + .components() + .any(|component| !matches!(component, Component::Normal(_))) + || path.is_absolute() + { + return Err(Error::InvalidArgument( + "snapshot path must be a safe relative path", + )); + } + Ok(()) +} + +/// Owns a hidden sibling directory until an atomic no-clobber publish succeeds. +/// Any error before the rename removes the stage on drop. +pub(crate) struct StagedDirectory { + path: PathBuf, + published: bool, +} + +impl StagedDirectory { + pub(crate) fn create(destination: &Path, operation: &str) -> Result { + reject_existing(destination)?; + let parent = destination_parent(destination)?; + std::fs::create_dir_all(parent)?; + let name = destination + .file_name() + .and_then(|name| name.to_str()) + .ok_or(Error::InvalidConfig( + "snapshot destination must have a UTF-8 name", + ))?; + let timestamp = SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap_or_default() + .as_nanos(); + + for attempt in 0..STAGE_ATTEMPTS { + let counter = STAGE_COUNTER.fetch_add(1, Ordering::Relaxed); + let path = parent.join(format!( + ".{name}.{operation}-{}-{timestamp}-{counter}-{attempt}", + std::process::id() + )); + match std::fs::create_dir(&path) { + Ok(()) => { + return Ok(Self { + path, + published: false, + }); + } + Err(error) if error.kind() == std::io::ErrorKind::AlreadyExists => continue, + Err(error) => return Err(Error::Io(error)), + } + } + + Err(Error::Io(std::io::Error::new( + std::io::ErrorKind::AlreadyExists, + "could not allocate a unique snapshot staging directory", + ))) + } + + pub(crate) fn path(&self) -> &Path { + &self.path + } + + pub(crate) fn publish(mut self, destination: &Path) -> Result { + publication_checkpoint(PublicationStep::Publish)?; + let identity = FileIdentity::read(&self.path)?; + rename_noreplace(&self.path, destination)?; + self.published = true; + let publication = PublishedDirectory { + hidden_path: self.path.clone(), + destination: destination.to_owned(), + identity, + at_destination: true, + committed: false, + }; + let parent = publication.parent()?; + let synced = publication_checkpoint(PublicationStep::ParentSync) + .and_then(|()| sync_directory(parent)); + match synced { + Ok(()) => Ok(publication), + Err(error) => Err(publication.rollback_after(error)), + } + } +} + +impl Drop for StagedDirectory { + fn drop(&mut self) { + if !self.published { + let _ = std::fs::remove_dir_all(&self.path); + } + } +} + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +struct FileIdentity { + device: u64, + inode: u64, +} + +impl FileIdentity { + fn read(path: &Path) -> Result { + let metadata = std::fs::symlink_metadata(path)?; + if !metadata.file_type().is_dir() { + return Err(Error::Io(std::io::Error::new( + std::io::ErrorKind::InvalidData, + "published snapshot path is not a directory", + ))); + } + Ok(Self { + device: metadata.dev(), + inode: metadata.ino(), + }) + } +} + +/// Keeps a just-published directory reversible until its caller finishes all +/// post-publication checks. Rollback only touches the directory when its Unix +/// device/inode identity still matches the stage that this process renamed. +pub(crate) struct PublishedDirectory { + hidden_path: PathBuf, + destination: PathBuf, + identity: FileIdentity, + at_destination: bool, + committed: bool, +} + +impl PublishedDirectory { + pub(crate) fn commit(mut self) { + self.committed = true; + } + + pub(crate) fn rollback_after(mut self, failure: Error) -> Error { + match self.rollback() { + Ok(()) => failure, + Err(rollback) => Error::SnapshotPublication { + path: self.destination.clone(), + reason: format!("operation failed: {failure}; rollback failed: {rollback}"), + }, + } + } + + fn parent(&self) -> Result<&Path> { + destination_parent(&self.destination) + } + + fn rollback(&mut self) -> Result<()> { + if !self.at_destination { + return Ok(()); + } + if FileIdentity::read(&self.destination)? != self.identity { + return Err(Error::Io(std::io::Error::new( + std::io::ErrorKind::ResourceBusy, + "published destination identity changed before rollback", + ))); + } + rename_noreplace(&self.destination, &self.hidden_path)?; + self.at_destination = false; + let parent = self.parent()?.to_owned(); + sync_directory(&parent)?; + std::fs::remove_dir_all(&self.hidden_path)?; + sync_directory(&parent)?; + Ok(()) + } +} + +fn destination_parent(destination: &Path) -> Result<&Path> { + match destination.parent() { + Some(parent) if !parent.as_os_str().is_empty() => Ok(parent), + Some(_) if destination.file_name().is_some() => Ok(Path::new(".")), + _ => Err(Error::InvalidConfig( + "snapshot destination must have a parent directory", + )), + } +} + +impl Drop for PublishedDirectory { + fn drop(&mut self) { + if !self.committed { + let _ = self.rollback(); + } + } +} + +fn reject_existing(destination: &Path) -> Result<()> { + match std::fs::symlink_metadata(destination) { + Ok(_) => Err(Error::Io(std::io::Error::new( + std::io::ErrorKind::AlreadyExists, + "snapshot destination already exists", + ))), + Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(()), + Err(error) => Err(Error::Io(error)), + } +} + +#[cfg(any( + target_os = "android", + target_os = "linux", + target_os = "macos", + target_os = "ios", + target_os = "tvos", + target_os = "visionos", + target_os = "watchos", + target_os = "redox", +))] +fn rename_noreplace(source: &Path, destination: &Path) -> Result<()> { + use rustix::fs::{CWD, RenameFlags, renameat_with}; + + renameat_with(CWD, source, CWD, destination, RenameFlags::NOREPLACE) + .map_err(|error| Error::Io(error.into())) +} + +#[cfg(not(any( + target_os = "android", + target_os = "linux", + target_os = "macos", + target_os = "ios", + target_os = "tvos", + target_os = "visionos", + target_os = "watchos", + target_os = "redox", +)))] +fn rename_noreplace(_source: &Path, _destination: &Path) -> Result<()> { + Err(Error::Io(std::io::Error::new( + std::io::ErrorKind::Unsupported, + "this platform lacks atomic no-clobber directory publication", + ))) +} + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub(crate) enum PublicationStep { + Copy, + Sync, + #[cfg(test)] + ChecksumMismatch, + Publish, + ParentSync, + PostCheck, +} + +#[cfg(test)] +std::thread_local! { + static FAIL_NEXT_PUBLICATION_STEP: std::cell::Cell> = const { + std::cell::Cell::new(None) + }; +} + +#[cfg(test)] +pub(crate) fn fail_next_publication_step(step: PublicationStep) { + FAIL_NEXT_PUBLICATION_STEP.with(|failure| failure.set(Some(step))); +} + +pub(crate) fn publication_checkpoint(step: PublicationStep) -> Result<()> { + #[cfg(test)] + if FAIL_NEXT_PUBLICATION_STEP.with(|failure| failure.get()) == Some(step) { + FAIL_NEXT_PUBLICATION_STEP.with(|failure| failure.set(None)); + return Err(Error::Io(std::io::Error::other(format!( + "injected {step:?} failure" + )))); + } + let _ = step; + Ok(()) +} + +pub(crate) fn inject_checksum_mismatch(digest: SnapshotDigest) -> SnapshotDigest { + #[cfg(test)] + if FAIL_NEXT_PUBLICATION_STEP.with(|failure| failure.get()) + == Some(PublicationStep::ChecksumMismatch) + { + FAIL_NEXT_PUBLICATION_STEP.with(|failure| failure.set(None)); + return SnapshotDigest { + crc32: digest.crc32 ^ 1, + ..digest + }; + } + digest +} + +#[cfg(test)] +mod tests { + use super::{StagedDirectory, snapshot_digest}; + use std::sync::{Arc, Barrier}; + use tempfile::tempdir; + + #[test] + fn digest_ignores_unselected_files_and_changes_with_paths_lengths_or_bytes() { + let directory = tempdir().unwrap(); + std::fs::create_dir(directory.path().join("nested")).unwrap(); + std::fs::write(directory.path().join("one"), b"first").unwrap(); + std::fs::write(directory.path().join("nested/two"), b"second").unwrap(); + std::fs::write(directory.path().join("stale"), b"ignored").unwrap(); + let selected = vec!["nested/two".to_owned(), "one".to_owned()]; + + let first = snapshot_digest(directory.path(), &selected).unwrap(); + std::fs::write(directory.path().join("stale"), b"changed").unwrap(); + assert_eq!(snapshot_digest(directory.path(), &selected).unwrap(), first); + + std::fs::write(directory.path().join("one"), b"different").unwrap(); + assert_ne!(snapshot_digest(directory.path(), &selected).unwrap(), first); + } + + #[test] + fn digest_rejects_symlinked_parents_and_non_files() { + let directory = tempdir().unwrap(); + let external = tempdir().unwrap(); + std::fs::write(external.path().join("outside"), b"external").unwrap(); + std::os::unix::fs::symlink(external.path(), directory.path().join("linked")).unwrap(); + let linked = vec!["linked/outside".to_owned()]; + assert!(snapshot_digest(directory.path(), &linked).is_err()); + + std::fs::create_dir(directory.path().join("not-a-file")).unwrap(); + let not_a_file = vec!["not-a-file".to_owned()]; + assert!(snapshot_digest(directory.path(), ¬_a_file).is_err()); + } + + #[test] + fn digest_rejects_unsafe_unsorted_and_duplicate_paths() { + let directory = tempdir().unwrap(); + std::fs::write(directory.path().join("one"), b"first").unwrap(); + std::fs::write(directory.path().join("two"), b"second").unwrap(); + + for paths in [ + vec!["../one".to_owned()], + vec!["/one".to_owned()], + vec!["two".to_owned(), "one".to_owned()], + vec!["one".to_owned(), "one".to_owned()], + ] { + assert!(snapshot_digest(directory.path(), &paths).is_err()); + } + } + + #[test] + fn simultaneous_no_clobber_publications_have_exactly_one_winner() { + let directory = tempdir().unwrap(); + let destination = directory.path().join("target"); + let left = StagedDirectory::create(&destination, "race-left").unwrap(); + let right = StagedDirectory::create(&destination, "race-right").unwrap(); + std::fs::write(left.path().join("winner"), b"left").unwrap(); + std::fs::write(right.path().join("winner"), b"right").unwrap(); + let barrier = Arc::new(Barrier::new(3)); + + let left_barrier = Arc::clone(&barrier); + let left_destination = destination.clone(); + let left_thread = std::thread::spawn(move || { + left_barrier.wait(); + left.publish(&left_destination).map(|publication| { + publication.commit(); + }) + }); + let right_barrier = Arc::clone(&barrier); + let right_destination = destination.clone(); + let right_thread = std::thread::spawn(move || { + right_barrier.wait(); + right.publish(&right_destination).map(|publication| { + publication.commit(); + }) + }); + barrier.wait(); + + let results = [left_thread.join().unwrap(), right_thread.join().unwrap()]; + assert_eq!(results.iter().filter(|result| result.is_ok()).count(), 1); + let loser = results + .iter() + .find_map(|result| result.as_ref().err()) + .unwrap(); + match loser { + crate::Error::Io(error) => { + assert_eq!(error.kind(), std::io::ErrorKind::AlreadyExists); + } + other => panic!("expected AlreadyExists, got {other:?}"), + } + assert!(matches!( + std::fs::read(destination.join("winner")) + .unwrap() + .as_slice(), + b"left" | b"right" + )); + let leftovers: Vec<_> = std::fs::read_dir(directory.path()) + .unwrap() + .filter_map(|entry| { + let entry = entry.unwrap(); + entry + .file_name() + .to_str() + .is_some_and(|name| name.starts_with(".target.race-")) + .then_some(entry.path()) + }) + .collect(); + assert!(leftovers.is_empty(), "stages survived: {leftovers:?}"); + } + + #[test] + fn rollback_failure_reports_that_the_published_target_may_remain() { + let directory = tempdir().unwrap(); + let destination = directory.path().join("target"); + let staged = StagedDirectory::create(&destination, "rollback").unwrap(); + std::fs::write(staged.path().join("data"), b"complete").unwrap(); + let publication = staged.publish(&destination).unwrap(); + std::fs::create_dir(&publication.hidden_path).unwrap(); + + let error = publication + .rollback_after(crate::Error::Io(std::io::Error::other("post-check failed"))); + match error { + crate::Error::SnapshotPublication { path, reason } => { + assert_eq!(path, destination); + assert!(reason.contains("post-check failed")); + assert!(reason.contains("rollback failed")); + } + other => panic!("expected an explicit rollback failure, got {other:?}"), + } + assert!(destination.join("data").exists()); + } +} diff --git a/src/storage.rs b/src/storage.rs index bd553a1..799faff 100644 --- a/src/storage.rs +++ b/src/storage.rs @@ -8,6 +8,7 @@ use serde::{Deserialize, Serialize}; use std::collections::{BTreeMap, HashMap, HashSet}; use std::fs::{File, OpenOptions, TryLockError}; use std::io::{Read, Seek, SeekFrom, Write}; +use std::os::unix::fs::MetadataExt; use std::path::Path; const DATABASE_MAGIC: &[u8; 8] = b"FTWDB001"; @@ -232,7 +233,7 @@ impl Database { fn open_mode(path: &Path, config: Config, read_only: bool) -> Result { validate_config(config)?; let mut file = if read_only { - OpenOptions::new().read(true).open(path)? + open_regular_file_read_only(path)? } else { OpenOptions::new() .create(true) @@ -965,6 +966,39 @@ fn scan_and_recover( }) } +/// Opens one existing regular file without following a final symlink or +/// blocking on a FIFO or device. The identity check closes the metadata/open +/// race for the final path component. +pub(crate) fn open_regular_file_read_only(path: &Path) -> Result { + use rustix::fs::{Mode, OFlags, open}; + + let path_metadata = std::fs::symlink_metadata(path)?; + if !path_metadata.file_type().is_file() { + return Err(Error::Io(std::io::Error::new( + std::io::ErrorKind::InvalidData, + format!("read-only path is not a regular file: {}", path.display()), + ))); + } + let descriptor = open( + path, + OFlags::RDONLY | OFlags::CLOEXEC | OFlags::NOFOLLOW | OFlags::NONBLOCK, + Mode::empty(), + ) + .map_err(|error| Error::Io(error.into()))?; + let file = File::from(descriptor); + let opened_metadata = file.metadata()?; + if !opened_metadata.file_type().is_file() + || opened_metadata.dev() != path_metadata.dev() + || opened_metadata.ino() != path_metadata.ino() + { + return Err(Error::Io(std::io::Error::new( + std::io::ErrorKind::InvalidData, + format!("read-only path changed while opening: {}", path.display()), + ))); + } + Ok(file) +} + /// Makes a directory's entries durable after creating, linking, renaming, or /// removing an entry. FTWDB v0.1 only builds on Unix because this guarantee /// depends on opening and syncing the directory itself. diff --git a/src/store.rs b/src/store.rs index c04e543..2037c1b 100644 --- a/src/store.rs +++ b/src/store.rs @@ -1,5 +1,9 @@ use crate::manifest::{self, Manifest, RollupDescriptor}; use crate::rollup::calendar_bucket_bounds; +use crate::snapshot::{ + PublicationStep, StagedDirectory, inject_checksum_mismatch, publication_checkpoint, + snapshot_digest, +}; use crate::storage::{sync_directory, sync_parent_directory}; use crate::transaction::Record; use crate::{ @@ -74,6 +78,17 @@ pub struct BackupReport { pub hard_link_fallback_error_kinds: Vec, } +#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)] +pub struct RestoreReport { + pub files: usize, + pub bytes: u64, + pub manifest_generation: u64, + pub raw_commits: u64, + pub raw_points: u64, + pub source_snapshot_crc32: u32, + pub destination_snapshot_crc32: u32, +} + impl BackupReport { fn record_copy(&mut self) { self.files += 1; @@ -175,6 +190,9 @@ impl Store { let root = path.as_ref().to_path_buf(); let manifest_directory = root.join(MANIFEST_DIRECTORY); let rollup_directory = root.join(ROLLUP_DIRECTORY); + require_real_directory(&root)?; + require_real_directory(&manifest_directory)?; + require_real_directory(&rollup_directory)?; let database = Database::open_read_only(root.join(ACTIVE_LOG))?; let manifest = Manifest::load(&manifest_directory)?; let mut store = Self { @@ -601,16 +619,6 @@ impl Store { pub fn backup_to(&mut self, destination: impl AsRef) -> Result { self.ensure_healthy()?; let destination = destination.as_ref(); - if destination.exists() { - return Err(Error::Io(std::io::Error::new( - std::io::ErrorKind::AlreadyExists, - "backup destination already exists", - ))); - } - let parent = destination.parent().ok_or(Error::InvalidConfig( - "backup destination must have a parent directory", - ))?; - std::fs::create_dir_all(parent)?; // Flushing is the only mutation `backup_to` performs on the source, // and a read-only handle has nothing buffered, so backups from a // read-only store are allowed and provably leave the source intact. @@ -619,36 +627,123 @@ impl Store { } self.check_integrity()?; - let name = destination - .file_name() - .and_then(|name| name.to_str()) - .ok_or(Error::InvalidConfig("backup destination must be UTF-8"))?; - let nonce = SystemTime::now() - .duration_since(UNIX_EPOCH) - .unwrap_or_default() - .as_nanos(); - let temporary = parent.join(format!(".{name}.backup-{}-{nonce}", std::process::id())); - std::fs::create_dir(&temporary)?; - let result = self.write_backup(&temporary); - let mut report = match result { - Ok(report) => report, + let staged = StagedDirectory::create(destination, "backup")?; + let mut report = self.write_snapshot(staged.path())?; + // Hold the stage lock through publication and the destination check. + // A writer therefore cannot change the active log between them. + let stage = Self::open_read_only(staged.path())?; + stage.check_integrity()?; + let publication = staged.publish(destination)?; + let checked = (|| { + publication_checkpoint(PublicationStep::PostCheck)?; + // The backup check must not reconcile a stale manifest or prune + // files from the copy it is meant to verify. + let backup = Self::open_read_only(destination)?; + backup.check_integrity()?; + report.bytes = directory_bytes(destination)?; + Ok(report) + })(); + match checked { + Ok(report) => { + drop(stage); + publication.commit(); + Ok(report) + } Err(error) => { - let _ = std::fs::remove_dir_all(&temporary); - return Err(error); + let error = publication.rollback_after(error); + drop(stage); + Err(error) } - }; - if let Err(error) = std::fs::rename(&temporary, destination) { - let _ = std::fs::remove_dir_all(&temporary); - return Err(Error::Io(error)); } - sync_directory(parent)?; + } - // Opening the published copy exercises raw recovery, manifest parsing, - // and every active segment before the caller treats it as a backup. - let backup = Self::open(destination)?; - backup.check_integrity()?; - report.bytes = directory_bytes(destination)?; - Ok(report) + /// Restores one fully valid, recovery-free snapshot into an absent target. + /// + /// The source stays open read-only under a shared lock. The target is built + /// and checked in a hidden sibling directory, compared with the source by + /// the versioned snapshot CRC32, and published with an atomic no-clobber + /// rename. No restore path overwrites an existing target. + pub fn restore_from( + backup: impl AsRef, + destination: impl AsRef, + ) -> Result { + let backup = Self::open_read_only(backup)?; + backup.restore_to(destination.as_ref()) + } + + fn restore_to(&self, destination: &Path) -> Result { + self.ensure_healthy()?; + let source_integrity = self.check_integrity()?; + self.require_clean_restore_source(&source_integrity)?; + let relative_paths = self.selected_snapshot_paths(); + let source_digest = snapshot_digest(&self.root, &relative_paths)?; + + let staged = StagedDirectory::create(destination, "restore")?; + self.write_snapshot(staged.path())?; + let stage_digest = + inject_checksum_mismatch(snapshot_digest(staged.path(), &relative_paths)?); + if stage_digest != source_digest { + return Err(snapshot_mismatch( + "source", + source_digest, + "stage", + stage_digest, + )); + } + + // Exercise raw replay, manifest selection, active rollup decoding, and + // descriptor checks before the directory gets a public name. + let stage = Self::open_read_only(staged.path())?; + let stage_integrity = stage.check_integrity()?; + stage.require_clean_restore_source(&stage_integrity)?; + + let publication = staged.publish(destination)?; + let checked = (|| { + publication_checkpoint(PublicationStep::PostCheck)?; + let restored = Self::open_read_only(destination)?; + let destination_integrity = restored.check_integrity()?; + restored.require_clean_restore_source(&destination_integrity)?; + let destination_digest = snapshot_digest(destination, &relative_paths)?; + if destination_digest != source_digest { + return Err(snapshot_mismatch( + "source", + source_digest, + "destination", + destination_digest, + )); + } + if destination_integrity.raw_commits != source_integrity.raw_commits + || destination_integrity.raw_points != source_integrity.raw_points + || destination_integrity.manifest_generation != source_integrity.manifest_generation + { + return Err(Error::Corruption { + offset: 0, + reason: "restored store counts differ from the checked backup".to_owned(), + }); + } + + Ok(RestoreReport { + files: destination_digest.files, + bytes: destination_digest.bytes, + manifest_generation: destination_integrity.manifest_generation, + raw_commits: destination_integrity.raw_commits, + raw_points: destination_integrity.raw_points, + source_snapshot_crc32: source_digest.crc32, + destination_snapshot_crc32: destination_digest.crc32, + }) + })(); + match checked { + Ok(report) => { + drop(stage); + publication.commit(); + Ok(report) + } + Err(error) => { + let error = publication.rollback_after(error); + drop(stage); + Err(error) + } + } } #[must_use] @@ -714,7 +809,49 @@ impl Store { Ok(buckets) } - fn write_backup(&self, temporary: &Path) -> Result { + fn selected_snapshot_paths(&self) -> Vec { + let mut files = vec![ACTIVE_LOG.to_owned()]; + files.extend( + self.active_rollups() + .map(|descriptor| format!("{ROLLUP_DIRECTORY}/{}", descriptor.file)), + ); + if self.manifest.generation > 0 { + files.push(format!( + "{MANIFEST_DIRECTORY}/MANIFEST.{:020}", + self.manifest.generation + )); + } + files.sort(); + files.dedup(); + files + } + + fn require_clean_restore_source(&self, report: &IntegrityReport) -> Result<()> { + if report.stale_rollup_files > 0 { + return Err(Error::Corruption { + offset: 0, + reason: format!( + "restore requires a clean backup; {} active rollup files trail the raw log", + report.stale_rollup_files + ), + }); + } + if report.raw_recovered_tail_bytes == 0 + && report.raw_recovered_tail == crate::RecoveredTail::None + { + return Ok(()); + } + let physical_bytes = std::fs::metadata(self.root.join(ACTIVE_LOG))?.len(); + Err(Error::Corruption { + offset: physical_bytes.saturating_sub(report.raw_recovered_tail_bytes), + reason: format!( + "restore requires a recovery-free backup; raw log has {} bytes of {}", + report.raw_recovered_tail_bytes, report.raw_recovered_tail + ), + }) + } + + fn write_snapshot(&self, temporary: &Path) -> Result { let manifests = temporary.join(MANIFEST_DIRECTORY); let rollups = temporary.join(ROLLUP_DIRECTORY); std::fs::create_dir(&manifests)?; @@ -726,6 +863,7 @@ impl Store { // Never hard-link the active log: future appends to the source inode // must not mutate the backup snapshot. + publication_checkpoint(PublicationStep::Copy)?; copy_and_sync(&self.root.join(ACTIVE_LOG), &temporary.join(ACTIVE_LOG))?; report.record_copy(); for descriptor in self.active_rollups() { @@ -741,6 +879,7 @@ impl Store { hard_link_or_copy(&self.manifest_directory.join(&file), &manifests.join(&file))?; report.record_link_or_copy(outcome); } + publication_checkpoint(PublicationStep::Sync)?; sync_directory(&manifests)?; sync_directory(&rollups)?; sync_directory(temporary)?; @@ -1184,9 +1323,35 @@ fn directory_bytes(path: &Path) -> Result { Ok(total) } +fn require_real_directory(path: &Path) -> Result<()> { + if std::fs::symlink_metadata(path)?.file_type().is_dir() { + return Ok(()); + } + Err(Error::Io(std::io::Error::new( + std::io::ErrorKind::InvalidData, + format!("snapshot path is not a real directory: {}", path.display()), + ))) +} + +fn snapshot_mismatch( + left_name: &str, + left: crate::snapshot::SnapshotDigest, + right_name: &str, + right: crate::snapshot::SnapshotDigest, +) -> Error { + Error::Corruption { + offset: 0, + reason: format!( + "snapshot mismatch: {left_name} files={} bytes={} crc32={:08x}, {right_name} files={} bytes={} crc32={:08x}", + left.files, left.bytes, left.crc32, right.files, right.bytes, right.crc32 + ), + } +} + #[cfg(test)] mod tests { use super::{BackupReport, LinkOrCopy, RollupSource, Store, hard_link_or_copy_with}; + use crate::snapshot::{PublicationStep, StagedDirectory, fail_next_publication_step}; use crate::{ CalendarUnit, Entity, EntityId, Point, RollupPolicy, RollupResolution, RollupTier, SeriesDefinition, SeriesSemantics, Transaction, @@ -1354,6 +1519,143 @@ mod tests { .collect() } + fn create_verified_backup(root: &Path, with_rollup: bool) -> std::path::PathBuf { + let source = root.join("source"); + let backup = root.join("backup"); + let mut store = Store::open(&source).unwrap(); + let tiers = with_rollup + .then_some(vec![RollupTier { + resolution: RollupResolution::FixedMicros(5 * SECOND), + retain_for_micros: None, + }]) + .unwrap_or_default(); + initialize(&mut store, tiers, None); + let mut transaction = Transaction::new(); + transaction.append_points(points()); + store.commit(transaction).unwrap(); + if with_rollup { + store.maintain(DAY).unwrap(); + } + store.backup_to(&backup).unwrap(); + backup + } + + fn assert_io_kind(error: crate::Error, expected: std::io::ErrorKind) { + match error { + crate::Error::Io(error) => assert_eq!(error.kind(), expected), + other => panic!("expected I/O error {expected:?}, got {other:?}"), + } + } + + #[cfg(target_os = "linux")] + fn create_fifo(path: &Path) { + rustix::fs::mkfifoat( + rustix::fs::CWD, + path, + rustix::fs::Mode::RUSR | rustix::fs::Mode::WUSR, + ) + .unwrap(); + } + + #[cfg(target_os = "macos")] + fn create_fifo(path: &Path) { + use std::os::unix::ffi::OsStrExt; + + let path = std::ffi::CString::new(path.as_os_str().as_bytes()).unwrap(); + // SAFETY: `path` is a live NUL-terminated string and mode has no + // platform-dependent bits beyond user read/write permissions. + let result = unsafe { libc::mkfifo(path.as_ptr(), 0o600) }; + if result != 0 { + panic!("mkfifo failed: {}", std::io::Error::last_os_error()); + } + } + + #[cfg(not(any(target_os = "linux", target_os = "macos")))] + fn create_fifo(_path: &Path) { + panic!("FIFO restore tests require Linux or macOS"); + } + + fn restore_with_deadline( + backup: std::path::PathBuf, + target: std::path::PathBuf, + fifo: &Path, + ) -> crate::Result { + let (sender, receiver) = std::sync::mpsc::channel(); + let worker = std::thread::spawn(move || { + sender.send(Store::restore_from(backup, target)).unwrap(); + }); + let mut timed_out = false; + let result = match receiver.recv_timeout(std::time::Duration::from_secs(5)) { + Ok(result) => result, + Err(std::sync::mpsc::RecvTimeoutError::Timeout) => { + timed_out = true; + let writer = rustix::fs::open( + fifo, + rustix::fs::OFlags::WRONLY + | rustix::fs::OFlags::CLOEXEC + | rustix::fs::OFlags::NONBLOCK, + rustix::fs::Mode::empty(), + ) + .expect("a blocked FIFO reader must admit a nonblocking writer"); + drop(writer); + receiver + .recv_timeout(std::time::Duration::from_secs(5)) + .expect("restore worker must exit after its FIFO reader is released") + } + Err(std::sync::mpsc::RecvTimeoutError::Disconnected) => { + panic!("restore worker disconnected without a result") + } + }; + worker.join().unwrap(); + assert!(!timed_out, "restore blocked while opening a selected FIFO"); + result + } + + fn flip_last_byte(path: &Path) { + let mut file = std::fs::OpenOptions::new() + .read(true) + .write(true) + .open(path) + .unwrap(); + file.seek(SeekFrom::End(-1)).unwrap(); + let mut byte = [0_u8; 1]; + std::io::Read::read_exact(&mut file, &mut byte).unwrap(); + file.seek(SeekFrom::End(-1)).unwrap(); + file.write_all(&[byte[0] ^ 0xff]).unwrap(); + file.sync_all().unwrap(); + } + + fn overwrite_byte(path: &Path, offset: u64, byte: u8) { + let mut file = std::fs::OpenOptions::new().write(true).open(path).unwrap(); + file.seek(SeekFrom::Start(offset)).unwrap(); + file.write_all(&[byte]).unwrap(); + file.sync_all().unwrap(); + } + + fn restore_stages(parent: &Path, target_name: &str) -> Vec { + let prefix = format!(".{target_name}.restore-"); + stored_files(parent) + .into_iter() + .filter(|name| name.starts_with(&prefix)) + .collect() + } + + fn active_manifest_path(backup: &Path) -> std::path::PathBuf { + let store = Store::open_read_only(backup).unwrap(); + let generation = store.manifest_generation(); + drop(store); + backup + .join("manifests") + .join(format!("MANIFEST.{generation:020}")) + } + + fn active_rollup_path(backup: &Path) -> std::path::PathBuf { + let store = Store::open_read_only(backup).unwrap(); + let file = store.active_rollups().next().unwrap().file.clone(); + drop(store); + backup.join("rollups").join(file) + } + #[test] fn open_creates_and_reopens_a_nested_root() { let directory = tempdir().unwrap(); @@ -2075,4 +2377,325 @@ mod tests { assert_eq!(backup.database().stats().unwrap().points, backup_points); assert_eq!(store.database().stats().unwrap().points, backup_points + 1); } + + #[test] + fn restore_preserves_the_selected_snapshot_and_is_independent() { + let directory = tempdir().unwrap(); + let backup = create_verified_backup(directory.path(), true); + let target = directory.path().join("restored"); + let backup_before = directory_snapshot(&backup); + + let report = Store::restore_from(&backup, &target).unwrap(); + assert_eq!(report.files, 3); + assert!(report.bytes > 0); + assert_eq!(report.raw_commits, 2); + assert_eq!(report.raw_points, 21); + assert_eq!( + report.source_snapshot_crc32, + report.destination_snapshot_crc32 + ); + let restored = Store::open_read_only(&target).unwrap(); + let integrity = restored.check_integrity().unwrap(); + assert_eq!(integrity.raw_commits, report.raw_commits); + assert_eq!(integrity.raw_points, report.raw_points); + assert_eq!(integrity.manifest_generation, report.manifest_generation); + drop(restored); + + let mut writable = Store::open(&target).unwrap(); + let mut transaction = Transaction::new(); + transaction.append_points(vec![Point::actual(1, 30 * SECOND, 30.0)]); + writable.commit(transaction).unwrap(); + writable.close().unwrap(); + assert_eq!(directory_snapshot(&backup), backup_before); + assert_eq!( + Store::open_read_only(&backup) + .unwrap() + .check_integrity() + .unwrap() + .raw_points, + 21 + ); + } + + #[test] + fn restore_never_changes_an_existing_empty_or_working_target() { + let directory = tempdir().unwrap(); + let backup = create_verified_backup(directory.path(), false); + + let empty = directory.path().join("empty"); + std::fs::create_dir(&empty).unwrap(); + let empty_before = directory_snapshot(&empty); + assert_io_kind( + Store::restore_from(&backup, &empty).unwrap_err(), + std::io::ErrorKind::AlreadyExists, + ); + assert_eq!(directory_snapshot(&empty), empty_before); + + let working = directory.path().join("working"); + let mut store = Store::open(&working).unwrap(); + initialize(&mut store, Vec::new(), None); + store.close().unwrap(); + let working_before = directory_snapshot(&working); + assert_io_kind( + Store::restore_from(&backup, &working).unwrap_err(), + std::io::ErrorKind::AlreadyExists, + ); + assert_eq!(directory_snapshot(&working), working_before); + + let dangling = directory.path().join("dangling"); + std::os::unix::fs::symlink("missing-target", &dangling).unwrap(); + assert_io_kind( + Store::restore_from(&backup, &dangling).unwrap_err(), + std::io::ErrorKind::AlreadyExists, + ); + assert_eq!( + std::fs::read_link(&dangling).unwrap(), + Path::new("missing-target") + ); + } + + #[test] + fn simultaneous_restores_publish_exactly_one_complete_target() { + let directory = tempdir().unwrap(); + let backup = create_verified_backup(directory.path(), true); + let target = directory.path().join("race-target"); + let barrier = std::sync::Arc::new(std::sync::Barrier::new(3)); + + let left_backup = backup.clone(); + let left_target = target.clone(); + let left_barrier = std::sync::Arc::clone(&barrier); + let left = std::thread::spawn(move || { + left_barrier.wait(); + Store::restore_from(left_backup, left_target) + }); + let right_backup = backup.clone(); + let right_target = target.clone(); + let right_barrier = std::sync::Arc::clone(&barrier); + let right = std::thread::spawn(move || { + right_barrier.wait(); + Store::restore_from(right_backup, right_target) + }); + barrier.wait(); + + let results = [left.join().unwrap(), right.join().unwrap()]; + assert_eq!(results.iter().filter(|result| result.is_ok()).count(), 1); + let loser = results + .into_iter() + .find_map(std::result::Result::err) + .unwrap(); + assert_io_kind(loser, std::io::ErrorKind::AlreadyExists); + let restored = Store::open_read_only(&target).unwrap(); + let integrity = restored.check_integrity().unwrap(); + assert_eq!(integrity.raw_commits, 2); + assert_eq!(integrity.raw_points, 21); + assert_eq!(integrity.active_rollup_files, 1); + assert!(restore_stages(directory.path(), "race-target").is_empty()); + } + + #[test] + fn stage_shared_lock_survives_rename_until_the_post_check_finishes() { + let directory = tempdir().unwrap(); + let backup = create_verified_backup(directory.path(), true); + let target = directory.path().join("target"); + let source = Store::open_read_only(&backup).unwrap(); + let staged = StagedDirectory::create(&target, "restore").unwrap(); + source.write_snapshot(staged.path()).unwrap(); + let stage = Store::open_read_only(staged.path()).unwrap(); + + let publication = staged.publish(&target).unwrap(); + match Store::open(&target) { + Err(crate::Error::Locked { path }) => { + assert_eq!(path, target.join("active.wlog")); + } + Err(other) => panic!("expected writer lock refusal, got {other:?}"), + Ok(_) => panic!("writer opened before the post-check finished"), + } + Store::open_read_only(&target) + .unwrap() + .check_integrity() + .unwrap(); + drop(stage); + publication.commit(); + + Store::open(&target).unwrap().close().unwrap(); + } + + #[test] + fn restore_rejects_corrupt_and_incomplete_selected_files() { + for case in ["raw-crc", "raw-format", "raw-short", "manifest", "rollup"] { + let directory = tempdir().unwrap(); + let backup = create_verified_backup(directory.path(), true); + let target = directory.path().join("target"); + match case { + "raw-crc" => flip_last_byte(&backup.join("active.wlog")), + "raw-format" => overwrite_byte(&backup.join("active.wlog"), 0, 0), + "raw-short" => { + let mut file = std::fs::OpenOptions::new() + .append(true) + .open(backup.join("active.wlog")) + .unwrap(); + file.write_all(b"partial").unwrap(); + file.sync_all().unwrap(); + } + "manifest" => flip_last_byte(&active_manifest_path(&backup)), + "rollup" => flip_last_byte(&active_rollup_path(&backup)), + _ => unreachable!(), + } + + assert!( + Store::restore_from(&backup, &target).is_err(), + "restore accepted {case} corruption" + ); + assert!(!target.exists(), "restore published {case} corruption"); + assert!(restore_stages(directory.path(), "target").is_empty()); + } + } + + #[test] + fn restore_rejects_stale_active_rollup_provenance() { + let directory = tempdir().unwrap(); + let backup = create_verified_backup(directory.path(), true); + { + let mut database = crate::Database::open(backup.join("active.wlog")).unwrap(); + let mut transaction = Transaction::new(); + transaction.append_points(vec![Point::actual(1, 30 * SECOND, 30.0)]); + database.commit(transaction).unwrap(); + database.close().unwrap(); + } + assert_eq!( + Store::open_read_only(&backup) + .unwrap() + .check_integrity() + .unwrap() + .stale_rollup_files, + 1 + ); + + let target = directory.path().join("target"); + let error = Store::restore_from(&backup, &target).unwrap_err(); + assert!(error.to_string().contains("active rollup files trail")); + assert!(!target.exists()); + assert!(restore_stages(directory.path(), "target").is_empty()); + } + + #[test] + fn restore_ignores_orphan_rollups_and_inactive_manifest_generations() { + let directory = tempdir().unwrap(); + let backup = create_verified_backup(directory.path(), true); + let first = directory.path().join("first"); + let baseline = Store::restore_from(&backup, &first).unwrap(); + + std::fs::write(backup.join("rollups/orphan.rseg"), b"not selected").unwrap(); + std::fs::write( + backup.join("manifests/MANIFEST.00000000000000000000"), + b"inactive generation", + ) + .unwrap(); + let second = directory.path().join("second"); + let restored = Store::restore_from(&backup, &second).unwrap(); + + assert_eq!( + restored.source_snapshot_crc32, + baseline.source_snapshot_crc32 + ); + assert_eq!( + restored.destination_snapshot_crc32, + baseline.destination_snapshot_crc32 + ); + assert!(!second.join("rollups/orphan.rseg").exists()); + assert!( + !second + .join("manifests/MANIFEST.00000000000000000000") + .exists() + ); + } + + #[test] + fn restore_rejects_a_selected_symlink() { + let directory = tempdir().unwrap(); + let backup = create_verified_backup(directory.path(), false); + let external = directory.path().join("external.wlog"); + std::fs::rename(backup.join("active.wlog"), &external).unwrap(); + std::os::unix::fs::symlink(&external, backup.join("active.wlog")).unwrap(); + + let target = directory.path().join("target"); + let error = Store::restore_from(&backup, &target).unwrap_err(); + assert!(error.to_string().contains("not a regular file")); + assert!(!target.exists()); + assert!(restore_stages(directory.path(), "target").is_empty()); + } + + #[test] + fn restore_rejects_selected_special_files_without_blocking() { + for selected in ["active", "manifest", "rollup"] { + let directory = tempdir().unwrap(); + let backup = create_verified_backup(directory.path(), true); + let path = match selected { + "active" => backup.join("active.wlog"), + "manifest" => active_manifest_path(&backup), + "rollup" => active_rollup_path(&backup), + _ => unreachable!(), + }; + std::fs::remove_file(&path).unwrap(); + create_fifo(&path); + + let target = directory.path().join("target"); + let error = restore_with_deadline(backup, target.clone(), &path).unwrap_err(); + assert!( + error.to_string().contains("not a regular file"), + "wrong {selected} special-file error: {error}" + ); + assert!(!target.exists()); + assert!(restore_stages(directory.path(), "target").is_empty()); + } + } + + #[test] + fn restore_failures_clean_the_stage_and_leave_no_target() { + for step in [ + PublicationStep::Copy, + PublicationStep::Sync, + PublicationStep::ChecksumMismatch, + PublicationStep::Publish, + PublicationStep::ParentSync, + PublicationStep::PostCheck, + ] { + let directory = tempdir().unwrap(); + let backup = create_verified_backup(directory.path(), true); + let target = directory.path().join("target"); + fail_next_publication_step(step); + + assert!( + Store::restore_from(&backup, &target).is_err(), + "injected {step:?} failure did not fail" + ); + assert!(!target.exists(), "{step:?} failure left a target"); + assert!( + restore_stages(directory.path(), "target").is_empty(), + "{step:?} failure left a stage" + ); + } + } + + #[test] + fn backup_rolls_back_final_parent_sync_and_post_check_failures() { + for step in [PublicationStep::ParentSync, PublicationStep::PostCheck] { + let directory = tempdir().unwrap(); + let source = directory.path().join("source"); + let target = directory.path().join("backup"); + let mut store = Store::open(&source).unwrap(); + initialize(&mut store, Vec::new(), None); + fail_next_publication_step(step); + + assert!(store.backup_to(&target).is_err()); + assert!(!target.exists(), "{step:?} failure left a backup"); + let prefix = ".backup.backup-"; + assert!( + stored_files(directory.path()) + .iter() + .all(|name| !name.starts_with(prefix)), + "{step:?} failure left a backup stage" + ); + } + } } diff --git a/tests/cli.rs b/tests/cli.rs index 4a71f66..31a65fb 100644 --- a/tests/cli.rs +++ b/tests/cli.rs @@ -190,6 +190,9 @@ fn usage_errors_exit_two_on_stderr_without_creating_files() { &["backup"], &["backup", "source"], &["backup", "source", "destination", "extra"], + &["restore"], + &["restore", "backup"], + &["restore", "backup", "target", "extra"], &["generate"], &["bench-ftwdb"], &["bench-ftwdb", "workload"], @@ -400,6 +403,7 @@ fn generated_store_commands_round_trip_without_read_only_mutation() { let workload = directory.path().join("workload"); let store = directory.path().join("store"); let backup = directory.path().join("backup"); + let restored = directory.path().join("restored"); let generated = ftw( directory.path(), @@ -524,6 +528,103 @@ fn generated_store_commands_round_trip_without_read_only_mutation() { generated_commits, ); assert_eq!(snapshot_tree(&backup), before_backup_check); + + let before_restore = snapshot_tree(&backup); + let restore = ftw( + directory.path(), + &["restore", path_str(&backup), path_str(&restored)], + ); + restore.assert_success(); + let restore_json = json_record(&restore); + assert_eq!(json_string(&restore_json, "format"), "ftwdb-restore-v1"); + assert_eq!(json_u64(&restore_json, "files"), files); + assert_eq!(json_u64(&restore_json, "raw_points"), generated_points); + assert_eq!(json_u64(&restore_json, "raw_commits"), generated_commits); + assert_eq!( + json_u64(&restore_json, "manifest_generation"), + manifest_generation + ); + let source_crc = json_string(&restore_json, "source_snapshot_crc32"); + let destination_crc = json_string(&restore_json, "destination_snapshot_crc32"); + assert_eq!(source_crc.len(), 8); + assert_eq!(source_crc, destination_crc); + let restored_tree = snapshot_tree(&restored); + assert_eq!(restored_tree, before_restore); + let restored_bytes: u64 = restored_tree + .iter() + .filter_map(|(_, entry)| match entry { + TreeEntry::File(bytes) => Some(bytes.len() as u64), + TreeEntry::Directory => None, + }) + .sum(); + assert_eq!(json_u64(&restore_json, "bytes"), restored_bytes); + check_store( + directory.path(), + &restored, + generated_points, + generated_commits, + ); + assert_eq!(snapshot_tree(&backup), before_restore); + + let existing = directory.path().join("existing-target"); + fs::create_dir(&existing).unwrap(); + let existing_before = snapshot_tree(&existing); + let refused = ftw( + directory.path(), + &["restore", path_str(&backup), path_str(&existing)], + ); + assert_runtime_error(&refused); + assert!(refused.stderr.contains("already exists")); + assert_eq!(snapshot_tree(&existing), existing_before); + + let relative = ftw( + directory.path(), + &["restore", path_str(&backup), "relative-restored"], + ); + relative.assert_success(); + check_store( + directory.path(), + &directory.path().join("relative-restored"), + generated_points, + generated_commits, + ); +} + +#[test] +fn restore_corruption_is_a_runtime_error_and_publishes_nothing() { + let directory = tempfile::tempdir().unwrap(); + let source = directory.path().join("source"); + let backup = directory.path().join("backup"); + let target = directory.path().join("target"); + + let generated = ftw( + directory.path(), + &["generate", "workload", "--sites", "1", "--days", "1"], + ); + generated.assert_success(); + let loaded = ftw( + directory.path(), + &["bench-ftwdb", "workload", path_str(&source)], + ); + loaded.assert_success(); + let backed_up = ftw( + directory.path(), + &["backup", path_str(&source), path_str(&backup)], + ); + backed_up.assert_success(); + let mut active = fs::OpenOptions::new() + .append(true) + .open(backup.join("active.wlog")) + .unwrap(); + active.write_all(b"partial").unwrap(); + active.sync_all().unwrap(); + + let output = ftw( + directory.path(), + &["restore", path_str(&backup), path_str(&target)], + ); + assert_runtime_error(&output); + assert!(!target.exists()); } #[test] diff --git a/tests/real_fixture.rs b/tests/real_fixture.rs index 390b6b4..95da71e 100644 --- a/tests/real_fixture.rs +++ b/tests/real_fixture.rs @@ -9,8 +9,11 @@ fn committed_real_fixture_loads_with_stable_identity() { let fixture = PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("bench/fixtures/ftw-real-v1/points.csv.gz"); let directory = tempfile::tempdir().unwrap(); + let source = directory.path().join("source"); + let backup = directory.path().join("backup"); + let restored = directory.path().join("restored"); let mut store = Store::open_with( - directory.path(), + &source, Config { durability: Durability::Manual, ..Config::default() @@ -33,4 +36,20 @@ fn committed_real_fixture_loads_with_stable_identity() { let integrity = store.check_integrity().unwrap(); assert_eq!(integrity.raw_points, 889_978); assert_eq!(integrity.raw_commits, 89); + + store.backup_to(&backup).unwrap(); + let restore = Store::restore_from(&backup, &restored).unwrap(); + assert_eq!(restore.raw_points, 889_978); + assert_eq!(restore.raw_commits, 89); + assert_eq!(restore.source_snapshot_crc32, 0x9ff1_a95b); + assert_eq!( + restore.source_snapshot_crc32, + restore.destination_snapshot_crc32 + ); + let restored_integrity = Store::open_read_only(&restored) + .unwrap() + .check_integrity() + .unwrap(); + assert_eq!(restored_integrity.raw_points, 889_978); + assert_eq!(restored_integrity.raw_commits, 89); }