//! 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]
    );
}