backup.rs
raw
//! Integration tests driving the backup pipeline end to end: a real
//! directory tree, a real repository, and real suspend/resume — with the
//! filesystem changing while the backup is suspended.
mod common;
use std::{
path::{Path, PathBuf},
sync::{Arc, atomic::AtomicBool, atomic::Ordering},
};
use beeping::pipeline::{EarlyFinish, Outcome, backup};
use repository::EntryKind;
use common::{
StoreAnswers, deterministic_bytes, manifest_map, open_repository, open_small_repository,
open_store, read_back, run_backup,
};
#[test]
fn full_backup_roundtrip() {
let dir = tempfile::tempdir().unwrap();
let root = dir.path().join("tree");
// A tree exercising every entry kind plus exclusion
std::fs::create_dir_all(root.join("sub/deeper")).unwrap();
std::fs::create_dir_all(root.join("empty")).unwrap();
std::fs::write(root.join("small.txt"), b"hello backup").unwrap();
std::fs::write(root.join("sub/nested.txt"), b"nested content").unwrap();
std::fs::write(root.join("sub/deeper/duplicate.bin"), b"same bytes").unwrap();
std::fs::write(root.join("duplicate.bin"), b"same bytes").unwrap();
std::fs::write(root.join("skipme.log"), b"not backed up").unwrap();
let big = deterministic_bytes(6 * 1024 * 1024, 42);
std::fs::write(root.join("big.bin"), &big).unwrap();
#[cfg(unix)]
{
use std::os::unix::ffi::OsStringExt as _;
let weird = std::ffi::OsString::from_vec(vec![b'f', 0xFF, 0xFE, b'x']);
std::fs::write(root.join(weird), b"non-utf8 name").unwrap();
std::os::unix::fs::symlink("small.txt", root.join("link")).unwrap();
}
let repository = open_repository(dir.path());
let suspend = Arc::new(AtomicBool::new(false));
let outcome = run_backup(
&repository,
&root,
&dir.path().join("state"),
&["*.log"],
&suspend,
);
let Outcome::Completed(snapshot) = outcome else {
panic!("expected completion, got {outcome:?}");
};
// The state directory is cleaned up after completion
assert!(!dir.path().join("state").exists());
let map = manifest_map(&repository, snapshot);
assert!(map.contains_key(Path::new("sub")));
assert!(map.contains_key(Path::new("sub/deeper")));
assert!(map.contains_key(Path::new("empty")));
assert!(!map.contains_key(Path::new("skipme.log")));
assert_eq!(
read_back(&repository, &map[Path::new("small.txt")]),
b"hello backup"
);
assert_eq!(read_back(&repository, &map[Path::new("big.bin")]), big);
// Identical files deduplicate to the same chunks
let EntryKind::File { chunks: a, .. } = &map[Path::new("duplicate.bin")].kind else {
panic!("expected file");
};
let EntryKind::File { chunks: b, .. } = &map[Path::new("sub/deeper/duplicate.bin")].kind else {
panic!("expected file");
};
assert_eq!(a, b);
#[cfg(unix)]
{
use std::os::unix::ffi::OsStringExt as _;
let weird = PathBuf::from(std::ffi::OsString::from_vec(vec![b'f', 0xFF, 0xFE, b'x']));
assert_eq!(
read_back(&repository, &map[&weird]),
b"non-utf8 name",
"non-UTF-8 file names must survive the manifest"
);
let EntryKind::Symlink { target } = &map[Path::new("link")].kind else {
panic!("expected symlink entry");
};
assert_eq!(target, Path::new("small.txt"));
let mode = map[Path::new("small.txt")].mode.unwrap();
assert_ne!(mode & 0o400, 0, "owner-readable bit should be recorded");
}
assert!(map[Path::new("small.txt")].mtime.is_some());
}
#[test]
fn suspend_resume_with_filesystem_changes() {
let dir = tempfile::tempdir().unwrap();
let root = dir.path().join("tree");
let state = dir.path().join("state");
// Enough files that a resumed run has real work left
for outer in 0..10 {
for inner in 0..10 {
let sub = root.join(format!("dir-{outer}/sub-{inner}"));
std::fs::create_dir_all(&sub).unwrap();
for file in 0..3 {
std::fs::write(
sub.join(format!("file-{file}.dat")),
deterministic_bytes(2000, (outer * 100 + inner * 10 + file) as u64),
)
.unwrap();
}
}
}
std::fs::write(root.join("doomed.txt"), b"will be deleted").unwrap();
std::fs::write(root.join("mutant.txt"), b"original content").unwrap();
let repository = open_repository(dir.path());
// With the suspend flag pre-set, the controller stops at its first
// tick: deterministic suspension with ~one tick of work done.
let suspend = Arc::new(AtomicBool::new(true));
let outcome = run_backup(&repository, &root, &state, &[], &suspend);
assert!(
matches!(outcome, Outcome::Suspended { .. }),
"got {outcome:?}"
);
assert!(state.exists(), "suspended state must persist");
// The filesystem changes while the backup is suspended
std::fs::remove_file(root.join("doomed.txt")).unwrap();
std::fs::write(root.join("mutant.txt"), b"changed while suspended").unwrap();
std::fs::write(root.join("newcomer.txt"), b"added while suspended").unwrap();
std::fs::remove_dir_all(root.join("dir-9")).unwrap();
suspend.store(false, Ordering::Release);
let outcome = run_backup(&repository, &root, &state, &[], &suspend);
let Outcome::Completed(snapshot) = outcome else {
panic!("expected completion, got {outcome:?}");
};
assert!(!state.exists());
let map = manifest_map(&repository, snapshot);
// Every recorded file's content must be fully retrievable, whatever
// mix of before/after the run captured
for entry in map.values() {
if matches!(entry.kind, EntryKind::File { .. }) {
read_back(&repository, entry);
}
}
// Files that never changed are all present and correct
for outer in 0..9 {
for inner in 0..10 {
for file in 0..3 {
let path = PathBuf::from(format!("dir-{outer}/sub-{inner}/file-{file}.dat"));
let entry = map
.get(&path)
.unwrap_or_else(|| panic!("{path:?} missing from manifest"));
assert_eq!(
read_back(&repository, entry),
deterministic_bytes(2000, (outer * 100 + inner * 10 + file) as u64)
);
}
}
}
// The mutated file holds one of its two valid read-time states
if let Some(entry) = map.get(Path::new("mutant.txt")) {
let content = read_back(&repository, entry);
assert!(
content == b"original content" || content == b"changed while suspended",
"mutant.txt holds neither version: {content:?}"
);
}
// The deleted file, if captured before deletion, holds its old
// content
if let Some(entry) = map.get(Path::new("doomed.txt")) {
assert_eq!(read_back(&repository, entry), b"will be deleted");
}
}
/// With the backlog limit tightened to 2, the scanner spends the whole
/// run inside the pacing gate; the pipeline must still finish, record
/// everything exactly once, and survive suspension while paced (the
/// gate is a bail-out point that leaves a directory to be rescanned).
#[test]
fn pacing_preserves_completeness_and_liveness() {
let dir = tempfile::tempdir().unwrap();
let root = dir.path().join("tree");
let state = dir.path().join("state");
for outer in 0..5 {
let sub = root.join(format!("dir-{outer}"));
std::fs::create_dir_all(&sub).unwrap();
for file in 0..40 {
std::fs::write(
sub.join(format!("file-{file:02}.dat")),
deterministic_bytes(500, (outer * 1000 + file) as u64),
)
.unwrap();
}
}
let repository = open_repository(dir.path());
let run = |suspend: &Arc<AtomicBool>| {
let mut options = backup::BackupOptions::new(&root, &state);
options.chunk_backlog_limit = 2;
let (events, receiver) = crossbeam_channel::unbounded();
drop(receiver);
backup::run(repository.clone(), options, events, suspend.clone()).unwrap()
};
// First run suspends while the scanner is (almost certainly) paced
let suspend = Arc::new(AtomicBool::new(true));
assert!(matches!(run(&suspend), Outcome::Suspended { .. }));
suspend.store(false, Ordering::Release);
let Outcome::Completed(snapshot) = run(&suspend) else {
panic!("expected completion");
};
// manifest_map asserts no duplicate paths; check nothing is missing
let map = manifest_map(&repository, snapshot);
for outer in 0..5 {
for file in 0..40 {
let path = PathBuf::from(format!("dir-{outer}/file-{file:02}.dat"));
let entry = map
.get(&path)
.unwrap_or_else(|| panic!("{path:?} missing from manifest"));
assert_eq!(
read_back(&repository, entry),
deterministic_bytes(500, (outer * 1000 + file) as u64)
);
}
}
}
/// With the upload channel squeezed to one slot and a single uploader,
/// every chunker blocks on the handoff; the run must still complete,
/// survive a suspension (exercising the bail-out paths in the send loop
/// and the per-file upload wait), and record every byte.
#[test]
fn upload_backpressure_preserves_completeness_and_liveness() {
let dir = tempfile::tempdir().unwrap();
let root = dir.path().join("tree");
let state = dir.path().join("state");
std::fs::create_dir_all(&root).unwrap();
// A multi-chunk file so one file's uploads span several channel
// handoffs, plus plenty of single-chunk files
let big = deterministic_bytes(6 * 1024 * 1024, 99);
std::fs::write(root.join("big.bin"), &big).unwrap();
for file in 0..60 {
std::fs::write(
root.join(format!("small-{file:02}.dat")),
deterministic_bytes(1500, file),
)
.unwrap();
}
let repository = open_repository(dir.path());
let run = |suspend: &Arc<AtomicBool>| {
let mut options = backup::BackupOptions::new(&root, &state);
options.upload_backlog_limit = 1;
options.uploader_threads = 1;
options.chunk_backlog_limit = 2;
let (events, receiver) = crossbeam_channel::unbounded();
drop(receiver);
backup::run(repository.clone(), options, events, suspend.clone()).unwrap()
};
let suspend = Arc::new(AtomicBool::new(true));
assert!(matches!(run(&suspend), Outcome::Suspended { .. }));
suspend.store(false, Ordering::Release);
let Outcome::Completed(snapshot) = run(&suspend) else {
panic!("expected completion");
};
let map = manifest_map(&repository, snapshot);
assert_eq!(read_back(&repository, &map[Path::new("big.bin")]), big);
for file in 0..60 {
let path = PathBuf::from(format!("small-{file:02}.dat"));
assert_eq!(
read_back(&repository, &map[&path]),
deterministic_bytes(1500, file),
"content mismatch for {path:?}"
);
}
}
/// A backup whose own state directory and (local) repository both live
/// inside the tree being backed up must complete safely and capture
/// neither — a repository backed up into itself would grow
/// geometrically, and live queue files are transient noise.
#[test]
fn own_state_and_repository_are_not_captured() {
let dir = tempfile::tempdir().unwrap();
let root = dir.path().join("tree");
std::fs::create_dir_all(root.join("docs")).unwrap();
std::fs::write(root.join("docs/real.txt"), b"actual data").unwrap();
std::fs::write(root.join("top.txt"), b"more data").unwrap();
// Both self-referential paths live inside the backup root
let repository = Arc::new(
repository::Repository::create(
Box::new(repository::LocalBackend::new(root.join("repo")).unwrap()),
"test-password",
)
.unwrap(),
);
let state = root.join("state");
let options = backup::BackupOptions::new(&root, &state);
let (events, receiver) = crossbeam_channel::unbounded();
let outcome = backup::run(
repository.clone(),
options,
events,
Arc::new(AtomicBool::new(false)),
)
.unwrap();
let Outcome::Completed(snapshot) = outcome else {
panic!("expected completion, got {outcome:?}");
};
let map = manifest_map(&repository, snapshot);
assert_eq!(
read_back(&repository, &map[Path::new("docs/real.txt")]),
b"actual data"
);
assert_eq!(
read_back(&repository, &map[Path::new("top.txt")]),
b"more data"
);
for path in map.keys() {
assert!(
!path.starts_with("repo") && !path.starts_with("state"),
"self-referential path {path:?} was captured"
);
}
// The skips were reported, not silent
let internal_skips = receiver
.try_iter()
.filter(|event| {
matches!(
event,
beeping::pipeline::Event::Skipped {
reason: beeping::pipeline::SkipReason::Internal,
..
}
)
})
.count();
assert_eq!(
internal_skips, 2,
"expected exactly the repository and the state dir to be skipped"
);
}
/// Large files are read a limited number at a time. Concurrency buys
/// nothing once the backend is the bottleneck, and every part-read file
/// is work that suspending throws away, so the pipeline holds only a
/// few open at once and lets the rest wait unopened.
#[test]
fn large_files_are_read_a_few_at_a_time() {
use beeping::pipeline::{Event, Stage};
use std::collections::HashSet;
let dir = tempfile::tempdir().unwrap();
let root = dir.path().join("tree");
std::fs::create_dir_all(&root).unwrap();
let mut large: HashSet<PathBuf> = HashSet::new();
for index in 0..6 {
let name = format!("large-{index}.dat");
std::fs::write(root.join(&name), deterministic_bytes(3 << 20, index)).unwrap();
large.insert(PathBuf::from(name));
}
// Small files share the pool, and must not be held up by the limit
for index in 0..60 {
std::fs::write(
root.join(format!("small-{index}.dat")),
deterministic_bytes(1000, 500 + index),
)
.unwrap();
}
let repository = open_repository(dir.path());
let mut options = backup::BackupOptions::new(&root, dir.path().join("state"));
options.chunker_threads = 6;
options.large_file_threshold = 1 << 20;
options.concurrent_large_files = Some(1);
let (events, receiver) = crossbeam_channel::unbounded();
let outcome = backup::run(
repository.clone(),
options,
events,
Arc::new(AtomicBool::new(false)),
)
.unwrap();
let Outcome::Completed(snapshot) = outcome else {
panic!("expected completion, got {outcome:?}");
};
// A large file is in a worker's hands from the moment the stage
// announces it until its entry is recorded
let mut in_hand: HashSet<PathBuf> = HashSet::new();
let mut workers: HashSet<usize> = HashSet::new();
let mut most = 0;
for event in receiver.try_iter() {
match event {
Event::Began {
stage: Stage::Chunk,
worker,
path,
} if large.contains(&path) => {
in_hand.insert(path);
workers.insert(worker);
most = most.max(in_hand.len());
}
Event::Stored { path, .. } => {
in_hand.remove(&path);
}
_ => {}
}
}
assert_eq!(
most, 1,
"the limit was exceeded: {most} large files at once"
);
assert!(
workers.len() > 1,
"the limit pinned every large file to one worker, rather than \
letting the pool take turns"
);
// And everything still got backed up
let map = manifest_map(&repository, snapshot);
for path in &large {
assert!(map.contains_key(path), "{path:?} missing");
}
assert_eq!(map.len(), 66);
}
/// A suspended backup is a backup: what it had stored is restorable
/// from the repository alone, without the machine it ran on.
#[test]
fn a_suspended_backup_can_be_restored_from() {
use beeping::pipeline::restore;
use repository::Coverage;
let dir = tempfile::tempdir().unwrap();
let root = dir.path().join("tree");
let state = dir.path().join("state");
let mut written: Vec<(PathBuf, Vec<u8>)> = Vec::new();
for outer in 0..8 {
let sub = root.join(format!("dir-{outer}"));
std::fs::create_dir_all(&sub).unwrap();
for file in 0..25 {
let path = sub.join(format!("file-{file}.dat"));
let bytes = deterministic_bytes(4000, (outer * 40 + file) as u64);
std::fs::write(&path, &bytes).unwrap();
written.push((path.strip_prefix(&root).unwrap().to_owned(), bytes));
}
}
let repository = open_repository(dir.path());
let run = |suspend: bool| {
let (events, receiver) = crossbeam_channel::unbounded();
drop(receiver);
backup::run(
repository.clone(),
backup::BackupOptions::new(&root, &state),
events,
Arc::new(AtomicBool::new(suspend)),
)
.unwrap()
};
assert!(matches!(run(true), Outcome::Suspended { .. }));
// Suspending published exactly one snapshot, and it says what it is
let published = repository.snapshots().unwrap();
assert_eq!(published.len(), 1, "a suspended run leaves one snapshot");
let partial = repository.load_snapshot(published[0]).unwrap();
assert_eq!(partial.coverage, Coverage::Partial);
assert!(partial.entries > 0, "it covers something");
// Everything it claims is really there, byte for byte
let target = dir.path().join("recovered");
let outcome = restore::run(
repository.clone(),
restore::RestoreOptions::new(partial.id, &target, dir.path().join("restore-state")),
crossbeam_channel::unbounded().0,
Arc::new(AtomicBool::new(false)),
)
.unwrap();
assert!(matches!(outcome, Outcome::Completed(_)));
let covered = manifest_map(&repository, partial.id);
let mut recovered = 0;
for (path, bytes) in &written {
let Some(entry) = covered.get(path) else {
// Not backed up before the suspension; must not be invented
assert!(
!target.join(path).exists(),
"{path:?} was restored but is not in the manifest"
);
continue;
};
assert!(matches!(entry.kind, EntryKind::File { .. }));
assert_eq!(
&std::fs::read(target.join(path)).unwrap(),
bytes,
"{path:?}"
);
recovered += 1;
}
assert!(recovered > 0, "the partial snapshot recovered nothing");
// Resuming to completion replaces the partial with a complete one
let Outcome::Completed(complete) = run(false) else {
panic!("expected completion");
};
let published = repository.snapshots().unwrap();
assert_eq!(
published,
vec![complete],
"the complete snapshot supersedes every partial one"
);
let final_map = manifest_map(&repository, complete);
assert_eq!(
repository.load_snapshot(complete).unwrap().coverage,
Coverage::Complete
);
for (path, _) in &written {
assert!(
final_map.contains_key(path),
"{path:?} missing from the finished backup"
);
}
}
/// A run suspended by beeping 0.2 left bare totals behind, with no
/// record of running time. Resuming it under a version that keeps both
/// must carry those totals forward: a permissive reader decodes the old
/// file into a zeroed session quite happily, silently discarding what
/// may be weeks of a long backup's history.
#[test]
fn totals_from_an_older_version_survive_the_resume() {
use beeping::pipeline::{Event, Progress};
let dir = tempfile::tempdir().unwrap();
let root = dir.path().join("tree");
let state = dir.path().join("state");
std::fs::create_dir_all(root.join("sub")).unwrap();
for file in 0..20 {
std::fs::write(
root.join("sub").join(format!("file-{file}.dat")),
deterministic_bytes(1500, file as u64),
)
.unwrap();
}
let repository = open_repository(dir.path());
let run = |suspend: bool| {
let (events, receiver) = crossbeam_channel::unbounded();
let outcome = backup::run(
repository.clone(),
backup::BackupOptions::new(&root, &state),
events,
Arc::new(AtomicBool::new(suspend)),
)
.unwrap();
(outcome, receiver.try_iter().collect::<Vec<Event>>())
};
let (outcome, _) = run(true);
assert!(matches!(outcome, Outcome::Suspended { .. }));
// Rewrite the state directory's totals the way beeping 0.2 did:
// the counters alone, with no elapsed time beside them
let legacy = Progress {
scanned: 900_001,
stored_files: 4_500_000,
stored_bytes: 2_500_000_000_000,
restored_files: 0,
restored_bytes: 0,
skipped: 77,
};
let mut encoded = Vec::new();
ciborium::into_writer(&legacy, &mut encoded).unwrap();
std::fs::write(state.join("progress"), encoded).unwrap();
let (outcome, events) = run(false);
assert!(matches!(outcome, Outcome::Completed(_)));
let resumed = events
.iter()
.find_map(|event| match event {
Event::Resumed(session) => Some(session),
_ => None,
})
.expect("resuming reports the baseline it inherited");
assert_eq!(
resumed.progress, legacy,
"the older version's totals must come through intact"
);
assert_eq!(
resumed.elapsed,
std::time::Duration::ZERO,
"0.2 recorded no running time, and none may be invented"
);
// And the run goes on counting from there: its final totals are the
// inherited ones plus exactly the work this session reported
let (files, bytes) = events
.iter()
.fold((0, 0), |(files, bytes), event| match event {
Event::Stored { bytes: stored, .. } => (files + 1, bytes + stored),
_ => (files, bytes),
});
let last = events
.iter()
.rev()
.find_map(|event| match event {
Event::Totals(session) => Some(session),
_ => None,
})
.expect("the run reports its totals");
assert_eq!(last.progress.stored_files, legacy.stored_files + files);
assert_eq!(last.progress.stored_bytes, legacy.stored_bytes + bytes);
assert!(last.elapsed > std::time::Duration::ZERO);
}
/// What a display needs, a run must actually say: how many workers each
/// stage has, which item each of them has in hand, how much is queued
/// behind them, and totals that agree with the per-item reports.
#[test]
fn a_run_reports_what_it_is_doing() {
use beeping::pipeline::{Event, Progress, Stage};
use std::collections::{HashMap, HashSet};
let dir = tempfile::tempdir().unwrap();
let root = dir.path().join("tree");
for outer in 0..4 {
let sub = root.join(format!("dir-{outer}"));
std::fs::create_dir_all(&sub).unwrap();
for file in 0..10 {
std::fs::write(
sub.join(format!("file-{file}.dat")),
deterministic_bytes(3000, (outer * 20 + file) as u64),
)
.unwrap();
}
}
let repository = open_repository(dir.path());
let mut options = backup::BackupOptions::new(&root, dir.path().join("state"));
options.scanner_threads = 2;
options.chunker_threads = 3;
options.uploader_threads = 2;
let (events, receiver) = crossbeam_channel::unbounded();
let outcome = backup::run(
repository,
options,
events,
Arc::new(AtomicBool::new(false)),
)
.unwrap();
assert!(matches!(outcome, Outcome::Completed(_)));
let events: Vec<Event> = receiver.try_iter().collect();
// Every staffed stage announces its pool exactly once
let staffing: HashMap<Stage, usize> = events
.iter()
.filter_map(|event| match event {
Event::Staffing { stage, threads } => Some((*stage, *threads)),
_ => None,
})
.collect();
assert_eq!(staffing.get(&Stage::Scan), Some(&2));
assert_eq!(staffing.get(&Stage::Chunk), Some(&3));
assert_eq!(staffing.get(&Stage::Upload), Some(&2));
// Workers name their items, and stay within their stage's pool. The
// finalization stages report the same way without a pool of their
// own, so their staffing is absent rather than zero.
let mut started: HashMap<Stage, usize> = HashMap::new();
let mut in_hand: HashSet<(Stage, usize)> = HashSet::new();
for event in &events {
match event {
Event::Began { stage, worker, .. } => {
if let Some(threads) = staffing.get(stage) {
assert!(
worker < threads,
"{stage:?} worker {worker} is outside its pool"
);
}
*started.entry(*stage).or_default() += 1;
in_hand.insert((*stage, *worker));
}
Event::Idle { stage, worker } => {
in_hand.remove(&(*stage, *worker));
}
_ => {}
}
}
assert_eq!(
started.get(&Stage::Scan),
Some(&5),
"root and four subtrees"
);
assert_eq!(started.get(&Stage::Chunk), Some(&40));
assert!(started.get(&Stage::Upload).is_some_and(|count| *count > 0));
assert!(
started.contains_key(&Stage::Record),
"finalization names the entries it folds, like any other worker"
);
assert!(
in_hand.is_empty(),
"no worker may be left looking busy: {in_hand:?}"
);
// The backlog is reported for every stage, the holding queue
// included, and the entries pile up there before finalization
let deepest_record = events
.iter()
.filter_map(|event| match event {
Event::Backlog(stages) => stages
.iter()
.find(|backlog| backlog.stage == Stage::Record)
.map(|backlog| backlog.waiting),
_ => None,
})
.max();
assert!(
deepest_record.is_some_and(|waiting| waiting > 0),
"manifest entries waiting for finalization were never reported"
);
for stage in [Stage::Scan, Stage::Chunk, Stage::Upload] {
assert!(
events.iter().any(|event| matches!(
event,
Event::Backlog(stages) if stages.iter().any(|backlog| backlog.stage == stage)
)),
"{stage:?} never reported its backlog"
);
}
// The totals the run keeps for itself must match what it reported
// item by item, or a display would show one of two different truths
let mut folded = Progress::default();
let mut transferred = 0u64;
for event in &events {
match event {
Event::Scanned { .. } => folded.scanned += 1,
Event::Stored { bytes, .. } => {
folded.stored_files += 1;
folded.stored_bytes += bytes;
}
Event::Skipped { .. } => folded.skipped += 1,
Event::Transferred { bytes } => transferred += bytes,
_ => {}
}
}
let last_totals = events
.iter()
.rev()
.find_map(|event| match event {
Event::Totals(session) => Some(session),
_ => None,
})
.expect("the run reports its totals");
assert_eq!(last_totals.progress, folded);
assert!(last_totals.elapsed > std::time::Duration::ZERO);
assert!(
transferred >= folded.stored_bytes,
"sealed chunks carry at least the content they hold"
);
}
/// A resumed run's progress totals continue from the previous
/// session's, rather than restarting at zero: the baseline delivered by
/// Event::Resumed must equal exactly what session one reported.
#[test]
fn progress_continues_across_resume() {
use beeping::pipeline::{Event, Progress, Session};
let dir = tempfile::tempdir().unwrap();
let root = dir.path().join("tree");
let state = dir.path().join("state");
for outer in 0..6 {
let sub = root.join(format!("dir-{outer}"));
std::fs::create_dir_all(&sub).unwrap();
for file in 0..15 {
std::fs::write(
sub.join(format!("file-{file}.dat")),
deterministic_bytes(2000, (outer * 50 + file) as u64),
)
.unwrap();
}
}
let repository = open_repository(dir.path());
let run = |suspend: bool| {
let options = backup::BackupOptions::new(&root, &state);
let (events, receiver) = crossbeam_channel::unbounded();
let flag = Arc::new(AtomicBool::new(suspend));
let outcome = backup::run(repository.clone(), options, events, flag).unwrap();
let events: Vec<Event> = receiver.try_iter().collect();
(outcome, events)
};
// Fold the per-item events independently of the totals the pipeline
// keeps for itself, so the two must agree
let fold = |events: &[Event]| {
let mut totals = Progress::default();
for event in events {
match event {
Event::Scanned { .. } => totals.scanned += 1,
Event::Stored { bytes, .. } => {
totals.stored_files += 1;
totals.stored_bytes += bytes;
}
Event::Skipped { .. } => totals.skipped += 1,
_ => {}
}
}
totals
};
let (outcome, first_events) = run(true);
assert!(matches!(outcome, Outcome::Suspended { .. }));
assert!(
!first_events
.iter()
.any(|event| matches!(event, Event::Resumed(_))),
"a fresh run must not claim to be resumed"
);
let session_one = fold(&first_events);
let (outcome, second_events) = run(false);
assert!(matches!(outcome, Outcome::Completed(_)));
let baselines: Vec<&Session> = second_events
.iter()
.filter_map(|event| match event {
Event::Resumed(session) => Some(session),
_ => None,
})
.collect();
assert_eq!(baselines.len(), 1, "exactly one baseline event");
assert_eq!(
baselines[0].progress, session_one,
"the resumed baseline must equal session one's reported totals"
);
assert!(
baselines[0].elapsed > std::time::Duration::ZERO,
"the baseline must carry session one's running time"
);
// Combined totals cover the whole tree (duplicates from a
// mid-directory suspension can only push the counts higher)
let combined_stored = session_one.stored_files + fold(&second_events).stored_files;
assert!(
combined_stored >= 90,
"90 files in the tree, but only {combined_stored} stored events in total"
);
}
/// The help text promises: excluding a directory prunes everything
/// beneath it, while excluding `dir/**` spares the directory itself
/// and drops only its contents. Hold the pipeline to that.
#[test]
fn directory_exclusion_semantics() {
let dir = tempfile::tempdir().unwrap();
let root = dir.path().join("tree");
std::fs::create_dir_all(root.join("cache/deep")).unwrap();
std::fs::write(root.join("cache/a.txt"), b"cached").unwrap();
std::fs::write(root.join("cache/deep/b.txt"), b"cached deeper").unwrap();
std::fs::create_dir_all(root.join("tmp")).unwrap();
std::fs::write(root.join("tmp/scratch.txt"), b"scratch").unwrap();
std::fs::write(root.join("keep.txt"), b"kept").unwrap();
let repository = open_repository(dir.path());
let suspend = Arc::new(AtomicBool::new(false));
let outcome = run_backup(
&repository,
&root,
&dir.path().join("state"),
&["cache", "tmp/**"],
&suspend,
);
let Outcome::Completed(snapshot) = outcome else {
panic!("expected completion, got {outcome:?}");
};
let map = manifest_map(&repository, snapshot);
assert!(map.contains_key(Path::new("keep.txt")));
// `cache` removes the directory and, with it, everything beneath
assert!(!map.contains_key(Path::new("cache")));
assert!(!map.contains_key(Path::new("cache/a.txt")));
assert!(!map.contains_key(Path::new("cache/deep")));
assert!(!map.contains_key(Path::new("cache/deep/b.txt")));
// `tmp/**` drops the contents but keeps the (now empty) directory
assert!(matches!(
map.get(Path::new("tmp")).map(|entry| &entry.kind),
Some(EntryKind::Directory)
));
assert!(!map.contains_key(Path::new("tmp/scratch.txt")));
}
/// Excludes changed between suspension and resume take effect for
/// everything not yet processed, while items already stored stay in
/// the snapshot. The invariant is exact: a file matching the new
/// exclude appears in the manifest if and only if session one already
/// stored it.
#[test]
fn excludes_changed_across_resume_apply_prospectively() {
use beeping::pipeline::Event;
let dir = tempfile::tempdir().unwrap();
let root = dir.path().join("tree");
let state = dir.path().join("state");
// Plenty of work, .skip files interleaved throughout, so the first
// session cannot plausibly finish and some .skip files stay queued
let mut skip_files = Vec::new();
for outer in 0..25 {
let sub = root.join(format!("dir-{outer:02}"));
std::fs::create_dir_all(&sub).unwrap();
for file in 0..20 {
std::fs::write(
sub.join(format!("file-{file:02}.dat")),
deterministic_bytes(2000, (outer * 100 + file) as u64),
)
.unwrap();
let skip = format!("dir-{outer:02}/file-{file:02}.skip");
std::fs::write(root.join(&skip), b"maybe excluded later").unwrap();
skip_files.push(PathBuf::from(skip));
}
}
let repository = open_repository(dir.path());
let run = |exclude: &[&str], suspend: bool| {
let mut options = backup::BackupOptions::new(&root, &state);
options.exclude = beeping::patterns::compile_set(exclude).unwrap();
let (events, receiver) = crossbeam_channel::unbounded();
let outcome = backup::run(
repository.clone(),
options,
events,
Arc::new(AtomicBool::new(suspend)),
)
.unwrap();
(outcome, receiver.try_iter().collect::<Vec<Event>>())
};
// Session one: no excludes, suspended after ~one controller tick
let (outcome, first_events) = run(&[], true);
assert!(matches!(outcome, Outcome::Suspended { .. }));
let stored_in_first: std::collections::HashSet<PathBuf> = first_events
.iter()
.filter_map(|event| match event {
Event::Stored { path, .. } => Some(path.clone()),
_ => None,
})
.collect();
// Session two: .skip files are now excluded
let (outcome, second_events) = run(&["*.skip"], false);
let Outcome::Completed(snapshot) = outcome else {
panic!("expected completion, got {outcome:?}");
};
let map = manifest_map(&repository, snapshot);
for skip in &skip_files {
assert_eq!(
map.contains_key(skip.as_path()),
stored_in_first.contains(skip),
"{skip:?}: newly-excluded files belong in the snapshot exactly \
when they were stored before the exclude existed"
);
}
// The re-check actually fired on previously-queued work
let excluded_in_second = second_events
.iter()
.filter(|event| {
matches!(
event,
Event::Skipped {
reason: beeping::pipeline::SkipReason::Excluded,
..
}
)
})
.count();
assert!(
excluded_in_second > 0,
"session two should have excluded something"
);
// Unaffected files are all present regardless of session
for outer in 0..25 {
for file in 0..20 {
let path = PathBuf::from(format!("dir-{outer:02}/file-{file:02}.dat"));
assert!(map.contains_key(&path), "{path:?} missing");
}
}
}
#[test]
fn state_dir_refuses_a_different_root() {
let dir = tempfile::tempdir().unwrap();
let root_a = dir.path().join("tree-a");
let root_b = dir.path().join("tree-b");
let state = dir.path().join("state");
for root in [&root_a, &root_b] {
std::fs::create_dir_all(root).unwrap();
std::fs::write(root.join("file.txt"), b"content").unwrap();
}
let repository = open_repository(dir.path());
// Suspend a backup of tree-a
let suspend = Arc::new(AtomicBool::new(true));
let outcome = run_backup(&repository, &root_a, &state, &[], &suspend);
assert!(matches!(outcome, Outcome::Suspended { .. }));
// Trying to resume it as a backup of tree-b must fail loudly
let mut options = backup::BackupOptions::new(&root_b, &state);
options.exclude = globset::GlobSet::empty();
let (events, _receiver) = crossbeam_channel::unbounded();
let suspend = Arc::new(AtomicBool::new(false));
match backup::run(repository, options, events, suspend) {
Err(beeping::pipeline::Error::StateMismatch { .. }) => {}
other => panic!("expected StateMismatch, got {other:?}"),
}
}
/// A run publishes what it has as it goes, so an interrupted backup is
/// restorable from the repository even if it was never suspended
/// politely — and what it publishes is whole: every chunk each partial
/// snapshot names is already stored.
///
/// Partial snapshots are retired as their successors supersede them, so
/// this has to be watched while the run is happening; afterwards there
/// is nothing left to look at.
#[test]
fn a_long_run_publishes_restorable_snapshots_as_it_goes() {
let dir = tempfile::tempdir().unwrap();
let root = dir.path().join("tree");
std::fs::create_dir_all(&root).unwrap();
for index in 0..800u64 {
std::fs::write(
root.join(format!("file-{index:04}.dat")),
deterministic_bytes(2048, index),
)
.unwrap();
}
let repository = open_repository(dir.path());
let mut options = backup::BackupOptions::new(&root, dir.path().join("state"));
options.checkpoint_entries = 40;
let (events, receiver) = crossbeam_channel::unbounded();
let watching = repository.clone();
let watched_root = root.clone();
let watcher = std::thread::spawn(move || {
let mut seen: Vec<(repository::SnapshotId, usize)> = Vec::new();
let mut verified: std::collections::HashSet<PathBuf> = std::collections::HashSet::new();
loop {
// Drain what the pipeline has said; dropping its end of the
// channel is how it announces the run is over
let finished = loop {
match receiver.try_recv() {
Ok(_) => continue,
Err(crossbeam_channel::TryRecvError::Empty) => break false,
Err(crossbeam_channel::TryRecvError::Disconnected) => break true,
}
};
for id in watching.snapshots().unwrap_or_default() {
if seen.iter().any(|(known, _)| *known == id) {
continue;
}
// It may be retired between being listed and being read
let Ok(snapshot) = watching.load_snapshot(id) else {
continue;
};
if snapshot.coverage != repository::Coverage::Partial {
continue;
}
let mut files = 0;
for entry in watching.manifest_entries(snapshot.manifest) {
let entry = entry.unwrap();
if let EntryKind::File { .. } = entry.kind {
files += 1;
// Each partial covers everything the ones before
// it did, so reading an entry back once is
// enough to know it was published whole
if verified.insert(entry.path.clone()) {
assert_eq!(
read_back(&watching, &entry),
std::fs::read(watched_root.join(&entry.path)).unwrap(),
"{:?} was published before its content was readable",
entry.path
);
}
}
}
seen.push((id, files));
}
if finished {
break;
}
std::thread::sleep(std::time::Duration::from_millis(5));
}
seen
});
let outcome = backup::run(
repository.clone(),
options,
events,
Arc::new(AtomicBool::new(false)),
)
.unwrap();
let Outcome::Completed(snapshot) = outcome else {
panic!("expected completion");
};
let seen = watcher.join().unwrap();
assert!(
seen.len() >= 2,
"a run this long should have published several times: {seen:?}"
);
assert!(
seen.iter().all(|(_, files)| *files > 0),
"an empty partial is not worth publishing: {seen:?}"
);
assert!(
seen.windows(2).all(|pair| pair[0].1 < pair[1].1),
"each partial should cover more than the one before it: {seen:?}"
);
// The finished snapshot stands alone, and has everything
assert_eq!(repository.snapshots().unwrap(), vec![snapshot]);
assert_eq!(manifest_map(&repository, snapshot).len(), 800);
}
/// Asked to finish early, a run publishes what it has and is done with:
/// no state left behind, and the same command afterwards is a new
/// backup rather than a resume.
#[test]
fn finishing_early_publishes_what_was_stored_and_ends_the_run() {
use repository::Coverage;
let dir = tempfile::tempdir().unwrap();
let root = dir.path().join("tree");
let state = dir.path().join("state");
for outer in 0..8 {
let sub = root.join(format!("dir-{outer}"));
std::fs::create_dir_all(&sub).unwrap();
for file in 0..25 {
let bytes = deterministic_bytes(4000, (outer * 40 + file) as u64);
std::fs::write(sub.join(format!("file-{file}.dat")), &bytes).unwrap();
}
}
let repository = open_repository(dir.path());
let run = |suspend: bool, early_finish: bool| {
let mut options = backup::BackupOptions::new(&root, &state);
options.early_finish = early_finish;
backup::run(
repository.clone(),
options,
crossbeam_channel::unbounded().0,
Arc::new(AtomicBool::new(suspend)),
)
.unwrap()
};
// A first session gets partway through and stands down, leaving a
// partial snapshot behind it
let Outcome::Suspended { restorable } = run(true, false) else {
panic!("expected a suspension");
};
let stopgap = restorable.expect("a suspension publishes what it has");
assert!(state.exists(), "a suspended run keeps its state");
let Outcome::FinishedEarly { snapshot, reason } = run(false, true) else {
panic!("expected an early finish");
};
assert_eq!(reason, EarlyFinish::Requested);
let snapshot = snapshot.expect("the first session had stored something");
// What the first session recorded is published as a finished
// snapshot: the run is over, so this is its last word rather than a
// stopgap waiting to be superseded
let published = repository.load_snapshot(snapshot).unwrap();
assert_eq!(published.coverage, Coverage::Complete);
assert!(published.entries > 0);
// Nothing new was recorded between the two, so this covers exactly
// the manifest the suspension published — a different object all
// the same, because it makes a different claim
assert_ne!(snapshot, stopgap);
assert_eq!(
repository.snapshots().unwrap(),
vec![snapshot],
"the early finish supersedes the partial the suspension left"
);
// The run is over: no state, no lock, and everything it published is
// really in the repository
assert!(!state.exists(), "an early finish clears its working state");
assert!(
repository.locks().unwrap().is_empty(),
"an early finish owes the repository nothing"
);
let covered = manifest_map(&repository, snapshot);
assert!(!covered.is_empty());
for entry in covered.values() {
if matches!(entry.kind, EntryKind::File { .. }) {
assert_eq!(
read_back(&repository, entry),
std::fs::read(root.join(&entry.path)).unwrap()
);
}
}
// Running again is a fresh backup, which finishes the whole tree
let Outcome::Completed(complete) = run(false, false) else {
panic!("expected completion");
};
assert_eq!(manifest_map(&repository, complete).len(), 8 + 200);
}
/// With nothing recorded to publish, an early finish publishes nothing:
/// a snapshot covering none of the tree says nothing worth keeping.
#[test]
fn an_early_finish_with_nothing_stored_publishes_nothing() {
let dir = tempfile::tempdir().unwrap();
let root = dir.path().join("tree");
std::fs::create_dir_all(&root).unwrap();
std::fs::write(root.join("file.dat"), b"never reached").unwrap();
let repository = open_repository(dir.path());
let state = dir.path().join("state");
let mut options = backup::BackupOptions::new(&root, &state);
options.early_finish = true;
let outcome = backup::run(
repository.clone(),
options,
crossbeam_channel::unbounded().0,
Arc::new(AtomicBool::new(false)),
)
.unwrap();
assert!(matches!(
outcome,
Outcome::FinishedEarly {
snapshot: None,
reason: EarlyFinish::Requested
}
));
assert!(repository.snapshots().unwrap().is_empty());
assert!(repository.locks().unwrap().is_empty());
assert!(!state.exists());
}
/// A store that runs out of room stops the backup rather than filling
/// up, and what the run had stored stays restorable — and the way out
/// of a full store works: narrow the backup to what fits, run it again
/// without the floor, and prune what the first run left.
#[test]
fn a_full_store_stops_the_backup_and_a_narrower_one_finishes() {
use repository::Coverage;
let dir = tempfile::tempdir().unwrap();
let root = dir.path().join("tree");
let state = dir.path().join("state");
std::fs::create_dir_all(&root).unwrap();
// Incompressible, undeduplicable content: what reaches the store is
// what came off the disk
const FILES: u64 = 32;
const FILE_BYTES: u64 = 1 << 20;
for file in 0..FILES {
let bytes = deterministic_bytes(FILE_BYTES as usize, file);
std::fs::write(root.join(format!("file-{file:02}.dat")), &bytes).unwrap();
}
// Room for about half the tree above the floor, which is deep enough
// past the first look for the run to have stored something and far
// enough from the end that it cannot have stored everything
const FLOOR: u64 = 16 << 20;
let repository = open_small_repository(dir.path(), FLOOR + (FILES / 2) * FILE_BYTES);
let mut options = backup::BackupOptions::new(&root, &state);
options.minimum_free_space = Some(FLOOR);
let outcome = backup::run(
repository.clone(),
options,
crossbeam_channel::unbounded().0,
Arc::new(AtomicBool::new(false)),
)
.unwrap();
let Outcome::FinishedEarly {
snapshot: Some(snapshot),
reason: EarlyFinish::StoreNearlyFull { free, floor },
} = outcome
else {
panic!("expected the store to stop the run: {outcome:?}");
};
assert_eq!(floor, FLOOR);
assert!(free < FLOOR, "the floor was reported uncrossed");
// The floor held: the run stopped near it rather than filling the
// store to the last byte. How far past it a run can get is one
// look's allowance plus the uploads in flight when the answer came.
assert!(
free > 0,
"the store was filled up despite a {} MiB floor",
FLOOR >> 20
);
// It stopped short, and what it stored is a finished backup of that
// much rather than a stopgap
let covered = manifest_map(&repository, snapshot);
assert_eq!(
repository.load_snapshot(snapshot).unwrap().coverage,
Coverage::Complete
);
assert!(!covered.is_empty(), "it stored nothing at all");
assert!(
(covered.len() as u64) < FILES,
"it stored the whole tree, so the floor never stopped it"
);
// Everything it did publish is really there, byte for byte
for entry in covered.values() {
assert_eq!(
read_back(&repository, entry),
std::fs::read(root.join(&entry.path)).unwrap(),
"{:?}",
entry.path
);
}
assert!(!state.exists(), "an early finish clears its working state");
assert!(repository.locks().unwrap().is_empty());
// Now the way out: exclude what did not fit and run again with the
// floor lifted, since the store is still below it.
let mut excludes: Vec<String> = Vec::new();
for file in 0..FILES {
let name = format!("file-{file:02}.dat");
if !covered.contains_key(Path::new(&name)) {
excludes.push(name);
}
}
assert!(!excludes.is_empty(), "nothing was left over to exclude");
let mut globs = globset::GlobSetBuilder::new();
for pattern in &excludes {
globs.add(globset::Glob::new(pattern).unwrap());
}
let mut options = backup::BackupOptions::new(&root, &state);
options.exclude = globs.build().unwrap();
let before = repository.backend().free_space().unwrap().unwrap();
let Outcome::Completed(complete) = backup::run(
repository.clone(),
options,
crossbeam_channel::unbounded().0,
Arc::new(AtomicBool::new(false)),
)
.unwrap() else {
panic!("the narrowed backup should have finished");
};
// It is a complete backup of what was kept, and it cost almost
// nothing to make: every chunk it names was already stored
assert_eq!(
repository.load_snapshot(complete).unwrap().coverage,
Coverage::Complete
);
let finished = manifest_map(&repository, complete);
assert_eq!(finished.len(), covered.len());
let after = repository.backend().free_space().unwrap().unwrap();
assert!(
before.saturating_sub(after) < FILE_BYTES,
"the second run stored {} bytes it did not need to",
before - after
);
// And the first run's snapshot can go, leaving a repository holding
// one honest snapshot that still restores
beeping::prune::prune(&repository, &[snapshot], false, false, |_| {}).unwrap();
assert_eq!(repository.snapshots().unwrap(), vec![complete]);
for entry in manifest_map(&repository, complete).values() {
if matches!(entry.kind, EntryKind::File { .. }) {
assert_eq!(
read_back(&repository, entry),
std::fs::read(root.join(&entry.path)).unwrap(),
"{:?} did not survive the prune",
entry.path
);
}
}
}
/// The limit for a store that cannot say what room it has left but can
/// say what it is holding — an FTP account with a quota. The
/// repository's own size is measured once and followed from there by
/// what the run writes, so no amount of watching costs a round trip.
#[test]
fn a_backup_stops_when_the_repository_reaches_its_maximum_size() {
use repository::Coverage;
let dir = tempfile::tempdir().unwrap();
let root = dir.path().join("tree");
let state = dir.path().join("state");
std::fs::create_dir_all(&root).unwrap();
const FILES: u64 = 24;
const FILE_BYTES: u64 = 1 << 20;
const LIMIT: u64 = 8 << 20;
for file in 0..FILES {
let bytes = deterministic_bytes(FILE_BYTES as usize, file);
std::fs::write(root.join(format!("file-{file:02}.dat")), &bytes).unwrap();
}
// Room enough for the whole tree; it is the limit that has to stop
// the run, not the store running out
let repository = open_store(dir.path(), u64::MAX, StoreAnswers::OnlyItsSize);
let mut options = backup::BackupOptions::new(&root, &state);
options.maximum_store_size = Some(LIMIT);
let outcome = backup::run(
repository.clone(),
options,
crossbeam_channel::unbounded().0,
Arc::new(AtomicBool::new(false)),
)
.unwrap();
let Outcome::FinishedEarly {
snapshot: Some(snapshot),
reason: EarlyFinish::StoreAtMaximum { used, limit },
} = outcome
else {
panic!("expected the size limit to stop the run: {outcome:?}");
};
assert_eq!(limit, LIMIT);
assert!(used >= LIMIT, "it stopped without reaching the limit");
let covered = manifest_map(&repository, snapshot);
assert_eq!(
repository.load_snapshot(snapshot).unwrap().coverage,
Coverage::Complete
);
assert!(!covered.is_empty(), "it stored nothing at all");
assert!(
(covered.len() as u64) < FILES,
"it stored the whole tree, so the limit never stopped it"
);
for entry in covered.values() {
assert_eq!(
read_back(&repository, entry),
std::fs::read(root.join(&entry.path)).unwrap(),
"{:?}",
entry.path
);
}
// The repository really is about that size, rather than the run
// having merely believed so
let held = repository.backend().used_space().unwrap().unwrap();
assert!(
held >= LIMIT && held < LIMIT + 4 * FILE_BYTES,
"the repository holds {held} against a {LIMIT} limit"
);
assert!(!state.exists());
assert!(repository.locks().unwrap().is_empty());
}
/// A store that will answer neither question is held to neither limit,
/// and says so rather than leaving one looking like a guarantee.
#[test]
fn a_store_that_says_nothing_about_itself_is_held_to_no_limits() {
use beeping::pipeline::{Event, StoreLimit};
let dir = tempfile::tempdir().unwrap();
let root = dir.path().join("tree");
std::fs::create_dir_all(&root).unwrap();
for file in 0..8 {
std::fs::write(root.join(format!("file-{file}.dat")), b"content").unwrap();
}
let repository = open_store(dir.path(), 1 << 20, StoreAnswers::Nothing);
let mut options = backup::BackupOptions::new(&root, dir.path().join("state"));
options.minimum_free_space = Some(64 << 30);
options.maximum_store_size = Some(1);
let (events, receiver) = crossbeam_channel::unbounded();
let outcome = backup::run(
repository.clone(),
options,
events,
Arc::new(AtomicBool::new(false)),
)
.unwrap();
// Limits that cannot be applied do not stop the run...
let Outcome::Completed(snapshot) = outcome else {
panic!("expected completion: {outcome:?}");
};
assert_eq!(manifest_map(&repository, snapshot).len(), 8);
// ...and neither do they pass unmentioned
let unapplied: Vec<StoreLimit> = receiver
.into_iter()
.filter_map(|event| match event {
Event::LimitNotApplied(limit) => Some(limit),
_ => None,
})
.collect();
assert_eq!(
unapplied,
vec![StoreLimit::MinimumFreeSpace, StoreLimit::MaximumStoreSize]
);
}