From 0738c89b7169f1743632de82ceadf69d97e3b1e1 Mon Sep 17 00:00:00 2001 From: Fredrik Ahlgren Date: Tue, 21 Jul 2026 16:56:36 +0200 Subject: [PATCH] Add strict salvage workflow --- README.md | 5 +- bench/sd-card-emulator/README.md | 5 +- bench/sd-card-emulator/linux-smoke.sh | 26 + docs/operations.md | 79 ++- docs/roadmap.md | 12 +- src/bin/ftw.rs | 59 +- src/error.rs | 8 + src/lib.rs | 7 +- src/snapshot.rs | 57 +- src/storage.rs | 762 +++++++++++++++++++++++++- src/store.rs | 586 +++++++++++++++++++- tests/cli.rs | 105 ++++ tests/real_fixture.rs | 22 +- 13 files changed, 1677 insertions(+), 56 deletions(-) diff --git a/README.md b/README.md index d71b53f..c2a8772 100644 --- a/README.md +++ b/README.md @@ -56,12 +56,13 @@ Inspect a database file: cargo run --release -- inspect ./data.ftwdb ``` -Verify or snapshot a directory store: +Verify, snapshot, restore, or salvage 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 +cargo run --release -- salvage ./damaged-energy.ftwdb ./salvaged-energy.ftwdb ``` ## Design documents @@ -71,7 +72,7 @@ cargo run --release -- restore ./backups/energy-2026-07-21.ftwdb ./restored-ener - [Storage format v1](docs/format.md) - [Immutable segment format](docs/segment-format.md) - [Persistent rollups and retention](docs/rollups.md) -- [Integrity checks, backup, and restore](docs/operations.md) +- [Integrity checks, backup, restore, and salvage](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 61fbfcb..be6a3e7 100644 --- a/bench/sd-card-emulator/README.md +++ b/bench/sd-card-emulator/README.md @@ -72,7 +72,10 @@ 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. +log SHA-256 values. It then adds a seven-byte short tail to an off-card copy, +salvages it to another new store on the card, and requires a `partial` result, +seven discarded bytes, 889,978 points, 89 commits, equal snapshot CRCs, and the +clean raw-log SHA-256 value. ```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 5eb4110..9340e5e 100755 --- a/bench/sd-card-emulator/linux-smoke.sh +++ b/bench/sd-card-emulator/linux-smoke.sh @@ -73,6 +73,18 @@ test "$before_hash" = "$after_hash" sha256sum "$out/restore-backup/active.wlog" >"$out/restore-backup-sha.txt" sha256sum "$mount_dir/restored-database/active.wlog" >"$out/restore-target-sha.txt" +# Keep the damaged source off the emulated card. A short tail must yield a +# verified partial store on the card without changing the clean raw prefix. +cp -a "$out/restore-backup" "$out/salvage-source" +printf 'partial' >>"$out/salvage-source/active.wlog" +sync +"$ftw" salvage "$out/salvage-source" "$mount_dir/salvaged-database" \ + >"$out/salvage.json" +"$ftw" check-store "$mount_dir/salvaged-database" \ + >"$out/salvage-check.json" +sha256sum "$mount_dir/salvaged-database/active.wlog" \ + >"$out/salvage-target-sha.txt" + json_u64() { local file=$1 local field=$2 @@ -103,12 +115,24 @@ 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")" +test "$(json_string "$out/salvage.json" status)" = partial +test "$(json_string "$out/salvage.json" stop_reason)" = incomplete-frame-header +test "$(json_u64 "$out/salvage.json" discarded_bytes)" = 7 +test "$(json_u64 "$out/salvage.json" recovered_points)" = 889978 +test "$(json_u64 "$out/salvage.json" recovered_commits)" = 89 +test "$(json_u64 "$out/salvage-check.json" raw_points)" = 889978 +test "$(json_u64 "$out/salvage-check.json" raw_commits)" = 89 +test "$(json_string "$out/salvage.json" source_prefix_crc32)" = \ + "$(json_string "$out/salvage.json" destination_snapshot_crc32)" +test "$(cut -d' ' -f1 "$out/restore-backup-sha.txt")" = \ + "$(cut -d' ' -f1 "$out/salvage-target-sha.txt")" linux_nbd_finish trap - EXIT printf 'linux_nbd_smoke=passed\n' printf 'restore_drill=passed\n' +printf 'salvage_drill=passed\n' printf 'fsck_exit=%d\n' "$fsck_exit" printf 'active_log_sha256=%s\n' "$after_hash" cat "$out/load.json" @@ -116,3 +140,5 @@ cat "$out/check-after.json" cat "$out/verification.json" cat "$out/restore.json" cat "$out/restore-check.json" +cat "$out/salvage.json" +cat "$out/salvage-check.json" diff --git a/docs/operations.md b/docs/operations.md index d6f0f77..e88189a 100644 --- a/docs/operations.md +++ b/docs/operations.md @@ -1,4 +1,4 @@ -# Integrity checks, backup, and restore +# Integrity checks, backup, restore, and salvage ## Integrity check @@ -109,7 +109,70 @@ 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. +on the result. + +## Strict salvage + +`ftw salvage ` copies the longest valid raw-log +prefix into a new store. The command exits with code 2 for missing or extra +arguments. A valid database header followed by a damaged frame produces a +verified `partial` result and exit code 0. Header, source-open, lock, +source-race, stage-check, or publication errors use exit code 1. The command +never changes the source. It has no replace option, and it leaves any existing +target, including an empty directory or dangling symlink, unchanged. + +Salvage opens the source directory without following a symlink, then opens +only `active.wlog` relative to that directory with no-follow and nonblocking +flags. Both paths must keep the same file identity during the run, and +`active.wlog` must be a regular file. The command takes a shared lock on that +file. It does not read a manifest or rollup, so damaged, stale, and orphan +derived files do not affect the recovered raw prefix. + +The scanner first requires a valid FTWDB-v1 database header. It then validates +each frame in order: bounds, kind and version, header and payload CRC, payload +duplicate commit IDs, decode, catalog and transaction rules, and point limits. +It stops before the first failed frame. It never searches for a later `WBAT` +marker, and it never copies frames after a failure. This means that salvage +cannot recover sound data that happens to follow damaged data. A complete last +frame with a bad CRC counts as damage; salvage cannot prove that an interrupted +write caused it. + +The fixed `stop_reason` values are: + +- `clean-eof`; +- `incomplete-frame-header` and `incomplete-frame-payload`; +- `invalid-frame-magic`, `unsupported-frame-version`, and + `frame-header-checksum-mismatch`; +- `invalid-legacy-frame-size`, `transaction-frame-too-large`, + `identified-transaction-too-short`, and `unknown-frame-kind`; +- `payload-checksum-mismatch`, `duplicate-commit-id`, + `invalid-transaction`, `transaction-point-count-too-large`, and + `invalid-catalog-transaction`. + +After a valid database header, each listed frame or transaction fault yields a +`partial` result, including a fault in the first frame. A header-only source is +`clean`; a bad first frame can produce a checked header-only target with zero +commits and points. An invalid or unsupported database header is fatal and +publishes no target. + +The new store contains the header and validated frames as `active.wlog`, plus +empty `manifests` and `rollups` directories. Salvage uses the same hidden-stage, +file sync, directory sync, identity rollback, and atomic no-clobber publication +as restore. It opens both the stage and target read-only and runs the full store +check with no recovery. The shared stage lock stays held through publication, +the target check, and any rollback. + +The source-prefix CRC32 uses the restore snapshot domain and one relative path, +`active.wlog`. It covers that path length and bytes, the recovered prefix +length, and the exact prefix bytes. The destination snapshot covers the same +path and bytes. Salvage compares the values before and after publication. CRC32 +detects many accidental changes but does not prove authenticity. + +On success, `ftwdb-salvage-v1` reports `status`, `source_bytes`, +`recovered_prefix_bytes`, `discarded_bytes`, `stop_offset`, `stop_reason`, +`recovered_commits`, `recovered_points`, `source_prefix_crc32`, and +`destination_snapshot_crc32`. `status` is `clean` only at a checked frame +boundary at end of file; otherwise it is `partial`. ## Sync and full-disk checks @@ -132,14 +195,16 @@ 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, backup, and -strict restore 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, +strict restore, and salvage 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. 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. +corruption refusal, and no-clobber behavior. The salvage checks cover clean and +partial results, every frame and transaction stop class, no resync, source +identity, special files, concurrent no-clobber publication, rollback, and the +sanitized fixture. Physical SD-card tests remain an M4 release gate. diff --git a/docs/roadmap.md b/docs/roadmap.md index 4852a24..6aa72bb 100644 --- a/docs/roadmap.md +++ b/docs/roadmap.md @@ -88,9 +88,13 @@ 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 +verified restore to a new target on the emulated card. Strict salvage now reads +only the raw log, stops at the first invalid frame without resync, ignores all +derived files, and publishes a checked prefix through the same atomic +no-clobber path. The sanitized fixture fixes clean salvage at 889,978 points, +89 commits, and snapshot CRC32 `9ff1a95b`; the NBD run also checks a seven-byte +torn suffix and a verified partial target. 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, 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, 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 87a8bef..2f73daf 100644 --- a/src/bin/ftw.rs +++ b/src/bin/ftw.rs @@ -1,8 +1,8 @@ use flate2::read::GzDecoder; use ftwdb::{ BackupReport, Config, Database, Durability, EnergyWorkload, Error, RestoreReport, - RollupResolution, Store, Transaction, WorkloadConfig, gauge_bucket_checksum, load_real_fixture, - load_tsbs_iot, + RollupResolution, SalvageReport, Store, Transaction, WorkloadConfig, gauge_bucket_checksum, + load_real_fixture, load_tsbs_iot, }; use std::env; use std::fs::File; @@ -42,6 +42,7 @@ fn main() -> ExitCode { Some("check-store") => check_store(&arguments[2..]), Some("backup") => backup(&arguments[2..]), Some("restore") => restore(&arguments[2..]), + Some("salvage") => salvage(&arguments[2..]), Some("generate") => generate(&arguments[2..]), Some("bench-ftwdb") => bench_ftwdb(&arguments[2..]), Some("bench-real-fixture") => bench_real_fixture(&arguments[2..]), @@ -318,6 +319,33 @@ fn restore_report_json(report: &RestoreReport) -> String { ) } +fn salvage(arguments: &[String]) -> CliResult<()> { + if arguments.len() != 2 { + return Err(usage_error( + "salvage requires a damaged store and absent target directory", + )); + } + let report = Store::salvage_from(&arguments[0], &arguments[1])?; + println!("{}", salvage_report_json(&report)); + Ok(()) +} + +fn salvage_report_json(report: &SalvageReport) -> String { + format!( + "{{\"format\":\"ftwdb-salvage-v1\",\"status\":\"{}\",\"source_bytes\":{},\"recovered_prefix_bytes\":{},\"discarded_bytes\":{},\"stop_offset\":{},\"stop_reason\":\"{}\",\"recovered_commits\":{},\"recovered_points\":{},\"source_prefix_crc32\":\"{:08x}\",\"destination_snapshot_crc32\":\"{:08x}\"}}", + report.status, + report.source_bytes, + report.recovered_prefix_bytes, + report.discarded_bytes, + report.stop_offset, + report.stop_reason, + report.recovered_commits, + report.recovered_points, + report.source_prefix_crc32, + report.destination_snapshot_crc32 + ) +} + fn backup_report_json(report: &BackupReport) -> String { let fallback_error_kinds = report .hard_link_fallback_error_kinds @@ -574,14 +602,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} 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]" + "usage:\n {program} inspect \n {program} check-store \n {program} backup \n {program} restore \n {program} salvage \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, restore_report_json}; - use ftwdb::{BackupReport, RestoreReport}; + use super::{backup_report_json, restore_report_json, salvage_report_json}; + use ftwdb::{BackupReport, RestoreReport, SalvageReport, SalvageStatus, SalvageStopReason}; #[test] fn backup_json_uses_fixed_error_kind_names() { @@ -618,4 +646,25 @@ mod tests { "{\"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\"}" ); } + + #[test] + fn salvage_json_has_fixed_status_reason_checksums_and_counts() { + let report = SalvageReport { + status: SalvageStatus::Partial, + source_bytes: 130, + recovered_prefix_bytes: 123, + discarded_bytes: 7, + stop_offset: 123, + stop_reason: SalvageStopReason::IncompleteFrameHeader, + recovered_commits: 8, + recovered_points: 9, + source_prefix_crc32: 0x0123_abcd, + destination_snapshot_crc32: 0x0123_abcd, + }; + + assert_eq!( + salvage_report_json(&report), + "{\"format\":\"ftwdb-salvage-v1\",\"status\":\"partial\",\"source_bytes\":130,\"recovered_prefix_bytes\":123,\"discarded_bytes\":7,\"stop_offset\":123,\"stop_reason\":\"incomplete-frame-header\",\"recovered_commits\":8,\"recovered_points\":9,\"source_prefix_crc32\":\"0123abcd\",\"destination_snapshot_crc32\":\"0123abcd\"}" + ); + } } diff --git a/src/error.rs b/src/error.rs index 2e49680..733b684 100644 --- a/src/error.rs +++ b/src/error.rs @@ -37,6 +37,9 @@ pub enum Error { path: PathBuf, reason: String, }, + SourceChanged { + path: PathBuf, + }, } impl fmt::Display for Error { @@ -75,6 +78,11 @@ impl fmt::Display for Error { "snapshot publication at {} could not be rolled back cleanly: {reason}", path.display() ), + Self::SourceChanged { path } => write!( + f, + "source file {} changed while it was being checked", + path.display() + ), } } } diff --git a/src/lib.rs b/src/lib.rs index 131fc33..5ccb0df 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -37,10 +37,13 @@ pub use real_fixture::{RealFixtureLoadReport, load_real_fixture}; pub use rollup::{CalendarGaugeRollup, FixedGaugeRollup, GaugeBucket}; pub use rollup_segment::{RollupSegment, RollupSegmentStats}; pub use segment::{Segment, SegmentStats}; -pub use storage::{Commit, Config, Database, Durability, PlanOutcome, Point, RecoveredTail, Stats}; +pub use storage::{ + Commit, Config, Database, Durability, PlanOutcome, Point, RecoveredTail, SalvageStopReason, + Stats, +}; pub use store::{ BackupReport, IntegrityReport, MaintenanceReport, RestoreReport, RetentionGate, RollupQuery, - RollupSource, Store, + RollupSource, SalvageReport, SalvageStatus, Store, }; pub use transaction::Transaction; pub use tsbs::{TsbsIotLoadReport, load_tsbs_iot}; diff --git a/src/snapshot.rs b/src/snapshot.rs index 1da71b3..83765bd 100644 --- a/src/snapshot.rs +++ b/src/snapshot.rs @@ -2,7 +2,7 @@ use crate::storage::sync_directory; use crate::{Error, Result}; use crc32fast::Hasher; use std::fs::File; -use std::io::Read; +use std::io::{Read, Seek, SeekFrom}; use std::os::unix::fs::MetadataExt; use std::path::{Component, Path, PathBuf}; use std::sync::atomic::{AtomicU64, Ordering}; @@ -93,6 +93,61 @@ pub(crate) fn snapshot_digest(root: &Path, relative_paths: &[String]) -> Result< }) } +/// Computes the one-file snapshot digest for an exact prefix of an already +/// opened regular file. Salvage uses this to compare its checked source prefix +/// with the complete `active.wlog` written to the new store. +pub(crate) fn snapshot_file_prefix_digest( + file: &mut File, + relative: &str, + prefix_bytes: u64, +) -> Result { + validate_relative_path(relative)?; + let metadata = file.metadata()?; + if !metadata.file_type().is_file() || metadata.len() < prefix_bytes { + return Err(Error::Io(std::io::Error::new( + std::io::ErrorKind::UnexpectedEof, + "snapshot prefix is no longer present in the source file", + ))); + } + 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 hasher = Hasher::new(); + hasher.update(SNAPSHOT_CRC32_DOMAIN); + hasher.update(&path_len.to_le_bytes()); + hasher.update(path_bytes); + hasher.update(&prefix_bytes.to_le_bytes()); + file.seek(SeekFrom::Start(0))?; + hash_exact_bytes(file, prefix_bytes, relative, &mut hasher)?; + Ok(SnapshotDigest { + files: 1, + bytes: prefix_bytes, + crc32: hasher.finalize(), + }) +} + +fn hash_exact_bytes( + file: &mut File, + mut remaining: u64, + relative: &str, + hasher: &mut Hasher, +) -> Result<()> { + let mut buffer = vec![0_u8; CHECKSUM_BUFFER_BYTES]; + 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; + } + Ok(()) +} + fn open_snapshot_file(root: &Path, relative: &str) -> Result { use rustix::fs::{Mode, OFlags, open, openat}; diff --git a/src/storage.rs b/src/storage.rs index 799faff..5e8229d 100644 --- a/src/storage.rs +++ b/src/storage.rs @@ -6,6 +6,7 @@ use crate::transaction::{Record, Transaction}; use crc32fast::hash; use serde::{Deserialize, Serialize}; use std::collections::{BTreeMap, HashMap, HashSet}; +use std::fmt; use std::fs::{File, OpenOptions, TryLockError}; use std::io::{Read, Seek, SeekFrom, Write}; use std::os::unix::fs::MetadataExt; @@ -136,6 +137,49 @@ pub enum RecoveredTail { IncompletePayload, } +/// Why salvage stopped after its longest fully validated raw-log prefix. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum SalvageStopReason { + CleanEof, + IncompleteFrameHeader, + InvalidFrameMagic, + UnsupportedFrameVersion, + FrameHeaderChecksumMismatch, + InvalidLegacyFrameSize, + TransactionFrameTooLarge, + IdentifiedTransactionTooShort, + UnknownFrameKind, + IncompleteFramePayload, + PayloadChecksumMismatch, + DuplicateCommitId, + InvalidTransaction, + TransactionPointCountTooLarge, + InvalidCatalogTransaction, +} + +impl fmt::Display for SalvageStopReason { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + let name = match self { + Self::CleanEof => "clean-eof", + Self::IncompleteFrameHeader => "incomplete-frame-header", + Self::InvalidFrameMagic => "invalid-frame-magic", + Self::UnsupportedFrameVersion => "unsupported-frame-version", + Self::FrameHeaderChecksumMismatch => "frame-header-checksum-mismatch", + Self::InvalidLegacyFrameSize => "invalid-legacy-frame-size", + Self::TransactionFrameTooLarge => "transaction-frame-too-large", + Self::IdentifiedTransactionTooShort => "identified-transaction-too-short", + Self::UnknownFrameKind => "unknown-frame-kind", + Self::IncompleteFramePayload => "incomplete-frame-payload", + Self::PayloadChecksumMismatch => "payload-checksum-mismatch", + Self::DuplicateCommitId => "duplicate-commit-id", + Self::InvalidTransaction => "invalid-transaction", + Self::TransactionPointCountTooLarge => "transaction-point-count-too-large", + Self::InvalidCatalogTransaction => "invalid-catalog-transaction", + }; + f.write_str(name) + } +} + impl std::fmt::Display for RecoveredTail { fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { formatter.write_str(match self { @@ -193,6 +237,9 @@ std::thread_local! { static FAIL_NEXT_SYNC: std::cell::Cell> = const { std::cell::Cell::new(None) }; + + static MUTATE_SALVAGE_SOURCE_AFTER_IDENTITY_CHECKS: std::cell::Cell> = + const { std::cell::Cell::new(None) }; } #[cfg(test)] @@ -200,6 +247,13 @@ fn fail_next_sync(kind: std::io::ErrorKind) { FAIL_NEXT_SYNC.with(|failure| failure.set(Some(kind))); } +#[cfg(test)] +pub(crate) fn mutate_salvage_source_after_identity_checks(passed_checks: usize) { + MUTATE_SALVAGE_SOURCE_AFTER_IDENTITY_CHECKS.with(|remaining| { + remaining.set(Some(passed_checks)); + }); +} + fn sync_database_file(file: &File) -> std::io::Result<()> { #[cfg(test)] if let Some(kind) = FAIL_NEXT_SYNC.with(std::cell::Cell::take) { @@ -794,6 +848,13 @@ struct Scan { catalog_records: u64, recovered_tail_bytes: u64, recovered_tail: RecoveredTail, + validated_bytes: u64, + salvage_stop_reason: Option, +} + +enum ScanMode { + Recover { simulate: bool }, + Salvage, } /// Replays the log, recovering from a torn tail. With `simulate_recovery` the @@ -804,6 +865,22 @@ fn scan_and_recover( max_batch_points: usize, max_transaction_bytes: usize, simulate_recovery: bool, +) -> Result { + scan_log( + file, + max_batch_points, + max_transaction_bytes, + ScanMode::Recover { + simulate: simulate_recovery, + }, + ) +} + +fn scan_log( + file: &mut File, + max_batch_points: usize, + max_transaction_bytes: usize, + mode: ScanMode, ) -> Result { let mut database_header = [0_u8; DATABASE_HEADER_BYTES]; file.seek(SeekFrom::Start(0))?; @@ -830,14 +907,29 @@ fn scan_and_recover( let mut catalog_records = 0_u64; let mut recovered_tail_bytes = 0_u64; let mut recovered_tail = RecoveredTail::None; + let mut salvage_stop_reason = None; + + macro_rules! stop_or_corruption { + ($reason:expr, $message:expr) => { + if matches!(mode, ScanMode::Salvage) { + salvage_stop_reason = Some($reason); + break; + } + return corruption(offset, $message); + }; + } while offset < original_len { let remaining = original_len - offset; if remaining < FRAME_HEADER_BYTES as u64 { - recovered_tail_bytes = remaining; - recovered_tail = RecoveredTail::IncompleteHeader; - if !simulate_recovery { - truncate_recovered_tail(file, offset)?; + if let ScanMode::Recover { simulate } = mode { + recovered_tail_bytes = remaining; + recovered_tail = RecoveredTail::IncompleteHeader; + if !simulate { + truncate_recovered_tail(file, offset)?; + } + } else { + salvage_stop_reason = Some(SalvageStopReason::IncompleteFrameHeader); } break; } @@ -846,15 +938,21 @@ fn scan_and_recover( file.seek(SeekFrom::Start(offset))?; file.read_exact(&mut frame_header)?; if &frame_header[..4] != FRAME_MAGIC { - return corruption(offset, "invalid frame magic"); + stop_or_corruption!(SalvageStopReason::InvalidFrameMagic, "invalid frame magic"); } let frame_version = u16::from_le_bytes(frame_header[4..6].try_into().unwrap()); if frame_version != FRAME_VERSION { - return corruption(offset, "unsupported frame version"); + stop_or_corruption!( + SalvageStopReason::UnsupportedFrameVersion, + "unsupported frame version" + ); } let header_checksum = u32::from_le_bytes(frame_header[20..24].try_into().unwrap()); if hash(&frame_header[..20]) != header_checksum { - return corruption(offset, "frame header checksum mismatch"); + stop_or_corruption!( + SalvageStopReason::FrameHeaderChecksumMismatch, + "frame header checksum mismatch" + ); } let frame_kind = u16::from_le_bytes(frame_header[6..8].try_into().unwrap()); @@ -862,35 +960,47 @@ fn scan_and_recover( let payload_len = u32::from_le_bytes(frame_header[12..16].try_into().unwrap()) as usize; let payload_checksum = u32::from_le_bytes(frame_header[16..20].try_into().unwrap()); if frame_kind == FRAME_KIND_LEGACY_POINTS { - let expected_payload_len = - item_count - .checked_mul(POINT_BYTES) - .ok_or_else(|| Error::Corruption { - offset, - reason: "frame point count overflows".to_owned(), - })?; + let Some(expected_payload_len) = item_count.checked_mul(POINT_BYTES) else { + stop_or_corruption!( + SalvageStopReason::InvalidLegacyFrameSize, + "frame point count overflows" + ); + }; if item_count > max_batch_points || payload_len != expected_payload_len { - return corruption(offset, "invalid legacy point frame size"); + stop_or_corruption!( + SalvageStopReason::InvalidLegacyFrameSize, + "invalid legacy point frame size" + ); } } else if frame_kind == FRAME_KIND_TRANSACTION || frame_kind == FRAME_KIND_IDENTIFIED_TRANSACTION { if payload_len > max_transaction_bytes { - return corruption(offset, "transaction frame exceeds configured maximum"); + stop_or_corruption!( + SalvageStopReason::TransactionFrameTooLarge, + "transaction frame exceeds configured maximum" + ); } if frame_kind == FRAME_KIND_IDENTIFIED_TRANSACTION && payload_len < COMMIT_ID_BYTES { - return corruption(offset, "identified transaction frame is too short"); + stop_or_corruption!( + SalvageStopReason::IdentifiedTransactionTooShort, + "identified transaction frame is too short" + ); } } else { - return corruption(offset, "unknown frame kind"); + stop_or_corruption!(SalvageStopReason::UnknownFrameKind, "unknown frame kind"); } let frame_len = FRAME_HEADER_BYTES as u64 + payload_len as u64; if remaining < frame_len { - recovered_tail_bytes = remaining; - recovered_tail = RecoveredTail::IncompletePayload; - if !simulate_recovery { - truncate_recovered_tail(file, offset)?; + if let ScanMode::Recover { simulate } = mode { + recovered_tail_bytes = remaining; + recovered_tail = RecoveredTail::IncompletePayload; + if !simulate { + truncate_recovered_tail(file, offset)?; + } + } else { + salvage_stop_reason = Some(SalvageStopReason::IncompleteFramePayload); } break; } @@ -903,7 +1013,7 @@ fn scan_and_recover( } else { "payload checksum mismatch before valid tail" }; - return corruption(offset, reason); + stop_or_corruption!(SalvageStopReason::PayloadChecksumMismatch, reason); } if frame_kind == FRAME_KIND_LEGACY_POINTS { @@ -920,14 +1030,32 @@ fn scan_and_recover( // tampering (say, concatenated logs); failing closed keeps // the exactly-once promise instead of silently replaying. if !commit_ids.insert(commit_id) { - return corruption(offset, "duplicate commit identifier"); + stop_or_corruption!( + SalvageStopReason::DuplicateCommitId, + "duplicate commit identifier" + ); } &payload[COMMIT_ID_BYTES..] } else { &payload[..] }; - let records = decode_transaction(transaction_payload, item_count, offset)?; - catalog.apply_recovered(&records, offset)?; + let records = match decode_transaction(transaction_payload, item_count, offset) { + Ok(records) => records, + Err(error) if matches!(mode, ScanMode::Salvage) => { + let _ = error; + salvage_stop_reason = Some(SalvageStopReason::InvalidTransaction); + break; + } + Err(error) => return Err(error), + }; + if let Err(error) = catalog.apply_recovered(&records, offset) { + if matches!(mode, ScanMode::Salvage) { + let _ = error; + salvage_stop_reason = Some(SalvageStopReason::InvalidCatalogTransaction); + break; + } + return Err(error); + } let recovered_point_count: usize = records .iter() .map(|record| match record { @@ -936,7 +1064,10 @@ fn scan_and_recover( }) .sum(); if recovered_point_count > max_batch_points { - return corruption(offset, "transaction point count exceeds maximum"); + stop_or_corruption!( + SalvageStopReason::TransactionPointCountTooLarge, + "transaction point count exceeds maximum" + ); } for record in &records { match record { @@ -954,6 +1085,10 @@ fn scan_and_recover( offset += frame_len; } + if matches!(mode, ScanMode::Salvage) && salvage_stop_reason.is_none() { + salvage_stop_reason = Some(SalvageStopReason::CleanEof); + } + Ok(Scan { index, catalog, @@ -963,9 +1098,197 @@ fn scan_and_recover( catalog_records, recovered_tail_bytes, recovered_tail, + validated_bytes: offset, + salvage_stop_reason, }) } +pub(crate) struct SalvageSource { + pub(crate) file: File, + pub(crate) path: std::path::PathBuf, + pub(crate) source_bytes: u64, + pub(crate) recovered_prefix_bytes: u64, + pub(crate) recovered_commits: u64, + pub(crate) recovered_points: u64, + pub(crate) stop_reason: SalvageStopReason, + identity: ReadOnlyFileIdentity, + root: File, + root_path: std::path::PathBuf, + root_identity: ReadOnlyDirectoryIdentity, +} + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +struct ReadOnlyFileIdentity { + device: u64, + inode: u64, + length: u64, + modified_seconds: i64, + modified_nanoseconds: i64, + changed_seconds: i64, + changed_nanoseconds: i64, +} + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +struct ReadOnlyDirectoryIdentity { + device: u64, + inode: u64, +} + +impl ReadOnlyDirectoryIdentity { + fn read(metadata: &std::fs::Metadata) -> Self { + Self { + device: metadata.dev(), + inode: metadata.ino(), + } + } +} + +impl ReadOnlyFileIdentity { + fn read(metadata: &std::fs::Metadata) -> Self { + Self { + device: metadata.dev(), + inode: metadata.ino(), + length: metadata.len(), + modified_seconds: metadata.mtime(), + modified_nanoseconds: metadata.mtime_nsec(), + changed_seconds: metadata.ctime(), + changed_nanoseconds: metadata.ctime_nsec(), + } + } +} + +impl SalvageSource { + pub(crate) fn open(root_path: &Path, file_name: &str) -> Result { + use rustix::fs::{Mode, OFlags, open, openat}; + + let root_descriptor = open( + root_path, + OFlags::RDONLY | OFlags::CLOEXEC | OFlags::NOFOLLOW | OFlags::DIRECTORY, + Mode::empty(), + ) + .map_err(|error| Error::Io(error.into()))?; + let root = File::from(root_descriptor); + let root_metadata = root.metadata()?; + let root_path_metadata = std::fs::symlink_metadata(root_path)?; + let root_identity = ReadOnlyDirectoryIdentity::read(&root_metadata); + if !root_path_metadata.file_type().is_dir() + || ReadOnlyDirectoryIdentity::read(&root_path_metadata) != root_identity + { + return Err(Error::Io(std::io::Error::new( + std::io::ErrorKind::InvalidData, + format!( + "salvage source is not a stable directory: {}", + root_path.display() + ), + ))); + } + + let path = root_path.join(file_name); + 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 = openat( + &root, + file_name, + OFlags::RDONLY | OFlags::CLOEXEC | OFlags::NOFOLLOW | OFlags::NONBLOCK, + Mode::empty(), + ) + .map_err(|error| Error::Io(error.into()))?; + let mut 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()), + ))); + } + match file.try_lock_shared() { + Ok(()) => {} + Err(TryLockError::WouldBlock) => { + return Err(Error::Locked { + path: path.to_owned(), + }); + } + Err(TryLockError::Error(error)) => return Err(Error::Io(error)), + } + let metadata = file.metadata()?; + if metadata.len() < DATABASE_HEADER_BYTES as u64 { + return Err(Error::InvalidHeader); + } + let scan = scan_log( + &mut file, + Config::default().max_batch_points, + Config::default().max_transaction_bytes, + ScanMode::Salvage, + )?; + Ok(Self { + file, + path: path.to_owned(), + source_bytes: metadata.len(), + recovered_prefix_bytes: scan.validated_bytes, + recovered_commits: scan.commits, + recovered_points: scan.points, + stop_reason: scan.salvage_stop_reason.unwrap(), + identity: ReadOnlyFileIdentity::read(&metadata), + root, + root_path: root_path.to_owned(), + root_identity, + }) + } + + pub(crate) fn ensure_unchanged(&self) -> Result<()> { + #[cfg(test)] + let mutate = MUTATE_SALVAGE_SOURCE_AFTER_IDENTITY_CHECKS.with(|remaining| { + let Some(checks) = remaining.get() else { + return false; + }; + if checks == 0 { + remaining.set(None); + true + } else { + remaining.set(Some(checks - 1)); + false + } + }); + #[cfg(test)] + if mutate { + let mut writer = OpenOptions::new().read(true).write(true).open(&self.path)?; + writer.seek(SeekFrom::End(-1))?; + let mut byte = [0_u8; 1]; + writer.read_exact(&mut byte)?; + writer.seek(SeekFrom::End(-1))?; + writer.write_all(&[byte[0] ^ 0xff])?; + writer.sync_all()?; + } + + let descriptor_identity = ReadOnlyFileIdentity::read(&self.file.metadata()?); + let path_metadata = std::fs::symlink_metadata(&self.path)?; + let path_identity = ReadOnlyFileIdentity::read(&path_metadata); + let root_descriptor_identity = ReadOnlyDirectoryIdentity::read(&self.root.metadata()?); + let root_path_metadata = std::fs::symlink_metadata(&self.root_path)?; + let root_path_identity = ReadOnlyDirectoryIdentity::read(&root_path_metadata); + if !path_metadata.file_type().is_file() + || descriptor_identity != self.identity + || path_identity != self.identity + || !root_path_metadata.file_type().is_dir() + || root_descriptor_identity != self.root_identity + || root_path_identity != self.root_identity + { + return Err(Error::SourceChanged { + path: self.path.clone(), + }); + } + Ok(()) + } +} + /// 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. @@ -1117,6 +1440,9 @@ fn decode_transaction(payload: &[u8], expected_records: usize, offset: u64) -> R if record_count != expected_records { return corruption(offset, "transaction record count mismatch"); } + if record_count > payload.len().saturating_sub(TRANSACTION_HEADER_BYTES) / RECORD_HEADER_BYTES { + return corruption(offset, "transaction record count exceeds payload bounds"); + } let mut cursor = TRANSACTION_HEADER_BYTES; let mut records = Vec::with_capacity(record_count); @@ -1235,14 +1561,19 @@ fn decode_point(raw: &[u8]) -> Point { #[cfg(test)] mod tests { - use super::{Config, Database, Durability, Point, fail_next_sync}; + use super::{ + Config, DATABASE_HEADER_BYTES, Database, Durability, FRAME_HEADER_BYTES, + FRAME_KIND_IDENTIFIED_TRANSACTION, FRAME_KIND_TRANSACTION, FRAME_VERSION, Point, + SalvageSource, SalvageStopReason, ScanMode, encode_frame_header, encode_transaction, + fail_next_sync, scan_log, + }; use crate::{ Entity, EntityId, Error, Plan, PlanStatus, RollupPolicy, Run, RunId, RunKind, RunStatus, SeriesDefinition, SeriesSemantics, Transaction, }; use std::collections::BTreeMap; use std::fs::OpenOptions; - use std::io::{Seek, SeekFrom, Write}; + use std::io::{Read, Seek, SeekFrom, Write}; use tempfile::tempdir; fn point(valid_time: i64, knowledge_time: i64, change_time: i64, value: f64) -> Point { @@ -1288,6 +1619,379 @@ mod tests { } } + fn header_only(path: &std::path::Path) { + Database::open(path).unwrap().close().unwrap(); + assert_eq!( + std::fs::metadata(path).unwrap().len(), + DATABASE_HEADER_BYTES as u64 + ); + } + + fn append_raw_frame(path: &std::path::Path, kind: u16, items: u32, payload: &[u8]) -> u64 { + let offset = std::fs::metadata(path).unwrap().len(); + let header = encode_frame_header( + kind, + items, + u32::try_from(payload.len()).unwrap(), + crc32fast::hash(payload), + ); + let mut file = OpenOptions::new().append(true).open(path).unwrap(); + file.write_all(&header).unwrap(); + file.write_all(payload).unwrap(); + file.sync_all().unwrap(); + offset + } + + fn legacy_log(path: &std::path::Path, frames: usize) -> Vec { + let mut database = Database::open(path).unwrap(); + let mut offsets = Vec::new(); + for frame in 0..frames { + offsets.push(database.stats().unwrap().file_bytes); + database + .append(&[point( + frame as i64, + frame as i64, + frame as i64, + frame as f64, + )]) + .unwrap(); + } + database.close().unwrap(); + offsets + } + + fn rewrite_frame_header( + path: &std::path::Path, + offset: u64, + rewrite: impl FnOnce(&mut [u8; FRAME_HEADER_BYTES]), + ) { + let mut file = OpenOptions::new() + .read(true) + .write(true) + .open(path) + .unwrap(); + let mut header = [0_u8; FRAME_HEADER_BYTES]; + file.seek(SeekFrom::Start(offset)).unwrap(); + file.read_exact(&mut header).unwrap(); + rewrite(&mut header); + let checksum = crc32fast::hash(&header[..20]); + header[20..24].copy_from_slice(&checksum.to_le_bytes()); + file.seek(SeekFrom::Start(offset)).unwrap(); + file.write_all(&header).unwrap(); + file.sync_all().unwrap(); + } + + fn salvage_scan(path: &std::path::Path, max_batch_points: usize) -> super::Scan { + let mut file = OpenOptions::new().read(true).open(path).unwrap(); + scan_log( + &mut file, + max_batch_points, + Config::default().max_transaction_bytes, + ScanMode::Salvage, + ) + .unwrap() + } + + #[test] + fn salvage_stops_before_the_first_bad_frame_and_never_resynchronizes() { + let directory = tempdir().unwrap(); + let path = directory.path().join("bad-middle-frame.ftwdb"); + let offsets = legacy_log(&path, 3); + let source_bytes = std::fs::metadata(&path).unwrap().len(); + + let mut file = OpenOptions::new() + .read(true) + .write(true) + .open(&path) + .unwrap(); + file.seek(SeekFrom::Start(offsets[1] + FRAME_HEADER_BYTES as u64)) + .unwrap(); + let mut byte = [0_u8; 1]; + file.read_exact(&mut byte).unwrap(); + byte[0] ^= 0x80; + file.seek(SeekFrom::Start(offsets[1] + FRAME_HEADER_BYTES as u64)) + .unwrap(); + file.write_all(&byte).unwrap(); + file.sync_all().unwrap(); + + let scan = salvage_scan(&path, Config::default().max_batch_points); + assert_eq!( + scan.salvage_stop_reason, + Some(SalvageStopReason::PayloadChecksumMismatch) + ); + assert_eq!(scan.validated_bytes, offsets[1]); + assert_eq!(scan.commits, 1); + assert_eq!(scan.points, 1); + assert_eq!( + source_bytes - scan.validated_bytes, + source_bytes - offsets[1] + ); + } + + #[test] + fn salvage_classifies_frame_headers_bounds_and_short_ends_as_partial() { + let directory = tempdir().unwrap(); + + let complete_bad_crc = directory.path().join("complete-bad-crc.ftwdb"); + let offsets = legacy_log(&complete_bad_crc, 2); + let mut file = OpenOptions::new() + .read(true) + .write(true) + .open(&complete_bad_crc) + .unwrap(); + file.seek(SeekFrom::End(-1)).unwrap(); + let mut byte = [0_u8; 1]; + file.read_exact(&mut byte).unwrap(); + byte[0] ^= 1; + file.seek(SeekFrom::End(-1)).unwrap(); + file.write_all(&byte).unwrap(); + file.sync_all().unwrap(); + let scan = salvage_scan(&complete_bad_crc, Config::default().max_batch_points); + assert_eq!( + scan.salvage_stop_reason, + Some(SalvageStopReason::PayloadChecksumMismatch) + ); + assert_eq!(scan.validated_bytes, offsets[1]); + assert_eq!((scan.commits, scan.points), (1, 1)); + + let short_header = directory.path().join("short-header.ftwdb"); + let offsets = legacy_log(&short_header, 2); + OpenOptions::new() + .write(true) + .open(&short_header) + .unwrap() + .set_len(offsets[1] + 7) + .unwrap(); + let scan = salvage_scan(&short_header, Config::default().max_batch_points); + assert_eq!( + scan.salvage_stop_reason, + Some(SalvageStopReason::IncompleteFrameHeader) + ); + assert_eq!(scan.validated_bytes, offsets[1]); + assert_eq!((scan.commits, scan.points), (1, 1)); + + let short_payload = directory.path().join("short-payload.ftwdb"); + let offsets = legacy_log(&short_payload, 2); + let length = std::fs::metadata(&short_payload).unwrap().len(); + OpenOptions::new() + .write(true) + .open(&short_payload) + .unwrap() + .set_len(length - 7) + .unwrap(); + let scan = salvage_scan(&short_payload, Config::default().max_batch_points); + assert_eq!( + scan.salvage_stop_reason, + Some(SalvageStopReason::IncompleteFramePayload) + ); + assert_eq!(scan.validated_bytes, offsets[1]); + assert_eq!((scan.commits, scan.points), (1, 1)); + + let bad_magic = directory.path().join("bad-magic.ftwdb"); + let offset = legacy_log(&bad_magic, 1)[0]; + rewrite_frame_header(&bad_magic, offset, |header| { + header[..4].copy_from_slice(b"NOPE"); + }); + assert_eq!( + salvage_scan(&bad_magic, Config::default().max_batch_points).salvage_stop_reason, + Some(SalvageStopReason::InvalidFrameMagic) + ); + + let unsupported_version = directory.path().join("unsupported-version.ftwdb"); + let offset = legacy_log(&unsupported_version, 1)[0]; + rewrite_frame_header(&unsupported_version, offset, |header| { + header[4..6].copy_from_slice(&(FRAME_VERSION + 1).to_le_bytes()); + }); + assert_eq!( + salvage_scan(&unsupported_version, Config::default().max_batch_points) + .salvage_stop_reason, + Some(SalvageStopReason::UnsupportedFrameVersion) + ); + + let bad_header_crc = directory.path().join("bad-header-crc.ftwdb"); + let offset = legacy_log(&bad_header_crc, 1)[0]; + let mut file = OpenOptions::new() + .read(true) + .write(true) + .open(&bad_header_crc) + .unwrap(); + file.seek(SeekFrom::Start(offset + 8)).unwrap(); + file.write_all(&2_u32.to_le_bytes()).unwrap(); + file.sync_all().unwrap(); + assert_eq!( + salvage_scan(&bad_header_crc, Config::default().max_batch_points).salvage_stop_reason, + Some(SalvageStopReason::FrameHeaderChecksumMismatch) + ); + + let invalid_legacy_size = directory.path().join("legacy-size.ftwdb"); + let offset = legacy_log(&invalid_legacy_size, 1)[0]; + rewrite_frame_header(&invalid_legacy_size, offset, |header| { + header[8..12].copy_from_slice(&2_u32.to_le_bytes()); + }); + assert_eq!( + salvage_scan(&invalid_legacy_size, Config::default().max_batch_points) + .salvage_stop_reason, + Some(SalvageStopReason::InvalidLegacyFrameSize) + ); + + let unknown_kind = directory.path().join("unknown-kind.ftwdb"); + let offset = legacy_log(&unknown_kind, 1)[0]; + rewrite_frame_header(&unknown_kind, offset, |header| { + header[6..8].copy_from_slice(&99_u16.to_le_bytes()); + }); + assert_eq!( + salvage_scan(&unknown_kind, Config::default().max_batch_points).salvage_stop_reason, + Some(SalvageStopReason::UnknownFrameKind) + ); + + let oversized = directory.path().join("oversized-transaction.ftwdb"); + header_only(&oversized); + let offset = append_raw_frame(&oversized, FRAME_KIND_TRANSACTION, 0, b"bad"); + rewrite_frame_header(&oversized, offset, |header| { + let length = u32::try_from(Config::default().max_transaction_bytes + 1).unwrap(); + header[12..16].copy_from_slice(&length.to_le_bytes()); + }); + assert_eq!( + salvage_scan(&oversized, Config::default().max_batch_points).salvage_stop_reason, + Some(SalvageStopReason::TransactionFrameTooLarge) + ); + } + + #[test] + fn salvage_classifies_all_transaction_failures_after_a_valid_header_as_partial() { + let directory = tempdir().unwrap(); + + let invalid_transaction = directory.path().join("invalid-transaction.ftwdb"); + header_only(&invalid_transaction); + append_raw_frame(&invalid_transaction, FRAME_KIND_TRANSACTION, 0, b"bad"); + let scan = salvage_scan(&invalid_transaction, Config::default().max_batch_points); + assert_eq!( + scan.salvage_stop_reason, + Some(SalvageStopReason::InvalidTransaction) + ); + assert_eq!(scan.validated_bytes, DATABASE_HEADER_BYTES as u64); + assert_eq!((scan.commits, scan.points), (0, 0)); + + let invalid_catalog = directory.path().join("invalid-catalog.ftwdb"); + header_only(&invalid_catalog); + let mut transaction = Transaction::new(); + transaction.define_series(power_series()); + let payload = encode_transaction(&transaction).unwrap(); + append_raw_frame(&invalid_catalog, FRAME_KIND_TRANSACTION, 1, &payload); + let scan = salvage_scan(&invalid_catalog, Config::default().max_batch_points); + assert_eq!( + scan.salvage_stop_reason, + Some(SalvageStopReason::InvalidCatalogTransaction) + ); + assert_eq!(scan.validated_bytes, DATABASE_HEADER_BYTES as u64); + assert_eq!((scan.commits, scan.points), (0, 0)); + + let duplicate = directory.path().join("duplicate-commit-id.ftwdb"); + header_only(&duplicate); + let mut identified = Transaction::new(); + identified.with_commit_id(77); + let payload = encode_transaction(&identified).unwrap(); + append_raw_frame(&duplicate, FRAME_KIND_IDENTIFIED_TRANSACTION, 0, &payload); + let duplicate_offset = + append_raw_frame(&duplicate, FRAME_KIND_IDENTIFIED_TRANSACTION, 0, &payload); + let scan = salvage_scan(&duplicate, Config::default().max_batch_points); + assert_eq!( + scan.salvage_stop_reason, + Some(SalvageStopReason::DuplicateCommitId) + ); + assert_eq!(scan.validated_bytes, duplicate_offset); + assert_eq!((scan.commits, scan.points), (1, 0)); + + let too_many_points = directory.path().join("too-many-points.ftwdb"); + header_only(&too_many_points); + let mut catalog = Transaction::new(); + catalog.upsert_entity(home()).define_series(power_series()); + let payload = encode_transaction(&catalog).unwrap(); + append_raw_frame(&too_many_points, FRAME_KIND_TRANSACTION, 2, &payload); + let mut first = point(1, 1, 1, 1.0); + first.run_id = 0; + let mut second = point(2, 2, 2, 2.0); + second.run_id = 0; + let mut transaction = Transaction::new(); + transaction.append_points(vec![first, second]); + let payload = encode_transaction(&transaction).unwrap(); + let rejected_offset = + append_raw_frame(&too_many_points, FRAME_KIND_TRANSACTION, 1, &payload); + let scan = salvage_scan(&too_many_points, 1); + assert_eq!( + scan.salvage_stop_reason, + Some(SalvageStopReason::TransactionPointCountTooLarge) + ); + assert_eq!(scan.validated_bytes, rejected_offset); + assert_eq!((scan.commits, scan.points), (1, 0)); + + let identified_too_short = directory.path().join("identified-too-short.ftwdb"); + header_only(&identified_too_short); + append_raw_frame( + &identified_too_short, + FRAME_KIND_IDENTIFIED_TRANSACTION, + 0, + b"short", + ); + assert_eq!( + salvage_scan(&identified_too_short, Config::default().max_batch_points) + .salvage_stop_reason, + Some(SalvageStopReason::IdentifiedTransactionTooShort) + ); + + let header_only_path = directory.path().join("header-only.ftwdb"); + header_only(&header_only_path); + let scan = salvage_scan(&header_only_path, Config::default().max_batch_points); + assert_eq!(scan.salvage_stop_reason, Some(SalvageStopReason::CleanEof)); + assert_eq!(scan.validated_bytes, DATABASE_HEADER_BYTES as u64); + assert_eq!((scan.commits, scan.points), (0, 0)); + } + + #[test] + fn normal_recovery_keeps_catalog_validation_before_the_point_limit() { + let directory = tempdir().unwrap(); + let path = directory.path().join("recovery-policy.ftwdb"); + header_only(&path); + let mut transaction = Transaction::new(); + transaction.append_points(vec![point(1, 1, 1, 1.0), point(2, 2, 2, 2.0)]); + let payload = encode_transaction(&transaction).unwrap(); + append_raw_frame(&path, FRAME_KIND_TRANSACTION, 1, &payload); + + match Database::open_with( + &path, + Config { + max_batch_points: 1, + ..Config::default() + }, + ) { + Err(Error::Corruption { reason, .. }) => { + assert!(reason.contains("invalid recovered transaction")); + } + Err(error) => panic!("wrong normal recovery error: {error}"), + Ok(_) => panic!("normal recovery accepted invalid catalog dependencies"), + } + } + + #[test] + fn salvage_source_identity_detects_an_uncooperative_in_place_writer() { + let directory = tempdir().unwrap(); + let source_root = directory.path().join("source"); + std::fs::create_dir(&source_root).unwrap(); + let active = source_root.join("active.wlog"); + legacy_log(&active, 1); + let source = SalvageSource::open(&source_root, "active.wlog").unwrap(); + + let mut writer = OpenOptions::new().write(true).open(&active).unwrap(); + writer.seek(SeekFrom::End(-1)).unwrap(); + writer.write_all(&[0xff]).unwrap(); + writer.sync_all().unwrap(); + + assert!(matches!( + source.ensure_unchanged(), + Err(Error::SourceChanged { path }) if path == active + )); + } + fn optimization_run() -> Run { Run { id: RunId(9), diff --git a/src/store.rs b/src/store.rs index 2037c1b..df05288 100644 --- a/src/store.rs +++ b/src/store.rs @@ -2,15 +2,17 @@ use crate::manifest::{self, Manifest, RollupDescriptor}; use crate::rollup::calendar_bucket_bounds; use crate::snapshot::{ PublicationStep, StagedDirectory, inject_checksum_mismatch, publication_checkpoint, - snapshot_digest, + snapshot_digest, snapshot_file_prefix_digest, }; -use crate::storage::{sync_directory, sync_parent_directory}; +use crate::storage::{SalvageSource, sync_directory, sync_parent_directory}; use crate::transaction::Record; use crate::{ CalendarGaugeRollup, Commit, Config, Database, Error, FixedGaugeRollup, GaugeBucket, Point, Result, RollupResolution, RollupSegment, SeriesSemantics, Transaction, }; use std::collections::{BTreeMap, HashMap}; +use std::fmt; +use std::io::{Read, Seek, SeekFrom}; use std::path::{Path, PathBuf}; use std::sync::RwLock; use std::time::{SystemTime, UNIX_EPOCH}; @@ -89,6 +91,35 @@ pub struct RestoreReport { pub destination_snapshot_crc32: u32, } +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum SalvageStatus { + Clean, + Partial, +} + +impl fmt::Display for SalvageStatus { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.write_str(match self { + Self::Clean => "clean", + Self::Partial => "partial", + }) + } +} + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub struct SalvageReport { + pub status: SalvageStatus, + pub source_bytes: u64, + pub recovered_prefix_bytes: u64, + pub discarded_bytes: u64, + pub stop_offset: u64, + pub stop_reason: crate::SalvageStopReason, + pub recovered_commits: u64, + pub recovered_points: u64, + pub source_prefix_crc32: u32, + pub destination_snapshot_crc32: u32, +} + impl BackupReport { fn record_copy(&mut self) { self.files += 1; @@ -746,6 +777,109 @@ impl Store { } } + /// Copies the longest raw-log prefix that validates from the first frame + /// into a new store. Derived manifests and rollups are never opened. + pub fn salvage_from( + damaged_store: impl AsRef, + destination: impl AsRef, + ) -> Result { + let mut source = SalvageSource::open(damaged_store.as_ref(), ACTIVE_LOG)?; + let destination = destination.as_ref(); + let relative_paths = vec![ACTIVE_LOG.to_owned()]; + let source_digest = snapshot_file_prefix_digest( + &mut source.file, + ACTIVE_LOG, + source.recovered_prefix_bytes, + )?; + source.ensure_unchanged()?; + + let staged = StagedDirectory::create(destination, "salvage")?; + write_salvage_stage(&mut source, staged.path())?; + let stage_digest = + inject_checksum_mismatch(snapshot_digest(staged.path(), &relative_paths)?); + if stage_digest != source_digest { + return Err(snapshot_mismatch( + "source prefix", + source_digest, + "stage", + stage_digest, + )); + } + + let stage = Self::open_read_only(staged.path())?; + let stage_integrity = stage.check_integrity()?; + stage.require_clean_restore_source(&stage_integrity)?; + if stage_integrity.raw_commits != source.recovered_commits + || stage_integrity.raw_points != source.recovered_points + { + return Err(Error::Corruption { + offset: 0, + reason: "salvage stage counts differ from the validated source prefix".to_owned(), + }); + } + + let publication = staged.publish(destination)?; + let checked = (|| { + publication_checkpoint(PublicationStep::PostCheck)?; + let salvaged = Self::open_read_only(destination)?; + let destination_integrity = salvaged.check_integrity()?; + salvaged.require_clean_restore_source(&destination_integrity)?; + let destination_digest = snapshot_digest(destination, &relative_paths)?; + if destination_digest != source_digest { + return Err(snapshot_mismatch( + "source prefix", + source_digest, + "destination", + destination_digest, + )); + } + if destination_integrity.raw_commits != source.recovered_commits + || destination_integrity.raw_points != source.recovered_points + { + return Err(Error::Corruption { + offset: 0, + reason: "salvaged store counts differ from the validated source prefix" + .to_owned(), + }); + } + source.ensure_unchanged()?; + let discarded_bytes = source + .source_bytes + .checked_sub(source.recovered_prefix_bytes) + .ok_or(Error::SourceChanged { + path: source.path.clone(), + })?; + Ok(SalvageReport { + status: if source.stop_reason == crate::SalvageStopReason::CleanEof { + SalvageStatus::Clean + } else { + SalvageStatus::Partial + }, + source_bytes: source.source_bytes, + recovered_prefix_bytes: source.recovered_prefix_bytes, + discarded_bytes, + stop_offset: source.recovered_prefix_bytes, + stop_reason: source.stop_reason, + recovered_commits: destination_integrity.raw_commits, + recovered_points: destination_integrity.raw_points, + source_prefix_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] pub fn root(&self) -> &Path { &self.root @@ -1244,6 +1378,33 @@ fn copy_and_sync(source: &Path, destination: &Path) -> std::io::Result { Ok(bytes) } +fn write_salvage_stage(source: &mut SalvageSource, temporary: &Path) -> Result<()> { + let manifests = temporary.join(MANIFEST_DIRECTORY); + let rollups = temporary.join(ROLLUP_DIRECTORY); + std::fs::create_dir(&manifests)?; + std::fs::create_dir(&rollups)?; + publication_checkpoint(PublicationStep::Copy)?; + source.file.seek(SeekFrom::Start(0))?; + let mut active = std::fs::OpenOptions::new() + .create_new(true) + .write(true) + .open(temporary.join(ACTIVE_LOG))?; + let mut prefix = std::io::Read::by_ref(&mut source.file).take(source.recovered_prefix_bytes); + let copied = std::io::copy(&mut prefix, &mut active)?; + if copied != source.recovered_prefix_bytes { + return Err(Error::SourceChanged { + path: source.path.clone(), + }); + } + active.sync_all()?; + source.ensure_unchanged()?; + publication_checkpoint(PublicationStep::Sync)?; + sync_directory(&manifests)?; + sync_directory(&rollups)?; + sync_directory(temporary)?; + Ok(()) +} + #[derive(Debug)] enum LinkOrCopy { Linked, @@ -1350,11 +1511,14 @@ fn snapshot_mismatch( #[cfg(test)] mod tests { - use super::{BackupReport, LinkOrCopy, RollupSource, Store, hard_link_or_copy_with}; + use super::{ + BackupReport, LinkOrCopy, RollupSource, SalvageStatus, Store, hard_link_or_copy_with, + }; use crate::snapshot::{PublicationStep, StagedDirectory, fail_next_publication_step}; + use crate::storage::mutate_salvage_source_after_identity_checks; use crate::{ CalendarUnit, Entity, EntityId, Point, RollupPolicy, RollupResolution, RollupTier, - SeriesDefinition, SeriesSemantics, Transaction, + SalvageStopReason, SeriesDefinition, SeriesSemantics, Transaction, }; use std::collections::BTreeMap; use std::error::Error as _; @@ -1611,6 +1775,42 @@ mod tests { result } + fn salvage_with_deadline( + source: 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::salvage_from(source, 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("salvage worker must exit after its FIFO reader is released") + } + Err(std::sync::mpsc::RecvTimeoutError::Disconnected) => { + panic!("salvage worker disconnected without a result") + } + }; + worker.join().unwrap(); + assert!(!timed_out, "salvage blocked while opening active.wlog"); + result + } + fn flip_last_byte(path: &Path) { let mut file = std::fs::OpenOptions::new() .read(true) @@ -1640,6 +1840,14 @@ mod tests { .collect() } + fn salvage_stages(parent: &Path, target_name: &str) -> Vec { + let prefix = format!(".{target_name}.salvage-"); + 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(); @@ -2698,4 +2906,374 @@ mod tests { ); } } + + #[test] + fn clean_salvage_ignores_all_derived_files_and_preserves_the_source() { + let directory = tempdir().unwrap(); + let source = directory.path().join("source"); + let target = directory.path().join("salvaged"); + { + let mut store = Store::open(&source).unwrap(); + initialize( + &mut store, + vec![RollupTier { + resolution: RollupResolution::FixedMicros(5 * SECOND), + retain_for_micros: None, + }], + None, + ); + let mut transaction = Transaction::new(); + transaction.append_points(points()); + store.commit(transaction).unwrap(); + store.maintain(DAY).unwrap(); + store.close().unwrap(); + } + let manifest = active_manifest_path(&source); + let rollup = active_rollup_path(&source); + flip_last_byte(&manifest); + flip_last_byte(&rollup); + std::fs::write(source.join("rollups/orphan.rseg"), b"broken orphan").unwrap(); + let source_before = directory_snapshot(&source); + + let report = Store::salvage_from(&source, &target).unwrap(); + assert_eq!(report.status, SalvageStatus::Clean); + assert_eq!(report.stop_reason, SalvageStopReason::CleanEof); + assert_eq!(report.source_bytes, report.recovered_prefix_bytes); + assert_eq!(report.discarded_bytes, 0); + assert_eq!(report.stop_offset, report.source_bytes); + assert_eq!(report.recovered_commits, 2); + assert_eq!(report.recovered_points, 21); + assert_eq!( + report.source_prefix_crc32, + report.destination_snapshot_crc32 + ); + let integrity = Store::open_read_only(&target) + .unwrap() + .check_integrity() + .unwrap(); + assert_eq!(integrity.manifest_generation, 0); + assert_eq!(integrity.active_rollup_files, 0); + assert_eq!(integrity.raw_commits, 2); + assert_eq!(integrity.raw_points, 21); + let mut appended = Store::open(&target).unwrap(); + appended.append(&[points()[0]]).unwrap(); + appended.close().unwrap(); + assert_eq!(directory_snapshot(&source), source_before); + } + + #[test] + fn short_tail_salvages_a_verified_partial_prefix_without_mutating_source() { + let directory = tempdir().unwrap(); + let source = create_verified_backup(directory.path(), false); + let active = source.join("active.wlog"); + let clean_bytes = std::fs::metadata(&active).unwrap().len(); + let mut file = std::fs::OpenOptions::new() + .append(true) + .open(&active) + .unwrap(); + file.write_all(b"partial").unwrap(); + file.sync_all().unwrap(); + drop(file); + let source_before = directory_snapshot(&source); + let target = directory.path().join("salvaged"); + + let report = Store::salvage_from(&source, &target).unwrap(); + assert_eq!(report.status, SalvageStatus::Partial); + assert_eq!(report.stop_reason, SalvageStopReason::IncompleteFrameHeader); + assert_eq!(report.recovered_prefix_bytes, clean_bytes); + assert_eq!(report.discarded_bytes, 7); + assert_eq!(report.stop_offset, clean_bytes); + assert_eq!(report.recovered_commits, 2); + assert_eq!(report.recovered_points, 21); + assert_eq!( + report.source_prefix_crc32, + report.destination_snapshot_crc32 + ); + assert_eq!( + std::fs::metadata(target.join("active.wlog")).unwrap().len(), + clean_bytes + ); + assert_eq!(directory_snapshot(&source), source_before); + } + + #[test] + fn header_only_and_first_frame_damage_publish_checked_zero_count_stores() { + let directory = tempdir().unwrap(); + + let clean_source = directory.path().join("clean-source"); + std::fs::create_dir_all(clean_source.join("manifests")).unwrap(); + std::fs::create_dir(clean_source.join("rollups")).unwrap(); + crate::Database::open(clean_source.join("active.wlog")) + .unwrap() + .close() + .unwrap(); + let clean_target = directory.path().join("clean-target"); + let clean = Store::salvage_from(&clean_source, &clean_target).unwrap(); + assert_eq!(clean.status, SalvageStatus::Clean); + assert_eq!(clean.stop_reason, SalvageStopReason::CleanEof); + assert_eq!((clean.recovered_commits, clean.recovered_points), (0, 0)); + + let damaged_source = directory.path().join("damaged-source"); + std::fs::create_dir_all(damaged_source.join("manifests")).unwrap(); + std::fs::create_dir(damaged_source.join("rollups")).unwrap(); + let active = damaged_source.join("active.wlog"); + crate::Database::open(&active).unwrap().close().unwrap(); + let mut file = std::fs::OpenOptions::new() + .append(true) + .open(&active) + .unwrap(); + file.write_all(b"partial").unwrap(); + file.sync_all().unwrap(); + drop(file); + let damaged_before = directory_snapshot(&damaged_source); + let damaged_target = directory.path().join("damaged-target"); + let partial = Store::salvage_from(&damaged_source, &damaged_target).unwrap(); + assert_eq!(partial.status, SalvageStatus::Partial); + assert_eq!( + partial.stop_reason, + SalvageStopReason::IncompleteFrameHeader + ); + assert_eq!(partial.discarded_bytes, 7); + assert_eq!( + (partial.recovered_commits, partial.recovered_points), + (0, 0) + ); + let checked = Store::open_read_only(&damaged_target) + .unwrap() + .check_integrity() + .unwrap(); + assert_eq!((checked.raw_commits, checked.raw_points), (0, 0)); + assert_eq!(directory_snapshot(&damaged_source), damaged_before); + } + + #[test] + fn invalid_database_header_is_fatal_and_publishes_nothing() { + let directory = tempdir().unwrap(); + let source = create_verified_backup(directory.path(), false); + overwrite_byte(&source.join("active.wlog"), 0, 0); + let source_before = directory_snapshot(&source); + let target = directory.path().join("target"); + + assert!(matches!( + Store::salvage_from(&source, &target), + Err(crate::Error::InvalidHeader) + )); + assert!(!target.exists()); + assert!(salvage_stages(directory.path(), "target").is_empty()); + assert_eq!(directory_snapshot(&source), source_before); + } + + #[test] + fn unsupported_database_version_and_header_crc_are_fatal_without_publication() { + for (kind, offset) in [("version", 8_u64), ("header-crc", 12_u64)] { + let directory = tempdir().unwrap(); + let source = create_verified_backup(directory.path(), false); + overwrite_byte(&source.join("active.wlog"), offset, 2); + let source_before = directory_snapshot(&source); + let target = directory.path().join("target"); + + let error = Store::salvage_from(&source, &target).unwrap_err(); + match kind { + "version" => assert!(matches!(error, crate::Error::UnsupportedVersion(2))), + "header-crc" => assert!(matches!(error, crate::Error::InvalidHeader)), + _ => unreachable!(), + } + assert!(!target.exists()); + assert!(salvage_stages(directory.path(), "target").is_empty()); + assert_eq!(directory_snapshot(&source), source_before); + } + } + + #[test] + fn salvage_source_change_after_staging_is_fatal_and_cleans_everything() { + let directory = tempdir().unwrap(); + let source = create_verified_backup(directory.path(), false); + let source_before = directory_snapshot(&source); + let target = directory.path().join("target"); + + // Pass the first check after scanning and digesting. Mutate the open + // source at the second check, after the stage copy has been synced. + mutate_salvage_source_after_identity_checks(1); + assert!(matches!( + Store::salvage_from(&source, &target), + Err(crate::Error::SourceChanged { path }) if path == source.join("active.wlog") + )); + assert_ne!(directory_snapshot(&source), source_before); + assert!(!target.exists()); + assert!(salvage_stages(directory.path(), "target").is_empty()); + } + + #[test] + fn salvage_source_change_during_post_check_rolls_back_the_target() { + let directory = tempdir().unwrap(); + let source = create_verified_backup(directory.path(), false); + let source_before = directory_snapshot(&source); + let target = directory.path().join("target"); + + // Pass the checks after the source digest and stage copy, then mutate + // at the final source check after the published target was checked. + mutate_salvage_source_after_identity_checks(2); + assert!(matches!( + Store::salvage_from(&source, &target), + Err(crate::Error::SourceChanged { path }) if path == source.join("active.wlog") + )); + assert_ne!(directory_snapshot(&source), source_before); + assert!(!target.exists()); + assert!(salvage_stages(directory.path(), "target").is_empty()); + } + + #[test] + fn salvage_respects_an_exclusive_writer_lock_then_succeeds_after_close() { + let directory = tempdir().unwrap(); + let source = create_verified_backup(directory.path(), false); + let target = directory.path().join("target"); + let writer = Store::open(&source).unwrap(); + + assert!(matches!( + Store::salvage_from(&source, &target), + Err(crate::Error::Locked { path }) if path == source.join("active.wlog") + )); + assert!(!target.exists()); + assert!(salvage_stages(directory.path(), "target").is_empty()); + + writer.close().unwrap(); + let report = Store::salvage_from(&source, &target).unwrap(); + assert_eq!(report.status, SalvageStatus::Clean); + assert_eq!((report.recovered_commits, report.recovered_points), (2, 21)); + } + + #[test] + fn salvage_never_changes_existing_targets() { + let directory = tempdir().unwrap(); + let source = 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::salvage_from(&source, &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::salvage_from(&source, &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::salvage_from(&source, &dangling).unwrap_err(), + std::io::ErrorKind::AlreadyExists, + ); + assert_eq!( + std::fs::read_link(&dangling).unwrap(), + Path::new("missing-target") + ); + } + + #[test] + fn simultaneous_salvages_publish_exactly_one_complete_target() { + let directory = tempdir().unwrap(); + let source = create_verified_backup(directory.path(), false); + let target = directory.path().join("race-target"); + let barrier = std::sync::Arc::new(std::sync::Barrier::new(3)); + let mut threads = Vec::new(); + for _ in 0..2 { + let source = source.clone(); + let target = target.clone(); + let barrier = std::sync::Arc::clone(&barrier); + threads.push(std::thread::spawn(move || { + barrier.wait(); + Store::salvage_from(source, target) + })); + } + barrier.wait(); + let results: Vec<_> = threads + .into_iter() + .map(|thread| thread.join().unwrap()) + .collect(); + assert_eq!(results.iter().filter(|result| result.is_ok()).count(), 1); + let loser = results.into_iter().find_map(Result::err).unwrap(); + assert_io_kind(loser, std::io::ErrorKind::AlreadyExists); + let integrity = Store::open_read_only(&target) + .unwrap() + .check_integrity() + .unwrap(); + assert_eq!(integrity.raw_commits, 2); + assert_eq!(integrity.raw_points, 21); + assert!(salvage_stages(directory.path(), "race-target").is_empty()); + } + + #[test] + fn salvage_rejects_symlink_fifo_and_non_file_active_log_without_blocking() { + for kind in ["symlink", "fifo", "directory"] { + let directory = tempdir().unwrap(); + let source = directory.path().join("source"); + std::fs::create_dir(&source).unwrap(); + let active = source.join("active.wlog"); + match kind { + "symlink" => std::os::unix::fs::symlink("outside", &active).unwrap(), + "fifo" => create_fifo(&active), + "directory" => std::fs::create_dir(&active).unwrap(), + _ => unreachable!(), + } + let target = directory.path().join("target"); + let error = salvage_with_deadline(source, target.clone(), &active).unwrap_err(); + assert!(error.to_string().contains("not a regular file")); + assert!(!target.exists()); + assert!(salvage_stages(directory.path(), "target").is_empty()); + } + } + + #[test] + fn salvage_rejects_a_symlink_or_non_directory_source_root() { + let directory = tempdir().unwrap(); + let real_source = create_verified_backup(directory.path(), false); + let source_before = directory_snapshot(&real_source); + + let linked_source = directory.path().join("linked-source"); + std::os::unix::fs::symlink(&real_source, &linked_source).unwrap(); + let linked_target = directory.path().join("linked-target"); + assert!(Store::salvage_from(&linked_source, &linked_target).is_err()); + assert!(!linked_target.exists()); + assert_eq!(directory_snapshot(&real_source), source_before); + + let file_source = directory.path().join("file-source"); + std::fs::write(&file_source, b"not a directory").unwrap(); + let file_before = std::fs::read(&file_source).unwrap(); + let file_target = directory.path().join("file-target"); + assert!(Store::salvage_from(&file_source, &file_target).is_err()); + assert!(!file_target.exists()); + assert_eq!(std::fs::read(&file_source).unwrap(), file_before); + } + + #[test] + fn salvage_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 source = create_verified_backup(directory.path(), false); + let target = directory.path().join("target"); + fail_next_publication_step(step); + assert!(Store::salvage_from(&source, &target).is_err()); + assert!(!target.exists(), "{step:?} failure left a target"); + assert!( + salvage_stages(directory.path(), "target").is_empty(), + "{step:?} failure left a stage" + ); + } + } } diff --git a/tests/cli.rs b/tests/cli.rs index 31a65fb..b315f02 100644 --- a/tests/cli.rs +++ b/tests/cli.rs @@ -193,6 +193,9 @@ fn usage_errors_exit_two_on_stderr_without_creating_files() { &["restore"], &["restore", "backup"], &["restore", "backup", "target", "extra"], + &["salvage"], + &["salvage", "damaged-store"], + &["salvage", "damaged-store", "target", "extra"], &["generate"], &["bench-ftwdb"], &["bench-ftwdb", "workload"], @@ -627,6 +630,108 @@ fn restore_corruption_is_a_runtime_error_and_publishes_nothing() { assert!(!target.exists()); } +#[test] +fn salvage_partial_prefix_is_successful_json_and_passes_check_store() { + let directory = tempfile::tempdir().unwrap(); + let source = directory.path().join("source"); + let backup = directory.path().join("damaged-backup"); + let target = directory.path().join("salvaged"); + + ftw( + directory.path(), + &["generate", "workload", "--sites", "1", "--days", "1"], + ) + .assert_success(); + let loaded = ftw( + directory.path(), + &["bench-ftwdb", "workload", path_str(&source)], + ); + loaded.assert_success(); + let loaded_json = json_record(&loaded); + let points = json_u64(&loaded_json, "points"); + let expected_commits = 1 + points.div_ceil(10_000); + let checked = check_store(directory.path(), &source, points, expected_commits); + let checked_json = json_record(&checked); + let commits = json_u64(&checked_json, "raw_commits"); + + ftw( + directory.path(), + &["backup", path_str(&source), path_str(&backup)], + ) + .assert_success(); + let prefix = fs::read(backup.join("active.wlog")).unwrap(); + let mut active = fs::OpenOptions::new() + .append(true) + .open(backup.join("active.wlog")) + .unwrap(); + active.write_all(b"partial").unwrap(); + active.sync_all().unwrap(); + drop(active); + let source_before = snapshot_tree(&backup); + + let salvaged = ftw( + directory.path(), + &["salvage", path_str(&backup), path_str(&target)], + ); + salvaged.assert_success(); + let record = json_record(&salvaged); + assert_eq!(json_string(&record, "format"), "ftwdb-salvage-v1"); + assert_eq!(json_string(&record, "status"), "partial"); + assert_eq!(json_u64(&record, "source_bytes"), prefix.len() as u64 + 7); + assert_eq!( + json_u64(&record, "recovered_prefix_bytes"), + prefix.len() as u64 + ); + assert_eq!(json_u64(&record, "discarded_bytes"), 7); + assert_eq!(json_u64(&record, "stop_offset"), prefix.len() as u64); + assert_eq!( + json_string(&record, "stop_reason"), + "incomplete-frame-header" + ); + assert_eq!(json_u64(&record, "recovered_commits"), commits); + assert_eq!(json_u64(&record, "recovered_points"), points); + let source_crc = json_string(&record, "source_prefix_crc32"); + assert_eq!(source_crc.len(), 8); + assert_eq!( + source_crc, + json_string(&record, "destination_snapshot_crc32") + ); + assert_eq!(fs::read(target.join("active.wlog")).unwrap(), prefix); + check_store(directory.path(), &target, points, commits); + assert_eq!(snapshot_tree(&backup), source_before); + + let existing = directory.path().join("existing-target"); + fs::create_dir(&existing).unwrap(); + fs::write(existing.join("keep"), b"unchanged").unwrap(); + let existing_before = snapshot_tree(&existing); + let refused = ftw( + directory.path(), + &["salvage", path_str(&backup), path_str(&existing)], + ); + assert_runtime_error(&refused); + assert!(refused.stderr.contains("already exists")); + assert_eq!(snapshot_tree(&existing), existing_before); +} + +#[test] +fn salvage_invalid_database_header_exits_one_and_publishes_nothing() { + let directory = tempfile::tempdir().unwrap(); + let source = directory.path().join("damaged"); + let target = directory.path().join("target"); + fs::create_dir(&source).unwrap(); + fs::write(source.join("active.wlog"), b"not an FTWDB header").unwrap(); + let source_before = snapshot_tree(&source); + + let output = ftw( + directory.path(), + &["salvage", path_str(&source), path_str(&target)], + ); + assert_runtime_error(&output); + assert!(output.stderr.contains("invalid FTWDB file header")); + assert!(!target.exists()); + assert_eq!(snapshot_tree(&source), source_before); +} + #[test] fn read_only_commands_do_not_create_missing_paths() { let directory = tempfile::tempdir().unwrap(); diff --git a/tests/real_fixture.rs b/tests/real_fixture.rs index 95da71e..b1b7272 100644 --- a/tests/real_fixture.rs +++ b/tests/real_fixture.rs @@ -1,5 +1,5 @@ use flate2::read::GzDecoder; -use ftwdb::{Config, Durability, Store, load_real_fixture}; +use ftwdb::{Config, Durability, SalvageStatus, SalvageStopReason, Store, load_real_fixture}; use std::fs::File; use std::io::BufReader; use std::path::PathBuf; @@ -12,6 +12,7 @@ fn committed_real_fixture_loads_with_stable_identity() { let source = directory.path().join("source"); let backup = directory.path().join("backup"); let restored = directory.path().join("restored"); + let salvaged = directory.path().join("salvaged"); let mut store = Store::open_with( &source, Config { @@ -52,4 +53,23 @@ fn committed_real_fixture_loads_with_stable_identity() { .unwrap(); assert_eq!(restored_integrity.raw_points, 889_978); assert_eq!(restored_integrity.raw_commits, 89); + + let salvage = Store::salvage_from(&backup, &salvaged).unwrap(); + assert_eq!(salvage.status, SalvageStatus::Clean); + assert_eq!(salvage.stop_reason, SalvageStopReason::CleanEof); + assert_eq!(salvage.source_bytes, salvage.recovered_prefix_bytes); + assert_eq!(salvage.discarded_bytes, 0); + assert_eq!(salvage.recovered_points, 889_978); + assert_eq!(salvage.recovered_commits, 89); + assert_eq!(salvage.source_prefix_crc32, 0x9ff1_a95b); + assert_eq!( + salvage.source_prefix_crc32, + salvage.destination_snapshot_crc32 + ); + let salvaged_integrity = Store::open_read_only(&salvaged) + .unwrap() + .check_integrity() + .unwrap(); + assert_eq!(salvaged_integrity.raw_points, 889_978); + assert_eq!(salvaged_integrity.raw_commits, 89); }