gunnar-server 3.0.0

//! ★ **twin-resync-space (2026-10-06): a full resync on a twin whose volume is the size of
//! main's data set.**
//!
//! The gap (cov-13, 21:15 UTC): a twin behind a retention hole takes `Mirror.SnapshotFetch` and
//! STAGES the snapshot beside its old copy (its kept base and its log) before the install
//! replaces it — about twice the data on the twin's volume. The emergency path ("main grew
//! alone while the twin was silent") grows the twin to main's set size and then resyncs it, so
//! a twin grown only that far filled up mid-resync, failed, and failed the same way on every
//! poll after: a copy that never came back, each attempt freezing main's writes for the length
//! of the stream.
//!
//! The volume here is capped IN-PROCESS ([`Capped`]): the bytes under the twin's data root count
//! against the cap and a write past it fails `ENOSPC`, as a full filesystem's does. The same
//! scenario on a REAL small filesystem is
//! [`resync_space_on_a_real_filesystem_the_size_of_mains_set`] (ignored unless a capped mount is
//! handed in).

use std::path::{Path, PathBuf};
use std::sync::atomic::AtomicU64;
use std::sync::Mutex;

use gunnar_grpc::ControlPlane as _;
use snapshot::Volume as _;

use super::*;

const MIB: u64 = 1 << 20;

/// The marker a twin writes before it gives its stale copy up for a resync.
const DROPPED_MARK: &str = "txlog-snapshot-dropped";

/// Bytes of every file under `root` (apparent sizes): what a volume holding only `root` uses.
fn bytes_under(root: &Path) -> u64 {
    let mut total = 0u64;
    let mut stack = vec![root.to_path_buf()];
    while let Some(d) = stack.pop() {
        let Ok(rd) = std::fs::read_dir(&d) else {
            continue;
        };
        for e in rd.flatten() {
            let Ok(ft) = e.file_type() else { continue };
            if ft.is_dir() {
                stack.push(e.path());
            } else if ft.is_file() {
                total += e.metadata().map(|m| m.len()).unwrap_or(0);
            }
        }
    }
    total
}

/// **A volume with a cap, in-process**: everything under `root` counts against `cap`; a staging
/// write that would pass it fails `ENOSPC`, and [`snapshot::Volume::space`] answers what
/// `statvfs` would. `watch` runs after every staging write — the tests look at the twin from
/// INSIDE its resync with it.
struct Capped {
    root: PathBuf,
    cap: AtomicU64,
    /// The most the volume held after a staging write.
    peak: AtomicU64,
    watch: Mutex<Option<Box<dyn FnMut() + Send>>>,
}

impl Capped {
    fn new(root: &Path, cap: u64) -> Arc<Self> {
        Arc::new(Self {
            root: root.to_path_buf(),
            cap: AtomicU64::new(cap),
            peak: AtomicU64::new(0),
            watch: Mutex::new(None),
        })
    }
}

impl snapshot::Volume for Capped {
    fn space(&self, _path: &Path) -> Option<(u64, u64)> {
        let cap = self.cap.load(Ordering::SeqCst);
        Some((cap, cap.saturating_sub(bytes_under(&self.root))))
    }

    fn write_at(&self, file: &std::fs::File, bytes: &[u8], off: u64) -> std::io::Result<()> {
        use std::os::unix::fs::FileExt as _;
        let grows = (off + bytes.len() as u64).saturating_sub(file.metadata()?.len());
        let used = bytes_under(&self.root);
        if used + grows > self.cap.load(Ordering::SeqCst) {
            return Err(std::io::Error::from_raw_os_error(
                rustix::io::Errno::NOSPC.raw_os_error(),
            ));
        }
        file.write_all_at(bytes, off)?;
        self.peak.fetch_max(used + grows, Ordering::SeqCst);
        if let Some(w) = self.watch.lock().unwrap().as_mut() {
            w();
        }
        Ok(())
    }
}

/// `n` incompressible one-blob packs pushed to `ns/app`, each on its own branch `<what>-<i>`.
fn push_noise(primary: &ServerStores, what: &str, n: u64, bytes: u64) {
    let app = primary
        .catalog()
        .get(&RepoRef::new("ns", "app").unwrap())
        .unwrap()
        .unwrap();
    let store = primary.repo_store(&app).unwrap();
    for i in 0..n {
        let seed = what.bytes().fold(0x5EED_u64, |h, b| {
            h.wrapping_mul(131).wrapping_add(u64::from(b))
        });
        let (p, o) = one_blob_pack(
            store.hash_kind(),
            &noise(seed.wrapping_add(i), bytes as usize),
        );
        store.serve().put_pack(&p).unwrap();
        store
            .serve()
            .update_ref(&format!("refs/heads/{what}-{i}"), None, Some(&o))
            .unwrap();
    }
}

/// **A twin that was away**: it joined behind a hole (a snapshot: its kept BASE holds three
/// packs), followed three more (its LOG holds them), and went silent while main took four more
/// — and main's log no longer reaches back to it. Its old copy is base + log; the resync it is
/// owed is a snapshot of all ten packs.
struct Away {
    pd: tempfile::TempDir,
    primary: Arc<ServerStores>,
    twin: Arc<ServerStores>,
    /// Main's head: the snapshot's `at`.
    head: u64,
    /// The twin's head before the resync.
    own: u64,
    /// The twin's old copy on its volume: its kept base and its log.
    stale: u64,
    /// The bytes a snapshot of main stages.
    snapshot: u64,
}

/// **`f` once every pack main was just told is indexed.** [`wait_until_indexed`] reads the
/// engine's progress rows, and a pack pushed a moment ago may not have its row yet — then a
/// snapshot (which needs reachability over the whole graph) answers "index in progress". Retried
/// on exactly that answer, bounded, saying what it waited for.
fn settled<T>(primary: &ServerStores, what: &str, mut f: impl FnMut() -> anyhow::Result<T>) -> T {
    let deadline = std::time::Instant::now() + std::time::Duration::from_secs(60);
    loop {
        wait_until_indexed(primary, what);
        match f() {
            Ok(v) => return v,
            Err(e)
                if format!("{e:#}").contains("index in progress")
                    && std::time::Instant::now() < deadline =>
            {
                std::thread::sleep(std::time::Duration::from_millis(50));
            }
            Err(e) => panic!("{what}: {e:#}"),
        }
    }
}

impl Away {
    fn new(twin_dir: &Path, pack: u64) -> Self {
        let pd = tempfile::tempdir().unwrap();
        let (primary, _facts) = primary(pd.path());
        push_noise(&primary, "base", 3, pack);
        let twin = open_twin(twin_dir);
        let joined_at = primary.txlog().head_seq();
        let join = Follower::new(
            Arc::clone(&twin),
            Box::new(LocalSource {
                stores: Arc::clone(&primary),
                retained_from: joined_at,
            }),
        );
        assert_eq!(
            settled(&primary, "the twin joins behind a hole", || join.step()),
            Progress::SnapshotInstalled { at: joined_at },
            "the twin joined through a snapshot: it keeps a base"
        );
        push_noise(&primary, "followed", 3, pack);
        let follow = Follower::new(
            Arc::clone(&twin),
            Box::new(LocalSource {
                stores: Arc::clone(&primary),
                retained_from: 0,
            }),
        );
        while !matches!(follow.step().unwrap(), Progress::CaughtUp { .. }) {}
        let own = twin.txlog().head_seq();
        assert_eq!(own, primary.txlog().head_seq());
        push_noise(&primary, "while-silent", 4, pack);
        let head = primary.txlog().head_seq();
        let snapshot = settled(&primary, "measuring main's snapshot", || {
            let mut bytes = 0u64;
            snapshot::produce(&primary, &mut |p| {
                bytes += p.blob.len() as u64;
                Ok(())
            })?;
            Ok(bytes)
        });
        let stale = bytes_under(&twin_dir.join(snapshot::BASE)) + twin.txlog().bytes();
        Away {
            pd,
            primary,
            twin,
            head,
            own,
            stale,
            snapshot,
        }
    }

    /// The source the twin comes back to: main, retaining only from its head.
    fn hole(&self) -> LocalSource {
        LocalSource {
            stores: Arc::clone(&self.primary),
            retained_from: self.head,
        }
    }

    /// **Main's data set**: what main's data takes on its set with its log trimmed away (the
    /// byte budget's job), plus 2 MiB. A twin grown to main's set is this big.
    fn mains_set(&self) -> u64 {
        bytes_under(self.pd.path()) - self.primary.txlog().bytes() + 2 * MIB
    }
}

/// What the twin looked like from INSIDE its resync, the first time its stale copy was gone.
#[derive(Debug)]
struct Window {
    state: String,
    why: String,
    serve: Result<u64, String>,
    alarms: Vec<String>,
}

fn look(twin: &Arc<ServerStores>) -> Window {
    let st = status(twin.root()).expect("a status");
    let serve = snapshot::serve(twin, &mut |_| Ok(())).map_err(|e| format!("{e:#}"));
    Window {
        state: st.state.to_string(),
        why: st.why,
        serve,
        alarms: alarms_of(twin),
    }
}

fn alarms_of(stores: &Arc<ServerStores>) -> Vec<String> {
    crate::CoreControlPlane::new(Arc::clone(stores))
        .health_alarms()
        .into_iter()
        .map(|a| format!("{}: {}", a.event, a.detail))
        .collect()
}

fn said<T: std::fmt::Debug>(r: &anyhow::Result<T>) -> String {
    match r {
        Ok(v) => format!("Ok({v:?})"),
        Err(e) => format!("Err({e:#})"),
    }
}

/// A recharge FROM the twin is refused while it holds no whole copy.
fn recharge_refused(twin: &Arc<ServerStores>) -> String {
    let fd = tempfile::tempdir().unwrap();
    let fresh = Arc::new(ServerStores::open(fd.path()).unwrap());
    let r = restore_from(
        Box::new(LocalSource {
            stores: Arc::clone(twin),
            retained_from: 0,
        }),
        fresh,
        "",
    );
    format!(
        "{:#}",
        r.expect_err("a recharge from a twin that holds no whole copy must be refused")
    )
}

/// After the resync: the twin holds main, a main recharged from it IS main, and it follows on.
fn resynced_and_following(a: &Away, twin: &Arc<ServerStores>, f: &Follower, who: &str) {
    assert_eq!(
        (twin.txlog().first_seq(), twin.txlog().head_seq()),
        (a.head + 1, a.head),
        "{who}: the twin's log resumes after the snapshot"
    );
    assert_eq!(
        snapshot::base_at(twin.root()).unwrap(),
        Some(a.head),
        "{who}: the twin keeps the new snapshot as its base"
    );
    for gone in [DROPPED_MARK, snapshot::STAGING, snapshot::INSTALLING] {
        assert!(
            !twin.root().join(gone).exists(),
            "{who}: {gone} is left on the twin after the resync"
        );
    }
    assert!(
        !alarms_of(twin).iter().any(|a| a.starts_with("twin.resync")),
        "{who}: the alarm is cleared: {:?}",
        alarms_of(twin)
    );
    {
        let (_d, main) = recharged(twin);
        same_state(&a.primary, &main, who);
        for i in 0..4 {
            assert!(
                refs_of(&main)
                    .iter()
                    .any(|(n, _, _)| n == &format!("refs/heads/while-silent-{i}")),
                "{who}: the push made while the twin was away is on the recharged main: {i}"
            );
        }
    }
    push_noise(&a.primary, "after-resync", 1, 64 << 10);
    assert!(
        matches!(f.step().unwrap(), Progress::Applied { count, .. } if count >= 1),
        "{who}: the twin follows on after the resync"
    );
    let st = status(twin.root()).unwrap();
    assert!(st.caught_up(), "{who}: caught up after the resync: {st:?}");
}

/// ★ **A twin whose volume is main's data set takes a full resync, and never fills up.**
///
/// RED before the fix: the snapshot was staged beside the twin's old copy, the volume filled
/// part-way (`No space left on device`), the attempt failed and the next one failed the same
/// way. GREEN: the twin stages beside its old copy while that fits, and when it does not it
/// GIVES THE STALE COPY UP — said at WARN, alarmed, written in its status — and the resync lands.
/// While it holds no copy it serves none: `Mirror.SnapshotFetch` on it (a recharge) is refused.
#[test]
fn resync_space_a_twin_the_size_of_mains_set_takes_a_full_resync_and_never_fills() {
    let td = tempfile::tempdir().unwrap();
    let a = Away::new(td.path(), 2 * MIB);
    let cap = a.mains_set();
    let rest = bytes_under(td.path()) - a.stale;
    assert!(rest + a.stale + a.snapshot > cap, "the geometry: staged BESIDE its old copy ({} + {} + {}) the resync must not fit the twin's volume ({cap})", rest, a.stale, a.snapshot);
    assert!(
        rest + a.snapshot + 2 * MIB <= cap,
        "the geometry: WITHOUT its old copy the resync fits ({} + {} + 2 MiB <= {cap})",
        rest,
        a.snapshot
    );

    let vol = Capped::new(td.path(), cap);
    let seen: Arc<Mutex<Option<Window>>> = Arc::new(Mutex::new(None));
    *vol.watch.lock().unwrap() = Some(Box::new({
        let (twin, seen, root) = (
            Arc::clone(&a.twin),
            Arc::clone(&seen),
            td.path().to_path_buf(),
        );
        move || {
            let mut s = seen.lock().unwrap();
            if s.is_none() && root.join(DROPPED_MARK).exists() {
                *s = Some(look(&twin));
            }
        }
    }));
    let f = Follower::new(Arc::clone(&a.twin), Box::new(a.hole())).with_volume(vol.clone());
    let got = f.step();
    assert!(
        matches!(got, Ok(Progress::SnapshotInstalled { at }) if at == a.head),
        "the resync landed on a volume the size of main's set: {}",
        said(&got)
    );
    assert!(vol.peak.load(Ordering::SeqCst) <= cap);

    let w = seen
        .lock()
        .unwrap()
        .take()
        .expect("the old copy was given up mid-resync: the volume holds one copy, not two");
    assert_eq!(w.state, "resyncing", "{w:?}");
    assert!(
        w.why.contains("NO copy"),
        "the status says the twin holds no copy meanwhile: {w:?}"
    );
    assert!(
        w.serve.as_ref().is_err_and(|e| e.contains("no whole copy")),
        "a twin without its copy serves none: {w:?}"
    );
    assert!(
        w.alarms
            .iter()
            .any(|a| a.starts_with("twin.resync_without_copy")),
        "the twin raises an alarm: {w:?}"
    );

    resynced_and_following(
        &a,
        &a.twin,
        &f,
        "after a resync on a volume the size of main's set",
    );
    row(
        "twin-resync-space: a twin whose volume is main's data set takes a full resync — its stale copy given up, alarmed — and a main recharged from it is main",
        true,
        &format!("cap {cap}, old copy {}, snapshot {}, peak {}", a.stale, a.snapshot, vol.peak.load(Ordering::SeqCst)),
    );
}

/// **The link dies after the twin gave its old copy up; the twin's process dies too.**
///
/// What is on its disk then holds NO copy: the twin must not serve it (a recharge is refused
/// by name), must not call itself caught up, must keep saying so across the restart — and the
/// first poll after it takes the resync again, cleanly, and follows on.
#[test]
fn resync_space_a_crash_mid_resync_restarts_it_and_the_twin_is_whole_and_caught_up() {
    struct CutAfterTheDrop {
        inner: LocalSource,
        twin_root: PathBuf,
        after: AtomicU64,
    }
    impl LogSource for CutAfterTheDrop {
        fn head(&self) -> anyhow::Result<(u64, u64)> {
            self.inner.head()
        }
        fn fetch(&self, from: u64, to: u64, max_bytes: u64) -> anyhow::Result<Vec<Entry>> {
            self.inner.fetch(from, to, max_bytes)
        }
        fn snapshot(&self, into: &mut snapshot::Stager) -> anyhow::Result<bool> {
            snapshot::serve(&self.inner.stores, &mut |p| {
                if self.twin_root.join(DROPPED_MARK).exists()
                    && self.after.fetch_add(1, Ordering::SeqCst) >= 2
                {
                    anyhow::bail!("the link to the primary died mid-resync (test)");
                }
                into.feed(p)
            })?;
            Ok(true)
        }
    }
    let td = tempfile::tempdir().unwrap();
    let a = Away::new(td.path(), 2 * MIB);
    let cap = a.mains_set();
    let f = Follower::new(
        Arc::clone(&a.twin),
        Box::new(CutAfterTheDrop {
            inner: a.hole(),
            twin_root: td.path().to_path_buf(),
            after: AtomicU64::new(0),
        }),
    )
    .with_volume(Capped::new(td.path(), cap));
    let got = f.step();
    assert!(
        got.as_ref()
            .is_err_and(|e| format!("{e:#}").contains("died mid-resync")),
        "the link died after the old copy was given up: {}",
        said(&got)
    );

    // What the cut left: no base, an empty log at the twin's old head, the marker.
    assert!(td.path().join(DROPPED_MARK).exists());
    assert!(
        !td.path().join(snapshot::BASE).exists(),
        "the stale base was given up"
    );
    assert_eq!(
        (a.twin.txlog().first_seq(), a.twin.txlog().head_seq()),
        (a.own + 1, a.own),
        "the stale log was given up; the head stays the twin's own"
    );
    let st = status(td.path()).unwrap();
    assert!(
        !st.caught_up() && st.state != State::Following && st.why.contains("NO copy"),
        "{st:?}"
    );
    let refused = recharge_refused(&a.twin);
    assert!(refused.contains("no whole copy"), "{refused}");
    assert!(
        alarms_of(&a.twin)
            .iter()
            .any(|x| x.starts_with("twin.resync_without_copy")),
        "{:?}",
        alarms_of(&a.twin)
    );

    // The process dies: every handle on the twin's root goes, its in-memory status with it.
    drop(f);
    let Away {
        pd,
        primary,
        twin,
        head,
        own,
        stale,
        snapshot,
    } = a;
    drop(twin);
    status_slots().remove(td.path());
    let twin = Arc::new(ServerStores::open(td.path()).unwrap());
    assert!(twin.is_twin());
    let a = Away {
        pd,
        primary,
        twin,
        head,
        own,
        stale,
        snapshot,
    };
    let refused = recharge_refused(&a.twin);
    assert!(
        refused.contains("no whole copy"),
        "after the restart, before any poll: {refused}"
    );
    assert!(
        alarms_of(&a.twin)
            .iter()
            .any(|x| x.starts_with("twin.resync_without_copy")),
        "the alarm survives the restart"
    );

    let f = Follower::new(Arc::clone(&a.twin), Box::new(a.hole()))
        .with_volume(Capped::new(td.path(), cap));
    let got = f.step();
    assert!(
        matches!(got, Ok(Progress::SnapshotInstalled { at }) if at == a.head),
        "the first poll after the restart takes the resync again: {}",
        said(&got)
    );
    resynced_and_following(&a, &a.twin, &f, "after a crash mid-resync and a restart");
    row("twin-resync-space: a crash mid-resync (old copy already given up) is refused as a source, alarmed across the restart, and the next poll resyncs whole", true, &format!("cap {cap}, snapshot at {}", a.head));
}

/// **A crash between the marker and the end of the give-up** leaves the marker beside a base
/// and a log that are still there. The marker wins: nothing is served from that copy, and the
/// resync finishes the give-up before it stages.
#[test]
fn resync_space_a_give_up_cut_short_by_a_crash_is_finished_before_the_resync() {
    let td = tempfile::tempdir().unwrap();
    let a = Away::new(td.path(), 2 * MIB);
    let cap = a.mains_set();
    std::fs::write(
        td.path().join(DROPPED_MARK),
        format!("dropped_unix_ms 1\nhead {}\n", a.own),
    )
    .unwrap();
    assert!(
        td.path().join(snapshot::BASE).exists() && a.twin.txlog().bytes() > 0,
        "the old copy is still on the disk"
    );
    let refused = recharge_refused(&a.twin);
    assert!(refused.contains("no whole copy"), "{refused}");
    let f = Follower::new(Arc::clone(&a.twin), Box::new(a.hole()))
        .with_volume(Capped::new(td.path(), cap));
    let got = f.step();
    assert!(
        matches!(got, Ok(Progress::SnapshotInstalled { at }) if at == a.head),
        "{}",
        said(&got)
    );
    resynced_and_following(&a, &a.twin, &f, "after a give-up cut short by a crash");
}

/// **A volume that cannot hold main's data even alone**: the twin gives its stale copy up,
/// still runs short, and says `[twin-too-small]` with the numbers — `recovering`, NO copy, an
/// alarm. It does NOT ask again while its volume is the same size (every `SnapshotFetch` holds
/// main's writes for the stream), and it resyncs the moment the volume is grown.
#[test]
fn resync_space_a_volume_too_small_even_alone_waits_in_recovering_until_grown_and_never_hammers_main(
) {
    struct Counting {
        inner: LocalSource,
        asked: AtomicU64,
    }
    impl LogSource for Counting {
        fn head(&self) -> anyhow::Result<(u64, u64)> {
            self.inner.head()
        }
        fn fetch(&self, from: u64, to: u64, max_bytes: u64) -> anyhow::Result<Vec<Entry>> {
            self.inner.fetch(from, to, max_bytes)
        }
        fn snapshot(&self, into: &mut snapshot::Stager) -> anyhow::Result<bool> {
            self.asked.fetch_add(1, Ordering::SeqCst);
            self.inner.snapshot(into)
        }
    }
    let td = tempfile::tempdir().unwrap();
    let a = Away::new(td.path(), 2 * MIB);
    let rest = bytes_under(td.path()) - a.stale;
    let cap = rest + a.snapshot - 2 * MIB;
    assert!(
        rest + a.stale <= cap,
        "the geometry: the twin's old copy fits ({} + {} <= {cap})",
        rest,
        a.stale
    );
    let vol = Capped::new(td.path(), cap);
    let source = Arc::new(Counting {
        inner: a.hole(),
        asked: AtomicU64::new(0),
    });
    struct Shared(Arc<Counting>);
    impl LogSource for Shared {
        fn head(&self) -> anyhow::Result<(u64, u64)> {
            self.0.head()
        }
        fn fetch(&self, from: u64, to: u64, max_bytes: u64) -> anyhow::Result<Vec<Entry>> {
            self.0.fetch(from, to, max_bytes)
        }
        fn snapshot(&self, into: &mut snapshot::Stager) -> anyhow::Result<bool> {
            self.0.snapshot(into)
        }
    }
    let f = Follower::new(Arc::clone(&a.twin), Box::new(Shared(Arc::clone(&source))))
        .with_volume(vol.clone());
    let got = f.step();
    let err = format!(
        "{:#}",
        got.as_ref()
            .map(drop)
            .expect_err("a volume smaller than main's data cannot take the resync")
    );
    assert!(
        err.contains(gunnar_grpc::TWIN_TOO_SMALL),
        "named as too small, not a bare ENOSPC: {err}"
    );
    let st = status(td.path()).unwrap();
    assert_eq!(st.state, State::Recovering, "{st:?}");
    assert!(
        st.why.contains("NO copy") && st.why.contains(gunnar_grpc::TWIN_TOO_SMALL),
        "{st:?}"
    );
    assert!(
        alarms_of(&a.twin)
            .iter()
            .any(|x| x.starts_with("twin.resync_without_copy")),
        "{:?}",
        alarms_of(&a.twin)
    );
    assert_eq!(source.asked.load(Ordering::SeqCst), 1);
    assert!(
        !td.path().join(snapshot::STAGING).exists(),
        "the half staged is not left taking the room"
    );

    // The same volume: it waits, and does not ask main for another snapshot it cannot hold.
    let again = f.step();
    assert!(
        again
            .as_ref()
            .is_err_and(|e| format!("{e:#}").contains(gunnar_grpc::TWIN_TOO_SMALL)),
        "{}",
        said(&again)
    );
    assert_eq!(source.asked.load(Ordering::SeqCst), 1, "a twin too small for main's data does not freeze main's writes with a resync it cannot hold");

    // Grown (monetize's twin lane, or by hand): the next poll resyncs.
    vol.cap.store(cap * 2, Ordering::SeqCst);
    let got = f.step();
    assert!(
        matches!(got, Ok(Progress::SnapshotInstalled { at }) if at == a.head),
        "{}",
        said(&got)
    );
    assert_eq!(source.asked.load(Ordering::SeqCst), 2);
    assert!(!td.path().join("txlog-snapshot-too-small").exists());
    resynced_and_following(&a, &a.twin, &f, "after the volume was grown");
}

/// **A twin mid-install serves no snapshot** (the latent half of the same rule): the install
/// marker means its control plane may already be the new snapshot's while its base is still the
/// old one. `serve` used to stream that base regardless.
#[test]
fn resync_space_a_twin_mid_install_serves_no_snapshot_and_no_recharge() {
    let td = tempfile::tempdir().unwrap();
    let a = Away::new(td.path(), 256 << 10);
    assert!(snapshot::base_at(td.path()).unwrap().is_some());
    crate::durable::write_atomic_unframed(
        td.path(),
        snapshot::INSTALLING,
        format!("{}\n", a.head).as_bytes(),
    )
    .unwrap();
    let served = snapshot::serve(&a.twin, &mut |_| Ok(()));
    assert!(
        served
            .as_ref()
            .is_err_and(|e| format!("{e:#}").contains("no whole copy")),
        "{}",
        said(&served)
    );
    let refused = recharge_refused(&a.twin);
    assert!(refused.contains("no whole copy"), "{refused}");
}

/// ★ **The same resync on a REAL filesystem the size of main's set.** Ignored in the suite: it
/// needs `GUNNAR_TWIN_CAPPED_DIR`, an empty directory on a small filesystem of its own (e.g. an
/// ext2 image mounted with fuse2fs, or a tmpfs with `size=`). The packs are sized from the
/// filesystem: the twin's old copy is half of it and main's data ~5/6 of it, so staged beside
/// the old copy the resync cannot fit and without it, it does.
#[test]
#[ignore = "needs GUNNAR_TWIN_CAPPED_DIR: an empty directory on a small filesystem of its own"]
fn resync_space_on_a_real_filesystem_the_size_of_mains_set() {
    let dir =
        PathBuf::from(std::env::var_os("GUNNAR_TWIN_CAPPED_DIR").expect("GUNNAR_TWIN_CAPPED_DIR"));
    let root = dir.join(format!("twin-{}", std::process::id()));
    std::fs::create_dir_all(&root).unwrap();
    let (total, free) = snapshot::Disk
        .space(&root)
        .expect("a measurable filesystem");
    let pack = total / 12;
    let a = Away::new(&root, pack);
    let (_, free_before) = snapshot::Disk.space(&root).unwrap();
    assert!(a.snapshot > free_before, "the geometry: staged beside its old copy the resync cannot fit ({} > {free_before} free of {total}; {free} at the start)", a.snapshot);
    let f = Follower::new(Arc::clone(&a.twin), Box::new(a.hole()));
    let got = f.step();
    assert!(
        matches!(got, Ok(Progress::SnapshotInstalled { at }) if at == a.head),
        "the resync landed on a real {total}-byte filesystem: {}",
        said(&got)
    );
    resynced_and_following(
        &a,
        &a.twin,
        &f,
        "on a real filesystem the size of main's set",
    );
    row(
        "twin-resync-space: the resync on a REAL filesystem the size of main's set",
        true,
        &format!("filesystem {total}, packs {pack}, snapshot {}", a.snapshot),
    );
    drop(f);
    drop(a);
    let _ = std::fs::remove_dir_all(&root);
}