gunnar-server 3.0.0

//! **The snapshot** — what a twin installs when the primary no longer retains the range it
//! needs (`.nornir/twin-transaction-log.md` §"Snapshot").
//!
//! A snapshot is the MUST-NOT-LOSE classes (M1–M9 of the note's inventory) at one sequence
//! `at`, spelled as the same entries the log carries — `Kek`, `ControlBlob`, `ControlDoc`,
//! `Secret`, `File`, and per repository `RepoStore{created}`, one `Pack`, one `Refs`, `LfsPut` —
//! each with `seq` 0, and a `SnapshotAt{at}` terminator. So the one `apply` installs it, and the
//! one wire piece (`LogChunk`) carries it.
//!
//! **Consistent at `at`.** The primary takes its CUT under [`gunnar_core::txlog::Log::freeze`]:
//! no framed mutation starts, every one in flight reaches its store, and only then is the head
//! read and the state fixed (a read view of the control database, the framed files' handles and
//! lengths, every repository's refs). The log is written BEFORE the store, so without the wait
//! an entry at the head could still be on its way into the store and the twin, resuming at
//! `at + 1`, would never see it. Writes pause for the cut, not for the stream (R5.1-A).
//!
//! **Reused, not rewritten.** The objects of a repository are the engine's own export: the
//! reachability `select` (or `reachable`) over every ref tip, then `emit_pack` — the znippy
//! arm answers that as a byte-range copy out of its archive, the same path a clone takes. The
//! control database is [`gunnar_core::control::ControlDb::export_each`], row by row.
//!
//! **Streamed, never held** (2026-09-15, Lane Q). A 44 GiB data set does not fit in a 1 GiB
//! twin, so neither end holds the snapshot:
//!
//! * the producer ([`produce`]) writes every item through ONE preallocated [`PIECE_BYTES`]
//!   buffer and hands each full slab to a sink as a [`Piece`] — `emit_pack` writes straight
//!   into it, files and LFS objects are copied through it, control rows go one by one. Over the
//!   wire the sink is a bounded channel into the gRPC stream (`Mirror.SnapshotFetch`), so a slow
//!   twin holds the producer back instead of the producer piling pieces up. An item's length is
//!   not known before a pack is emitted: every piece but an item's last says
//!   [`BLOB_LEN_AT_LAST`], the last carries the item's total;
//! * the twin ([`Stager`]) appends each piece's bytes to ONE staging file on its data disk as
//!   it arrives and keeps only the item table (kind, meta, offset, length). Nothing is touched
//!   until the terminator has arrived and the file is synced;
//! * the install ([`install`]) reads each item back from the staging file — a small one into a
//!   reused buffer, a large one as a read-only mapping (page cache the kernel can reclaim, the
//!   push spool's rule in `storage::PackBytes`) — and deletes the staging file at the end.
//!
//! **Installed whole.** The twin wipes its repository stores and every framed file the snapshot
//! does not carry, force-sets every document to the snapshot's pointer, applies the rest with
//! [`super::apply_raw`], and restarts its log after `at`. A marker makes a crash mid-install ask
//! for the snapshot again rather than follow from a half-installed state; a crash while the
//! snapshot is still arriving leaves the stores untouched and the staging file is started over.
//!
//! **A twin keeps it** (ruling 2026-09-22, "the twin only stores"): a twin applies only the
//! control-plane items and keeps the staging file as [`BASE`] — its copy of everything at or
//! below `at` — and [`serve`] streams that base when the twin is asked for a snapshot. The box
//! recharged from it installs the base whole, and that is where the indexes are built.
//!
//! **Room on a twin's volume** (twin-resync-space, 2026-10-06). The staging file needs the
//! snapshot's size free on the twin's data disk, and it is staged BESIDE the copy it replaces —
//! the kept base and the log after it — so a resync can need about twice main's data. A twin is
//! grown to main's data set, not to twice it (cov-13: "main grew alone while the twin was
//! silent", then the twin is grown to main's set and resynced), and such a twin filled up
//! part-way, failed, and failed the same way on every later poll. So a twin's [`Stager`] makes
//! room ([`Stager::making_room`]): it stages beside the old copy while that fits (the old copy is
//! kept until the new one is whole — the common case, a twin with room), and the moment the
//! volume would run below [`staging_floor`] (or says `ENOSPC`) it GIVES THE STALE COPY UP: the
//! [`DROPPED`] marker first (durable), then its base and its log. It is a replica that was
//! already behind a hole main had trimmed; main holds the truth. The window is explicit — a
//! WARN `ALERT` line, the `twin.resync_without_copy` health alarm, the state word `resyncing`
//! with "holds NO copy" — and a twin in it serves nothing ([`not_whole`]: no snapshot, no log,
//! so no recharge from it). A crash anywhere in it restarts the resync on the next poll (the
//! marker makes [`installing`] true). A volume too small for main's data even alone says
//! [`TooSmall`] (`[twin-too-small]`), waits in `recovering`, and asks main for no further
//! snapshot until its volume is larger ([`TOO_SMALL`]) — every `SnapshotFetch` costs main a
//! frozen cut (R5.1-A) and a whole stream.
//!
//! Why not in place, store by store: a store-only twin's copy is ONE base file plus the log after
//! it, not stores. Any mix of the old and the new is consistent at no sequence (the control plane
//! at one point, the data at another), so an in-place swap has exactly the same no-copy window as
//! giving the old one up — with a layout change and a resumable snapshot protocol on top. And
//! growing every twin to twice main's set (a preflight) costs that volume for ever, to keep a
//! copy that is already behind a hole.
//!
//! Limits, named: the freeze is held for the CUT, not the stream (R5.1-A, 2026-10-07: it used to
//! pause the primary's writes for as long as the twin took to receive everything) — the cut reads
//! every repository's refs, so it grows with the number of repositories; a gc and a repository
//! store's removal wait for a stream in progress (at most `stores::SNAPSHOT_PIN_WAIT`); the control
//! database's read view keeps its file open while it streams; a repository's object-id list from
//! `select` is held in memory (ids, not objects); the staging file needs the snapshot's size free
//! on the twin's data disk (beside what it replaces while that fits; see "Room on a twin's volume").

use std::collections::{BTreeMap, BTreeSet};
use std::io::Write;
use std::os::unix::fs::FileExt;
use std::path::{Path, PathBuf};
use std::sync::Mutex;

use anyhow::{anyhow, Context};
use gunnar_core::control::{BlobId, ExportRow};
use gunnar_core::txlog::{self, Crc32c, Kind, Meta, RefEdit};
use gunnar_store::git_storage_trait::Caps;

use crate::stores::ServerStores;

/// `<data-root>/txlog-snapshot-installing`: an install is in progress. Per box, never framed.
pub const INSTALLING: &str = "txlog-snapshot-installing";

/// `<data-root>/txlog-snapshot-staging/`: a snapshot arriving (or arrived and installing). Per
/// box, never framed; removed when the install ends or the next attempt starts.
pub const STAGING: &str = "txlog-snapshot-staging";

/// `<data-root>/txlog-snapshot-dropped`: **this twin gave its stale copy up** (its kept base and
/// its log) to make room for a full resync and holds NO copy until the resync installs
/// (twin-resync-space, 2026-10-06). Written BEFORE the first byte of the old copy goes; removed
/// by [`install`] once the new copy is whole. Per box, never framed.
pub const DROPPED: &str = "txlog-snapshot-dropped";

/// `<data-root>/txlog-snapshot-too-small`: the last resync ran out of room even with the stale
/// copy given up; it holds the volume's size then. The twin asks main for no further snapshot
/// until its volume is larger. Removed by [`install`]. Per box, never framed.
pub const TOO_SMALL: &str = "txlog-snapshot-too-small";

/// One piece's bytes at most: the producer's one buffer, and the wire's `LogChunk` slab (under
/// every client's default 4 MiB message ceiling with room for the meta).
pub const PIECE_BYTES: usize = 1 << 20;

/// `blob_len` on every piece of an item but its last: the length is known when the item ends.
pub const BLOB_LEN_AT_LAST: u64 = u64::MAX;

/// Items at or under this size are read back into the install's reused buffer; larger ones are
/// mapped.
const INLINE_BYTES: u64 = PIECE_BYTES as u64;

/// **One piece of a snapshot item**, borrowed: `meta` on the item's first piece only, `blob` a
/// slab of at most [`PIECE_BYTES`], `blob_len` the item's total on its `last` piece and
/// [`BLOB_LEN_AT_LAST`] before it.
#[derive(Debug, Clone, Copy)]
pub struct Piece<'a> {
    pub kind: u16,
    pub meta: &'a [u8],
    pub blob_len: u64,
    pub blob: &'a [u8],
    pub last: bool,
}

/// Where the producer's pieces go.
pub type Sink<'s> = dyn FnMut(Piece<'_>) -> anyhow::Result<()> + 's;

fn not_found(e: &std::io::Error) -> bool {
    e.kind() == std::io::ErrorKind::NotFound
}

/// **The producer's one buffer**, and the item being written through it. A full slab is handed
/// to the sink only when more bytes arrive, so an item's last piece is never an empty extra one.
struct Pieces<'s, 'k> {
    sink: &'k mut Sink<'s>,
    buf: Vec<u8>,
    kind: u16,
    meta: Vec<u8>,
    first: bool,
    total: u64,
    /// The sink's error, kept whole while `Write` can only say "other".
    failed: Option<anyhow::Error>,
    /// Pieces handed over, for the log line.
    pieces: u64,
}

impl<'s, 'k> Pieces<'s, 'k> {
    fn new(sink: &'k mut Sink<'s>) -> Self {
        Self {
            sink,
            buf: Vec::with_capacity(PIECE_BYTES),
            kind: 0,
            meta: Vec::new(),
            first: true,
            total: 0,
            failed: None,
            pieces: 0,
        }
    }

    fn begin(&mut self, kind: Kind, meta: &Meta) {
        debug_assert!(self.buf.is_empty() && self.first);
        self.kind = kind as u16;
        self.meta = meta.to_bytes();
        self.total = 0;
    }

    fn hand_over(&mut self, last: bool) -> anyhow::Result<()> {
        let piece = Piece {
            kind: self.kind,
            meta: if self.first { &self.meta } else { &[] },
            blob_len: if last { self.total } else { BLOB_LEN_AT_LAST },
            blob: &self.buf,
            last,
        };
        let r = (self.sink)(piece);
        self.buf.clear();
        self.first = false;
        self.pieces += 1;
        r
    }

    /// The item ends: its last piece, with its total.
    fn end(&mut self) -> anyhow::Result<()> {
        if let Some(err) = self.failed.take() {
            return Err(err);
        }
        self.hand_over(true)?;
        self.first = true;
        Ok(())
    }

    /// A whole small item.
    fn item(&mut self, kind: Kind, meta: &Meta, blob: &[u8]) -> anyhow::Result<()> {
        self.begin(kind, meta);
        self.write_all(blob)?;
        self.end()
    }

    /// A file's bytes as one item, copied through the buffer — at most `limit` of them (the
    /// length the file had at a snapshot's cut; `u64::MAX` = to its end).
    fn file(
        &mut self,
        kind: Kind,
        meta: &Meta,
        file: &std::fs::File,
        limit: u64,
    ) -> anyhow::Result<()> {
        self.file_at(kind, meta, file, 0, limit)
    }

    /// [`Self::file`] from byte `from` of the file: `limit` bytes of it (a frame's blob inside a
    /// log segment).
    fn file_at(
        &mut self,
        kind: Kind,
        meta: &Meta,
        file: &std::fs::File,
        from: u64,
        limit: u64,
    ) -> anyhow::Result<()> {
        self.begin(kind, meta);
        let mut at = 0u64;
        loop {
            if self.buf.len() == PIECE_BYTES {
                self.hand_over(false)?;
            }
            let start = self.buf.len();
            let room = ((PIECE_BYTES - start) as u64).min(limit - at) as usize;
            if room == 0 {
                break;
            }
            // SAFETY-free spare capacity use: grow the length over the spare room, read into it,
            // cut back to what arrived. The buffer never reallocates (capacity PIECE_BYTES).
            self.buf.resize(start + room, 0);
            let n = match file.read_at(&mut self.buf[start..], from + at) {
                Ok(n) => n,
                Err(e) => {
                    self.buf.truncate(start);
                    return Err(e.into());
                }
            };
            self.buf.truncate(start + n);
            if n == 0 {
                break;
            }
            at += n as u64;
            self.total += n as u64;
        }
        self.end()
    }
}

impl Write for Pieces<'_, '_> {
    fn write(&mut self, mut data: &[u8]) -> std::io::Result<usize> {
        let len = data.len();
        while !data.is_empty() {
            if self.buf.len() == PIECE_BYTES {
                if let Err(err) = self.hand_over(false) {
                    let say = format!("{err:#}");
                    self.failed = Some(err);
                    return Err(std::io::Error::other(say));
                }
            }
            let n = (PIECE_BYTES - self.buf.len()).min(data.len());
            self.buf.extend_from_slice(&data[..n]);
            self.total += n as u64;
            data = &data[n..];
        }
        Ok(len)
    }

    fn flush(&mut self) -> std::io::Result<()> {
        Ok(())
    }
}

/// **What a snapshot carries, fixed at one instant** (R5.1-A, 2026-10-07): the sequence `at`, the
/// KEK, a read view of the control database, every framed file as an open handle and its length,
/// and every repository's refs, HEAD and tips. Taken under the log's freeze — that is the whole
/// freeze now — and streamed from after it.
struct Cut {
    at: u64,
    kek: Vec<u8>,
    control: gunnar_core::ControlView,
    files: Vec<(&'static str, std::fs::File, u64)>,
    repos: Vec<RepoCut>,
}

/// One repository at the cut.
struct RepoCut {
    repo: gunnar_core::Repo,
    key: String,
    hash: String,
    edits: Vec<RefEdit>,
    tips: Vec<Vec<u8>>,
    /// **Its pushes caught between their two frames** ([`pending_packs`]): each `Pack` frame at or
    /// below the cut that no `Refs` frame of this repository follows yet, as an open handle on its
    /// segment, the blob's offset and length. Streamed beside the reachable export.
    pending: Vec<(std::fs::File, u64, u64)>,
}

/// Pure: **the `Pack` frames a cut at `at` must carry beside the reachable export** — per
/// repository, the ones after its last `Refs` frame. A push is two frames: its `Pack` (framed in
/// the ingest, under its own mutation guard) and, after the policy gates, its `Refs`. A cut that
/// falls between them finds the pushed objects in the store but reachable from no tip, and the
/// reachability export leaves them out; the `Refs` frame after the cut then names objects the
/// snapshot never had (R6 on the estate-box gate, 2026-10-09: the recharged main re-applied
/// seq 4104 for twenty minutes). `frames` is `(seq, kind, meta)` in sequence order.
///
/// **Several pushes into one repository** (R6 residual, 5.3.3): a `Refs` frame does not say which
/// pack it closes — A's pack, B's pack, A's refs | cut | B's refs — so it closes ONE push, not the
/// repository's every open pack. Per repository the open pushes are counted (a pack opens one, a
/// refs frame that sets a ref closes one; a refs frame that only deletes closes none), and every
/// pack since the count was last zero is carried: a carried pack whose objects are already
/// reachable costs bytes, a missing one costs the recharge.
pub fn pending_packs(frames: &[(u64, u16, Meta)], at: u64) -> BTreeMap<String, Vec<u64>> {
    let mut open: BTreeMap<String, (u64, Vec<u64>)> = BTreeMap::new();
    for (seq, kind, meta) in frames {
        if *seq > at {
            break;
        }
        match meta {
            Meta::Pack { repo, .. } if *kind == Kind::Pack as u16 => {
                let e = open.entry(repo.clone()).or_default();
                e.0 += 1;
                e.1.push(*seq);
            }
            Meta::Refs { repo, edits, .. }
                if *kind == Kind::Refs as u16 && edits.iter().any(|e| e.new.is_some()) =>
            {
                if let Some(e) = open.get_mut(repo) {
                    e.0 = e.0.saturating_sub(1);
                    if e.0 == 0 {
                        open.remove(repo);
                    }
                }
            }
            _ => {}
        }
    }
    open.into_iter()
        .map(|(repo, (_, seqs))| (repo, seqs))
        .collect()
}

impl Cut {
    /// Under the freeze: nothing framed is in flight, so the stores ARE the log's head.
    fn take(stores: &ServerStores, at: u64) -> anyhow::Result<Self> {
        // M6 — the key the secrets are sealed under: the file when there is one (a twin's is the
        // primary's, applied), else the key this process holds (handed in by the environment).
        let kek = match std::fs::read(gunnar_core::seal::kek_path(stores.control_dir())) {
            Ok(bytes) => bytes,
            Err(e) if not_found(&e) => stores.kek_bytes(),
            Err(e) => return Err(e).context("reading the KEK for a snapshot"),
        };
        // M4, M5 — a read view: redb keeps this instant for as long as the stream needs it.
        let control = stores
            .control_db()
            .view()
            .context("a view of the control database for a snapshot")?;
        // M7, M8 — each framed file as an OPEN handle and its length now: an append-only log is
        // read up to that length, and a document replaced later (a rename) is still this inode.
        let mut files = Vec::new();
        for rel in txlog::FRAMED_FILES {
            match std::fs::File::open(stores.root().join(rel)) {
                Ok(f) => {
                    let len = f
                        .metadata()
                        .with_context(|| format!("measuring {rel} for a snapshot"))?
                        .len();
                    files.push((*rel, f, len));
                }
                Err(e) if not_found(&e) => {}
                Err(e) => return Err(e).with_context(|| format!("reading {rel} for a snapshot")),
            }
        }
        // M9, M2 — every repository and its refs; the objects are content-addressed and stay
        // (the snapshot pin holds back a gc and a purge), so their bytes stream after the freeze.
        // The pushes caught between their `Pack` and their `Refs` at this instant ([`pending_packs`]).
        let mut frames: Vec<(u64, u16, Meta)> = Vec::new();
        stores
            .txlog()
            .scan_meta(|seq, kind, meta| {
                if kind == Kind::Pack as u16 || kind == Kind::Refs as u16 {
                    if let Ok(m) = Meta::from_bytes(meta) {
                        frames.push((seq, kind, m));
                    }
                }
            })
            .context("scanning the log for pushes caught between their frames")?;
        let mut pending_seqs = pending_packs(&frames, at);
        let mut repos = Vec::new();
        for repo in stores.catalog().list(None)? {
            let key = repo.key();
            let hash = repo.hash_kind.to_string();
            let store = stores
                .repo_store(&repo)
                .with_context(|| format!("opening {key} for a snapshot"))?;
            let git = store.unlogged();
            let mut edits = Vec::new();
            let mut tips: Vec<Vec<u8>> = Vec::new();
            for r in git.refs()? {
                if r.name == "HEAD" {
                    continue;
                }
                if let Some(oid) = &r.oid {
                    tips.push(oid.clone());
                }
                edits.push(RefEdit {
                    name: r.name,
                    old: None,
                    new: r.oid.as_deref().map(hex::encode),
                    peeled: r.peeled.as_deref().map(hex::encode),
                    symref: r.symref_target,
                });
            }
            if let Some(target) = git.head()?.and_then(|h| h.symref_target) {
                edits.push(RefEdit {
                    name: "HEAD".into(),
                    old: None,
                    new: None,
                    peeled: None,
                    symref: Some(target),
                });
            }
            let mut pending = Vec::new();
            for seq in pending_seqs.remove(&key).unwrap_or_default() {
                let at = stores
                    .txlog()
                    .locate_blob(seq)
                    .with_context(|| format!("locating the pending pack of {key} at seq {seq}"))?;
                let f = std::fs::File::open(&at.path).with_context(|| {
                    format!("opening the segment of {key}'s pending pack at seq {seq}")
                })?;
                pending.push((f, at.blob_offset, at.blob_len));
            }
            repos.push(RepoCut {
                repo,
                key,
                hash,
                edits,
                tips,
                pending,
            });
        }
        Ok(Cut {
            at,
            kek,
            control,
            files,
            repos,
        })
    }
}

/// **Produce a snapshot of `stores` into `sink`** — a primary for a twin behind a retention
/// hole, or a twin for a restore from it. Answers `at`.
///
/// **The log is frozen for the CUT only** (R5.1-A, 2026-10-07). It used to be frozen for the
/// whole stream — every framed write on the primary (every push, control write and audited RPC)
/// waited until the twin had the last piece of every repository's pack. Now the freeze holds while
/// [`Cut::take`] fixes the state at `at` — the head, a read view of the control database, the
/// framed files' handles and lengths, every repository's refs and tips — and the bytes stream from
/// the cut afterwards, writes going on beside them. A snapshot pin
/// ([`ServerStores::snapshot_pin`], taken BEFORE the freeze) holds back a gc and the removal of a
/// repository store until the stream is done, so what the cut names is still there to read. LFS
/// objects are listed after the freeze: an object stored later is also a later `LfsPut` (applied
/// again, idempotent by oid) and one removed later is a later `LfsDelete` — an object gone by the
/// time it is reached is skipped, and the replay ends in the same state.
pub fn produce(stores: &ServerStores, sink: &mut Sink<'_>) -> anyhow::Result<u64> {
    let log = stores.txlog();
    let _pin = stores.snapshot_pin();
    let t0 = std::time::Instant::now();
    let cut = {
        let _frozen = log.freeze();
        Cut::take(stores, log.head_seq())?
    };
    let at = cut.at;
    tracing::info!(target: gunnar_core::observe::TARGET_TWIN, at, repos = cut.repos.len(), frozen_ms = t0.elapsed().as_millis() as u64, "a snapshot's cut is taken; writes go on while it streams");
    let mut out = Pieces::new(sink);

    out.item(Kind::Kek, &Meta::Kek {}, &cut.kek)?;

    cut.control
        .export_each(&mut |row| match row {
            ExportRow::Blob(id, bytes) => out.item(
                Kind::ControlBlob,
                &Meta::ControlBlob {
                    id: hex::encode(id.as_bytes()),
                },
                bytes,
            ),
            ExportRow::Doc(name, id) => out.item(
                Kind::ControlDoc,
                &Meta::ControlDoc {
                    name: name.to_string(),
                    expected: None,
                    new: hex::encode(id.as_bytes()),
                },
                &[],
            ),
            ExportRow::Secret(key, value) => out.item(
                Kind::Secret,
                &Meta::Secret {
                    key: key.to_string(),
                    deleted: value.is_none(),
                },
                value.unwrap_or_default(),
            ),
        })
        .context("exporting the control database")??;
    // The last-seen table at the cut, as ONE item (2026-10-21): a twin that joins later
    // has every stamp, and keeps it when promoted.
    let activity = cut
        .control
        .principal_activity()
        .context("reading the last-seen table for a snapshot")?;
    if !activity.is_empty() {
        out.item(
            Kind::Activity,
            &Meta::Activity {},
            &gunnar_core::activity::encode_rows(&activity)?,
        )?;
    }

    for (rel, f, len) in &cut.files {
        out.file(
            Kind::File,
            &Meta::File {
                path: rel.to_string(),
                deleted: false,
            },
            f,
            *len,
        )
        .with_context(|| format!("streaming {rel} for a snapshot"))?;
    }

    // M1, M3 — per repository: its store, its objects (the engine's export over the cut's tips,
    // written straight into the pieces), its refs and HEAD as they were at the cut, its LFS.
    for r in &cut.repos {
        let key = &r.key;
        out.item(
            Kind::RepoStore,
            &Meta::RepoStore {
                repo: key.clone(),
                op: "created".into(),
                hash: r.hash.clone(),
            },
            &[],
        )?;
        let store = stores
            .repo_store(&r.repo)
            .with_context(|| format!("opening {key} for a snapshot"))?;
        let git = store.unlogged();
        if !r.tips.is_empty() {
            let want: Vec<&[u8]> = r.tips.iter().map(Vec::as_slice).collect();
            let objects = match git.select(&want, &[])? {
                Some(set) => set.objects,
                None => git.reachable(&want, &[])?,
            };
            let objs: Vec<&[u8]> = objects.iter().map(Vec::as_slice).collect();
            out.begin(
                Kind::Pack,
                &Meta::Pack {
                    repo: key.clone(),
                    hash: r.hash.clone(),
                },
            );
            let emitted = git.emit_pack(&objs, &[], &Caps::modern(), &mut out);
            if let Some(err) = out.failed.take() {
                return Err(err).with_context(|| format!("streaming the objects of {key}"));
            }
            emitted.with_context(|| format!("exporting the objects of {key}"))?;
            out.end()?;
        }
        // ★ Each push caught between its two frames: its pack, as it was framed, so the `Refs` frame
        // after the cut finds every object it names (R6, 2026-10-09).
        for (f, offset, len) in &r.pending {
            out.file_at(
                Kind::Pack,
                &Meta::Pack {
                    repo: key.clone(),
                    hash: r.hash.clone(),
                },
                f,
                *offset,
                *len,
            )
            .with_context(|| format!("streaming a pending pack of {key}"))?;
        }
        if !r.edits.is_empty() {
            out.item(
                Kind::Refs,
                &Meta::Refs {
                    repo: key.clone(),
                    atomic: false,
                    edits: r.edits.clone(),
                },
                &[],
            )?;
        }
        let lfs_objects = store.git_dir().join("lfs").join("objects");
        for path in files_under(&lfs_objects)? {
            let Some(name) = path.file_name().and_then(|n| n.to_str()) else {
                continue;
            };
            if gunnar_lfs::Oid::parse(name).is_err() {
                continue;
            }
            let f = match std::fs::File::open(&path) {
                Ok(f) => f,
                // Removed after the cut: its `LfsDelete` is a later frame (see above).
                Err(e) if not_found(&e) => continue,
                Err(e) => return Err(e).with_context(|| format!("opening LFS {name} of {key}")),
            };
            let size = f.metadata()?.len();
            out.file(
                Kind::LfsPut,
                &Meta::Lfs {
                    repo: key.clone(),
                    oid: name.to_string(),
                    size,
                },
                &f,
                size,
            )
            .with_context(|| format!("streaming LFS {name} of {key}"))?;
        }
    }
    out.item(Kind::SnapshotAt, &Meta::SnapshotAt { at }, &[])?;
    tracing::info!(target: gunnar_core::observe::TARGET_TWIN, at, pieces = out.pieces, seconds = t0.elapsed().as_secs(), "produced a snapshot");
    Ok(at)
}

fn files_under(dir: &Path) -> anyhow::Result<Vec<PathBuf>> {
    let mut out = Vec::new();
    let mut stack = vec![dir.to_path_buf()];
    while let Some(d) = stack.pop() {
        let rd = match std::fs::read_dir(&d) {
            Ok(rd) => rd,
            Err(e) if not_found(&e) => continue,
            Err(e) => return Err(e).with_context(|| format!("listing {}", d.display())),
        };
        for entry in rd {
            let entry = entry?;
            let ft = entry.file_type()?;
            if ft.is_dir() {
                stack.push(entry.path());
            } else if ft.is_file() {
                out.push(entry.path());
            }
        }
    }
    out.sort();
    Ok(out)
}

// ── the volume a snapshot is staged on ──────────────────────────────────────

/// **The volume a snapshot is staged on** (twin-resync-space, 2026-10-06): what the stager asks
/// of it — its size and what is still free — and how a piece is written to it. [`Disk`] is the
/// box's own filesystem (one `statvfs`, one `pwrite`); a test caps a directory in-process, so a
/// twin's volume "the size of main's data set" is a thing the suite can run anywhere.
pub trait Volume: Send + Sync {
    /// `(total, free)` bytes of the volume holding `path` — free as an unprivileged write sees
    /// it (`f_bavail`: the root reserve is not the box's to stage into); `None` when it cannot be
    /// measured, which is never read as "full".
    fn space(&self, path: &Path) -> Option<(u64, u64)>;

    /// `bytes` at `off` of `file`, a file on this volume.
    fn write_at(&self, file: &std::fs::File, bytes: &[u8], off: u64) -> std::io::Result<()> {
        file.write_all_at(bytes, off)
    }
}

/// **The box's own filesystem.**
#[derive(Debug, Clone, Copy, Default)]
pub struct Disk;

impl Volume for Disk {
    fn space(&self, path: &Path) -> Option<(u64, u64)> {
        let v = rustix::fs::statvfs(path).ok()?;
        let frsize = if v.f_frsize == 0 {
            v.f_bsize
        } else {
            v.f_frsize
        };
        let total = v.f_blocks.saturating_mul(frsize);
        (total > 0).then(|| (total, v.f_bavail.saturating_mul(frsize)))
    }
}

/// **What a twin keeps free for itself while a snapshot is staged** (twin-resync-space): 1 % of
/// its volume, at least 1 MiB and at most 256 MiB — room for the install's control-plane writes,
/// its own log lines and its status. A twin is grown to main's data set, and main keeps its own
/// 2 GiB floor and its log budget free inside that set, so main's snapshot plus this fits it.
pub fn staging_floor(total: u64) -> u64 {
    (total / 100).clamp(1 << 20, 256 << 20)
}

fn is_no_space(e: &std::io::Error) -> bool {
    matches!(e.raw_os_error(), Some(c) if c == rustix::io::Errno::NOSPC.raw_os_error() || c == rustix::io::Errno::DQUOT.raw_os_error())
}

/// What a twin gives up for room: its stale copy. Answers the bytes freed.
pub type GiveUp = dyn FnMut(&Short) -> anyhow::Result<u64> + Send;

struct Room {
    give_up: Box<GiveUp>,
    given_up: bool,
}

/// **The volume ran short mid-resync**: bytes staged so far, free of total, the floor kept.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct Short {
    pub staged: u64,
    pub free: u64,
    pub total: u64,
    pub floor: u64,
    /// The volume refused a write (`ENOSPC`) rather than the measure saying so first.
    pub enospc: bool,
}

/// **A resync this twin's volume cannot hold, even with its stale copy given up** — said with
/// [`gunnar_grpc::TWIN_TOO_SMALL`], so the follower waits in `recovering` (and asks main for no
/// further snapshot until the volume is larger) instead of failing the same way every poll.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct TooSmall {
    pub staged: u64,
    pub free: u64,
    pub total: u64,
    pub floor: u64,
}

impl std::fmt::Display for TooSmall {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(
            f,
            "{} this twin's data volume cannot hold its primary's data: a full resync had staged {} when {} of {} was left free (it keeps {} for itself), with its stale copy already given up. Grow this twin's volume — to at least main's data set — and it takes the resync again; until then it waits in `recovering` and asks main for no snapshot it cannot hold",
            gunnar_grpc::TWIN_TOO_SMALL,
            super::human(self.staged),
            super::human(self.free),
            super::human(self.total),
            super::human(self.floor),
        )
    }
}

impl std::error::Error for TooSmall {}

/// **What the [`DROPPED`] marker says**: when, the head the twin held, its base's `at`, and the
/// bytes the give-up freed. Read tolerantly — the marker's PRESENCE is the fact; a field it does
/// not carry (a crash mid-write of an older build's) reads 0.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct Dropped {
    pub at_unix_ms: i64,
    pub head: u64,
    pub base_at: Option<u64>,
    pub freed: u64,
}

impl Dropped {
    fn to_text(&self) -> String {
        format!(
            "dropped_unix_ms {}\nhead {}\nbase_at {}\nfreed {}\n",
            self.at_unix_ms,
            self.head,
            self.base_at
                .map(|a| a.to_string())
                .unwrap_or_else(|| "-".into()),
            self.freed
        )
    }

    fn parse(text: &str) -> Self {
        let mut d = Dropped::default();
        for line in text.lines() {
            let mut w = line.split_whitespace();
            match (w.next(), w.next()) {
                (Some("dropped_unix_ms"), Some(v)) => d.at_unix_ms = v.parse().unwrap_or(0),
                (Some("head"), Some(v)) => d.head = v.parse().unwrap_or(0),
                (Some("base_at"), Some(v)) => d.base_at = v.parse().ok(),
                (Some("freed"), Some(v)) => d.freed = v.parse().unwrap_or(0),
                _ => {}
            }
        }
        d
    }

    /// The sentence the status, the alarm and the refusals carry.
    pub fn sentence(&self) -> String {
        let base = self
            .base_at
            .map(|a| format!(", its kept base at {a}"))
            .unwrap_or_default();
        format!(
            "this twin gave its stale copy up (it held up to seq {}{base}; {} freed) to make room for a full resync: it holds NO copy until the resync lands, and main is the only copy meanwhile",
            self.head,
            super::human(self.freed)
        )
    }
}

/// **Mark the give-up** — durable BEFORE the first byte of the stale copy goes.
pub fn mark_dropped(root: &Path, d: &Dropped) -> anyhow::Result<()> {
    crate::durable::write_atomic_unframed(root, DROPPED, d.to_text().as_bytes())
        .context("marking the stale copy given up")
}

/// The give-up in progress or done, when this box has marked one.
pub fn dropped(root: &Path) -> Option<Dropped> {
    match std::fs::read_to_string(root.join(DROPPED)) {
        Ok(text) => Some(Dropped::parse(&text)),
        Err(e) if not_found(&e) => None,
        // Unreadable but there: the copy is still to be treated as gone.
        Err(_) => Some(Dropped::default()),
    }
}

/// **Remember a resync the volume could not hold**: its size then.
pub fn note_too_small(root: &Path, t: &TooSmall) -> anyhow::Result<()> {
    crate::durable::write_atomic_unframed(
        root,
        TOO_SMALL,
        format!("total {}\nstaged {}\n", t.total, t.staged).as_bytes(),
    )
    .context("noting the resync the volume could not hold")
}

/// The volume size the last resync ran out of room on, when one did.
pub fn too_small_total(root: &Path) -> Option<u64> {
    let text = std::fs::read_to_string(root.join(TOO_SMALL)).ok()?;
    text.lines()
        .find_map(|l| l.strip_prefix("total "))
        .and_then(|v| v.trim().parse().ok())
}

/// Bytes of every file under `dir` (0 when it is not there).
pub fn bytes_under(dir: &Path) -> u64 {
    files_under(dir)
        .map(|fs| {
            fs.iter()
                .filter_map(|p| std::fs::metadata(p).ok())
                .map(|m| m.len())
                .sum()
        })
        .unwrap_or(0)
}

/// **Why this box holds no whole copy now**, or `None` (twin-resync-space): a twin that gave its
/// stale copy up for a resync, or one mid-install — its control plane may already be the new
/// snapshot's while its base is still the old one. A twin with either serves nothing: no
/// snapshot ([`serve`]) and no log, so no recharge reads from it.
pub fn not_whole(root: &Path) -> Option<String> {
    if let Some(d) = dropped(root) {
        return Some(format!("this twin holds no whole copy: {}", d.sentence()));
    }
    if root.join(INSTALLING).exists() {
        return Some(format!(
            "this twin holds no whole copy: it is inside a snapshot install ({INSTALLING}) — its control plane may already be the new snapshot's while its base is still the old one — and takes the snapshot again on its next poll"
        ));
    }
    None
}

// ── the twin's half: staged on disk as it arrives ───────────────────────────

/// One staged item: what it is, and where its blob lies in the staging file.
#[derive(Debug, Clone)]
struct StagedItem {
    kind: u16,
    meta: Vec<u8>,
    off: u64,
    len: u64,
}

/// **A snapshot arriving**: every piece's bytes appended to `<data-root>/txlog-snapshot-staging/
/// blobs` as it comes, the item table in memory. Dropped unfinished, the staging directory goes.
pub struct Stager {
    dir: PathBuf,
    file: Option<std::fs::File>,
    off: u64,
    items: Vec<StagedItem>,
    /// Each staged item's crc32c over `meta ‖ blob` (restore-after-snapshot, 2026-10-07): kept
    /// with a twin's base, sent with every item it serves, checked by the box it restores.
    crcs: Vec<u32>,
    open: Option<StagedItem>,
    /// The open item's running crc.
    open_crc: Crc32c,
    at: Option<u64>,
    /// Kept on drop: a restore's staging, resumed by the next stream ([`RestoreSnapshot`]).
    keep: bool,
    /// The volume the staging file is on.
    volume: std::sync::Arc<dyn Volume>,
    /// A twin's: what it may give up when the volume runs short ([`Stager::making_room`]).
    room: Option<Room>,
}

impl Stager {
    /// Start over under `root`, on the box's own filesystem: whatever an earlier attempt left is
    /// removed first.
    pub fn begin(root: &Path) -> anyhow::Result<Self> {
        Self::begin_on(root, std::sync::Arc::new(Disk))
    }

    /// Start over under `root`, staging on `volume`.
    pub fn begin_on(root: &Path, volume: std::sync::Arc<dyn Volume>) -> anyhow::Result<Self> {
        let dir = root.join(STAGING);
        match std::fs::remove_dir_all(&dir) {
            Ok(()) => {}
            Err(e) if not_found(&e) => {}
            Err(e) => return Err(e).with_context(|| format!("clearing {}", dir.display())),
        }
        std::fs::create_dir_all(&dir).with_context(|| format!("creating {}", dir.display()))?;
        let file = std::fs::OpenOptions::new()
            .create_new(true)
            .write(true)
            .read(true)
            .open(dir.join("blobs"))
            .context("creating the staging file")?;
        Ok(Self {
            dir,
            file: Some(file),
            off: 0,
            items: Vec::new(),
            crcs: Vec::new(),
            open: None,
            open_crc: Crc32c::new(),
            at: None,
            volume,
            room: None,
            keep: false,
        })
    }

    /// **A twin's stager makes room** (twin-resync-space, 2026-10-06): before each piece the
    /// volume is measured, and when the piece would leave less than [`staging_floor`] free — or
    /// the volume answers `ENOSPC` — `give_up` is called ONCE: it gives this twin's stale copy up
    /// (the [`DROPPED`] marker first) and answers the bytes it freed. Short again after that, the
    /// stream ends with [`TooSmall`]. `given_up`: the copy is already gone (a resync restarted
    /// after the marker was written), so a shortage is [`TooSmall`] at once.
    pub fn making_room(mut self, give_up: Box<GiveUp>, given_up: bool) -> Self {
        self.room = Some(Room { give_up, given_up });
        self
    }

    /// Room for `len` more bytes, giving the stale copy up when there is none. `enospc`: the
    /// volume has just refused a write, whatever the measure says.
    fn room_for(&mut self, len: u64, enospc: bool) -> anyhow::Result<()> {
        let staged = self.off;
        let Some(room) = self.room.as_mut() else {
            return Ok(());
        };
        let (total, free) = match self.volume.space(&self.dir) {
            Some(m) => m,
            // Unmeasurable is never "full" — unless the volume itself said so.
            None if enospc => (0, 0),
            None => return Ok(()),
        };
        let floor = staging_floor(total);
        if !enospc && free >= len.saturating_add(floor) {
            return Ok(());
        }
        if room.given_up {
            return Err(TooSmall {
                staged,
                free,
                total,
                floor,
            }
            .into());
        }
        room.given_up = true;
        let freed = (room.give_up)(&Short {
            staged,
            free,
            total,
            floor,
            enospc,
        })?;
        match self.volume.space(&self.dir) {
            Some((total, free)) if free < len.saturating_add(staging_floor(total)) => {
                tracing::warn!(target: gunnar_core::observe::TARGET_TWIN, freed, free, total, "the stale copy is given up and the volume is still short for the resync");
                Err(TooSmall {
                    staged,
                    free,
                    total,
                    floor: staging_floor(total),
                }
                .into())
            }
            _ => Ok(()),
        }
    }

    /// One piece's bytes at `off`, with room made for them on a twin.
    fn write_piece(&mut self, blob: &[u8], off: u64) -> anyhow::Result<()> {
        let len = blob.len() as u64;
        let mut refused = false;
        loop {
            self.room_for(len, refused)?;
            match self
                .volume
                .write_at(self.file.as_ref().expect("staging file"), blob, off)
            {
                Ok(()) => return Ok(()),
                // The volume ran out before the measure said so (a filesystem's own overhead):
                // the same answer — give the stale copy up once, then it is too small.
                Err(e) if is_no_space(&e) && self.room.is_some() => {
                    if refused {
                        let (total, free) = self.volume.space(&self.dir).unwrap_or((0, 0));
                        return Err(TooSmall {
                            staged: self.off,
                            free,
                            total,
                            floor: staging_floor(total),
                        }
                        .into());
                    }
                    refused = true;
                }
                Err(e) => return Err(e).context("writing a snapshot piece to the staging file"),
            }
        }
    }

    /// Take one piece.
    pub fn feed(&mut self, p: Piece<'_>) -> anyhow::Result<()> {
        anyhow::ensure!(
            self.at.is_none(),
            "a snapshot piece arrived after the SnapshotAt terminator"
        );
        let off = self.off;
        match self.open.as_ref() {
            Some(open) => anyhow::ensure!(
                open.kind == p.kind,
                "a snapshot piece of kind {} arrived inside an item of kind {}",
                p.kind,
                open.kind
            ),
            None => {
                self.open = Some(StagedItem {
                    kind: p.kind,
                    meta: p.meta.to_vec(),
                    off,
                    len: 0,
                });
                self.open_crc = Crc32c::new();
                self.open_crc.update(p.meta);
            }
        }
        if !p.blob.is_empty() {
            self.write_piece(p.blob, off)?;
            self.off += p.blob.len() as u64;
            self.open.as_mut().expect("the open item").len += p.blob.len() as u64;
            self.open_crc.update(p.blob);
        }
        if !p.last {
            return Ok(());
        }
        let done = self.open.take().expect("the open item");
        anyhow::ensure!(
            done.len == p.blob_len,
            "snapshot item of kind {}: {} of {} blob bytes arrived",
            done.kind,
            done.len,
            p.blob_len
        );
        self.crcs.push(self.open_crc.finish());
        if done.kind == Kind::SnapshotAt as u16 {
            match Meta::from_bytes(&done.meta) {
                Ok(Meta::SnapshotAt { at }) => self.at = Some(at),
                other => anyhow::bail!("a SnapshotAt terminator whose meta is {other:?}"),
            }
        }
        self.items.push(done);
        Ok(())
    }

    /// Has the terminator arrived?
    pub fn terminated(&self) -> Option<u64> {
        self.at
    }

    /// Every piece arrived: synced, ready to install.
    pub fn finish(mut self) -> anyhow::Result<Staged> {
        anyhow::ensure!(
            self.open.is_none(),
            "the snapshot stream ended inside an item"
        );
        let at = self
            .at
            .ok_or_else(|| anyhow!("the snapshot stream carried no SnapshotAt terminator"))?;
        let file = self.file.take().expect("staging file");
        file.sync_all().context("syncing the staging file")?;
        let staged = Staged {
            at,
            dir: std::mem::take(&mut self.dir),
            file,
            items: std::mem::take(&mut self.items),
            crcs: std::mem::take(&mut self.crcs),
            bytes: self.off,
        };
        Ok(staged)
    }
}

impl Drop for Stager {
    fn drop(&mut self) {
        if !self.dir.as_os_str().is_empty() && !self.keep {
            self.file.take();
            let _ = std::fs::remove_dir_all(&self.dir);
        }
    }
}

/// **A snapshot that arrived whole**, on the twin's data disk. Dropped, the staging directory
/// goes.
pub struct Staged {
    /// The log's head the state was taken at.
    pub at: u64,
    dir: PathBuf,
    file: std::fs::File,
    items: Vec<StagedItem>,
    /// Each item's crc32c (meta ‖ blob), as it was staged.
    crcs: Vec<u32>,
    /// Blob bytes staged.
    pub bytes: u64,
}

impl Staged {
    /// How many items, the terminator included.
    pub fn items(&self) -> usize {
        self.items.len()
    }
}

impl Drop for Staged {
    fn drop(&mut self) {
        let _ = std::fs::remove_dir_all(&self.dir);
    }
}

// ── the twin's kept base (ruling 2026-09-22, "the twin only stores") ────────

/// `<data-root>/txlog-snapshot-base/`: **the snapshot a twin installed, KEPT as bytes** —
/// `blobs` (the staging file, unchanged) and `items` (the item table). Per box, never framed.
///
/// A store-only twin does not apply a snapshot's data items (packs, refs, LFS, repository
/// stores, lines, files), and `restart_after(at)` moves its log past them — so this file IS
/// its copy of everything at or below `at`. A recharge from the twin streams it first
/// ([`serve`]) and the twin's log from `at + 1` after it; the new main installs it whole, and
/// that is where the indexes are built.
pub const BASE: &str = "txlog-snapshot-base";
const BASE_ITEMS: &str = "items";
const BASE_BLOBS: &str = "blobs";
/// Each item's crc32c (meta ‖ blob), `u32 LE` per item (restore-after-snapshot, 2026-10-07).
/// Absent beside a base kept before: the base then serves no crc and is named by its item table.
const BASE_CRCS: &str = "crcs";

/// The item table, length-prefixed: `kind u16 | meta_len u32 | meta | off u64 | len u64`, LE.
fn encode_items(items: &[StagedItem]) -> Vec<u8> {
    let mut out = Vec::with_capacity(items.iter().map(|i| 22 + i.meta.len()).sum());
    for i in items {
        out.extend_from_slice(&i.kind.to_le_bytes());
        out.extend_from_slice(&(i.meta.len() as u32).to_le_bytes());
        out.extend_from_slice(&i.meta);
        out.extend_from_slice(&i.off.to_le_bytes());
        out.extend_from_slice(&i.len.to_le_bytes());
    }
    out
}

fn decode_items(mut b: &[u8]) -> anyhow::Result<Vec<StagedItem>> {
    fn take<'a>(b: &mut &'a [u8], n: usize) -> anyhow::Result<&'a [u8]> {
        anyhow::ensure!(b.len() >= n, "the kept snapshot's item table is truncated");
        let (h, t) = b.split_at(n);
        *b = t;
        Ok(h)
    }
    let mut out = Vec::new();
    while !b.is_empty() {
        let kind = u16::from_le_bytes(take(&mut b, 2)?.try_into()?);
        let mlen = u32::from_le_bytes(take(&mut b, 4)?.try_into()?) as usize;
        let meta = take(&mut b, mlen)?.to_vec();
        let off = u64::from_le_bytes(take(&mut b, 8)?.try_into()?);
        let len = u64::from_le_bytes(take(&mut b, 8)?.try_into()?);
        out.push(StagedItem {
            kind,
            meta,
            off,
            len,
        });
    }
    Ok(out)
}

/// **Keep `snap` as this twin's base**: its item table written beside the staging file, and
/// the staging directory renamed over the previous base. Called by [`install`] on a twin BEFORE
/// the log restarts after `at`, so there is no instant at which the log has moved past data
/// that no file on this box holds (a crash before the rename leaves the install marker, and
/// the twin asks for the snapshot again).
fn keep_as_base(root: &Path, snap: &Staged) -> anyhow::Result<()> {
    crate::durable::write_atomic_unframed(&snap.dir, BASE_ITEMS, &encode_items(&snap.items))
        .context("writing the kept snapshot's item table")?;
    if snap.crcs.len() == snap.items.len() {
        let crcs: Vec<u8> = snap.crcs.iter().flat_map(|c| c.to_le_bytes()).collect();
        crate::durable::write_atomic_unframed(&snap.dir, BASE_CRCS, &crcs)
            .context("writing the kept snapshot's item crcs")?;
    }
    let base = root.join(BASE);
    let old = root.join(format!("{BASE}.old"));
    match std::fs::remove_dir_all(&old) {
        Ok(()) => {}
        Err(e) if not_found(&e) => {}
        Err(e) => return Err(e).with_context(|| format!("clearing {}", old.display())),
    }
    match std::fs::rename(&base, &old) {
        Ok(()) => {}
        Err(e) if not_found(&e) => {}
        Err(e) => return Err(e).with_context(|| format!("moving {} aside", base.display())),
    }
    std::fs::rename(&snap.dir, &base)
        .with_context(|| format!("keeping the snapshot as {}", base.display()))?;
    crate::durable::sync_dir(root)
        .with_context(|| format!("flushing {} into place", base.display()))?;
    let _ = std::fs::remove_dir_all(&old);
    Ok(())
}

/// The `at` of the base this twin keeps, when it keeps one.
pub fn base_at(root: &Path) -> anyhow::Result<Option<u64>> {
    let items = match std::fs::read(root.join(BASE).join(BASE_ITEMS)) {
        Ok(b) => decode_items(&b)?,
        Err(e) if not_found(&e) => return Ok(None),
        Err(e) => return Err(e).context("reading the kept snapshot's item table"),
    };
    match items.last().map(|i| (i.kind, Meta::from_bytes(&i.meta))) {
        Some((k, Ok(Meta::SnapshotAt { at }))) if k == Kind::SnapshotAt as u16 => Ok(Some(at)),
        _ => anyhow::bail!("the kept snapshot does not end with its SnapshotAt terminator"),
    }
}

/// **Stream the base this twin keeps**, piece by piece in the producer's shape (meta on an
/// item's first piece, [`BLOB_LEN_AT_LAST`] until its last), through ONE preallocated
/// [`PIECE_BYTES`] buffer. Answers the base's `at`.
pub fn stream_base(root: &Path, sink: &mut Sink<'_>) -> anyhow::Result<u64> {
    stream_base_from(root, 0, 0, "", &mut |p, _| sink(p)).map(|(at, _)| at)
}

/// **Where a piece of a twin's base goes** (restore-after-snapshot, 2026-10-07): the item's index
/// in the base, the byte of its blob the piece starts at, the item's crc on its last piece (when
/// the base kept crcs), and which base ([`base_id`]). What lets a box being restored stage a base
/// across several capped calls and ask for the rest from where it holds it.
#[derive(Debug, Clone, Copy)]
pub struct PiecePos<'a> {
    pub item: u64,
    pub offset: u64,
    pub crc: Option<u32>,
    pub id: &'a str,
}

/// Where the pieces of a resumable snapshot go: each with its [`PiecePos`] — `None` for a
/// snapshot that cannot be resumed (a primary's own, a new cut every time).
pub type PosSink<'s> = dyn FnMut(Piece<'_>, Option<PiecePos<'_>>) -> anyhow::Result<()> + 's;

/// **A resumed snapshot asked of a twin whose base is another one now** (it took a new snapshot
/// since): never spliced — the asker starts again from the first item.
#[derive(Debug)]
pub struct SnapshotChanged {
    pub asked: String,
    pub now: String,
}

/// The refusal word for [`SnapshotChanged`].
pub const SNAPSHOT_CHANGED: &str = "snapshot-changed";

impl std::fmt::Display for SnapshotChanged {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(
            f,
            "REFUSED [{SNAPSHOT_CHANGED}] the rest of snapshot {} was asked for, and this twin's base is snapshot {} now (it took a \
             new one since): two snapshots are never spliced — ask for {} from its first item",
            self.asked, self.now, self.now
        )
    }
}

impl std::error::Error for SnapshotChanged {}

fn read_crcs(dir: &Path, items: usize) -> Option<Vec<u32>> {
    let b = std::fs::read(dir.join(BASE_CRCS)).ok()?;
    (b.len() == items * 4).then(|| {
        b.chunks_exact(4)
            .map(|c| u32::from_le_bytes(c.try_into().expect("four bytes")))
            .collect()
    })
}

/// **The name of the base this twin keeps**: a digest of its item table, its item crcs and its
/// size — two bases differ in it whenever any item differs (a new snapshot, even at the same `at`).
pub fn base_id(root: &Path) -> anyhow::Result<Option<String>> {
    let dir = root.join(BASE);
    let items = match std::fs::read(dir.join(BASE_ITEMS)) {
        Ok(b) => b,
        Err(e) if not_found(&e) => return Ok(None),
        Err(e) => return Err(e).context("reading the kept snapshot's item table"),
    };
    let mut h = blake3::Hasher::new();
    h.update(b"gunnar-snapshot-base-v1\n");
    h.update(&items);
    h.update(&std::fs::read(dir.join(BASE_CRCS)).unwrap_or_default());
    h.update(
        &std::fs::metadata(dir.join(BASE_BLOBS))
            .map(|m| m.len())
            .unwrap_or(0)
            .to_le_bytes(),
    );
    Ok(Some(h.finalize().to_hex()[..32].to_string()))
}

/// The kept base's size in blob bytes (0 = none).
pub fn base_bytes(root: &Path) -> u64 {
    std::fs::metadata(root.join(BASE).join(BASE_BLOBS))
        .map(|m| m.len())
        .unwrap_or(0)
}

/// **[`stream_base`] from item `from_item`, byte `from_offset` of its blob, on** — each piece
/// with its [`PiecePos`] (restore-after-snapshot, 2026-10-07). The items before `from_item` and
/// the bytes below the offset are not read: a resumed 617 MB base costs the twin its rest per
/// call. `expect_id` (non-empty): the base the asker holds part of — another one now is
/// [`SnapshotChanged`]. An item resumed part-way carries its meta on its first piece again.
/// Answers the base's `at` and its id.
pub fn stream_base_from(
    root: &Path,
    from_item: u64,
    from_offset: u64,
    expect_id: &str,
    sink: &mut PosSink<'_>,
) -> anyhow::Result<(u64, String)> {
    let dir = root.join(BASE);
    let items = decode_items(
        &std::fs::read(dir.join(BASE_ITEMS)).context("reading the kept snapshot's item table")?,
    )?;
    let crcs = read_crcs(&dir, items.len());
    let file = std::fs::File::open(dir.join(BASE_BLOBS)).context("opening the kept snapshot")?;
    let at = base_at(root)?.ok_or_else(|| anyhow!("the kept snapshot vanished"))?;
    let id = base_id(root)?.ok_or_else(|| anyhow!("the kept snapshot vanished"))?;
    if !expect_id.trim().is_empty() && expect_id.trim() != id {
        return Err(SnapshotChanged {
            asked: expect_id.trim().to_owned(),
            now: id,
        }
        .into());
    }
    anyhow::ensure!(
        (from_item as usize) < items.len() || (from_item == 0 && items.is_empty()),
        "item {from_item} of snapshot {id} was asked for, and it has {} item(s)",
        items.len()
    );
    let mut buf = vec![0u8; PIECE_BYTES];
    for (i, item) in items.iter().enumerate().skip(from_item as usize) {
        let mut done = if i as u64 == from_item {
            from_offset
        } else {
            0
        };
        anyhow::ensure!(
            done <= item.len,
            "byte {done} of item {i} of snapshot {id} was asked for, and it is {} byte(s)",
            item.len
        );
        let mut first = true;
        loop {
            let n = (item.len - done).min(PIECE_BYTES as u64) as usize;
            file.read_exact_at(&mut buf[..n], item.off + done)
                .context("reading the kept snapshot")?;
            let offset = done;
            done += n as u64;
            let last = done == item.len;
            let pos = PiecePos {
                item: i as u64,
                offset,
                crc: if last {
                    crcs.as_ref().map(|c| c[i])
                } else {
                    None
                },
                id: &id,
            };
            sink(
                Piece {
                    kind: item.kind,
                    meta: if first { &item.meta } else { &[] },
                    blob_len: if last { item.len } else { BLOB_LEN_AT_LAST },
                    blob: &buf[..n],
                    last,
                },
                Some(pos),
            )?;
            first = false;
            if last {
                break;
            }
        }
    }
    Ok((at, id))
}

/// **`Mirror.SnapshotFetch`, on whichever box is asked.** A primary produces one from its
/// stores ([`produce`]). A twin applies no data, so a snapshot of its stores would be a
/// snapshot of its control plane with every repository missing — it serves the base it KEPT
/// instead, which is exactly the state at its log's `first - 1`. A twin with no base has a log
/// that starts at 1 and needs no snapshot; asking it for one is refused by name rather than
/// answered with half a box.
pub fn serve(stores: &ServerStores, sink: &mut Sink<'_>) -> anyhow::Result<u64> {
    serve_from(stores, 0, 0, "", &mut |p, _| sink(p))
}

/// [`serve`], resumable on a twin (restore-after-snapshot, 2026-10-07): its base from item
/// `from_item`, byte `from_offset` on, every piece with its [`PiecePos`] ([`stream_base_from`]).
/// A primary's snapshot is a new cut every time and is not resumed: asked from anywhere but its
/// start it refuses by name, and its pieces carry no position.
pub fn serve_from(
    stores: &ServerStores,
    from_item: u64,
    from_offset: u64,
    expect_id: &str,
    sink: &mut PosSink<'_>,
) -> anyhow::Result<u64> {
    if !stores.is_twin() {
        anyhow::ensure!(
            from_item == 0 && from_offset == 0 && expect_id.trim().is_empty(),
            "this box is a primary: its snapshot is a new cut every time it is asked, so it cannot be resumed at item {from_item} \
             byte {from_offset} — ask for it from its start (a resumed snapshot is a twin's kept base)"
        );
        return produce(stores, &mut |p| sink(p, None));
    }
    // ★ twin-resync-space: a twin that gave its stale copy up, or is mid-install, holds no state
    // at any one sequence — serving its base then would be serving half a set.
    if let Some(why) = not_whole(stores.root()) {
        anyhow::bail!(
            "{why}. Nothing is streamed: recharge from main, or wait for the resync to land"
        );
    }
    match base_at(stores.root())? {
        Some(_) => stream_base_from(stores.root(), from_item, from_offset, expect_id, sink)
            .map(|(at, _)| at),
        None => anyhow::bail!(
            "this twin keeps no snapshot: its log starts at {} and holds everything from there — \
             replay the log; a snapshot of a store-only twin's stores would carry no repository",
            stores.txlog().first_seq()
        ),
    }
}

/// An item's blob read back: small ones into the reused buffer, large ones mapped.
enum Blob<'b> {
    Inline(&'b [u8]),
    Mapped(memmap2::Mmap),
}

impl std::ops::Deref for Blob<'_> {
    type Target = [u8];
    fn deref(&self) -> &[u8] {
        match self {
            Blob::Inline(b) => b,
            Blob::Mapped(m) => m,
        }
    }
}

fn blob_of<'b>(
    file: &std::fs::File,
    item: &StagedItem,
    buf: &'b mut Vec<u8>,
) -> anyhow::Result<Blob<'b>> {
    if item.len <= INLINE_BYTES {
        buf.clear();
        buf.resize(item.len as usize, 0);
        file.read_exact_at(buf, item.off)
            .context("reading a staged snapshot item")?;
        return Ok(Blob::Inline(buf));
    }
    // SAFETY: the staging file is this box's alone, written only by the stager, which has
    // finished; nothing writes it while the mapping lives.
    let map = unsafe {
        memmap2::MmapOptions::new()
            .offset(item.off)
            .len(item.len as usize)
            .map(file)
    }
    .context("mapping a staged snapshot item")?;
    Ok(Blob::Mapped(map))
}

/// **Install `snap` into `stores`** — a twin, or a fresh box being restored (both follower-only).
/// Afterwards the stores hold exactly the snapshot's state and the log's head is `snap.at`.
pub fn install(stores: &ServerStores, snap: &Staged) -> anyhow::Result<()> {
    anyhow::ensure!(
        stores.txlog().is_follower(),
        "a snapshot installs only on a twin or on a box being restored (a follower-only log)"
    );
    let metas: Vec<Option<Meta>> = snap
        .items
        .iter()
        .map(|i| Meta::from_bytes(&i.meta).ok())
        .collect();
    match (snap.items.last(), metas.last()) {
        (Some(i), Some(Some(Meta::SnapshotAt { at })))
            if i.kind == Kind::SnapshotAt as u16 && *at == snap.at => {}
        _ => anyhow::bail!(
            "the snapshot does not end with its SnapshotAt({}) terminator",
            snap.at
        ),
    }
    let root = stores.root();
    // A store-only twin (ruling 2026-09-22) installs the control plane and KEEPS the rest.
    let twin = stores.is_twin();
    let was_dropped = dropped(root).is_some();
    crate::durable::write_atomic_unframed(root, INSTALLING, format!("{}\n", snap.at).as_bytes())?;

    // Whole, not merged: every store this box holds goes, and every framed file the snapshot
    // does not carry.
    for repo in stores.catalog().list(None)? {
        stores
            .reclaim_repo(&repo.path())
            .with_context(|| format!("clearing {} before the snapshot", repo.key()))?;
    }
    let carried: BTreeSet<&str> = metas
        .iter()
        .filter_map(|m| match m {
            Some(Meta::File { path, .. }) => Some(path.as_str()),
            _ => None,
        })
        .collect();
    // A secret this box holds that the snapshot does not name (one a fresh box minted before it
    // became a twin) is not the source's: tombstoned, as the source would have it.
    let named: BTreeSet<&str> = metas
        .iter()
        .filter_map(|m| match m {
            Some(Meta::Secret { key, .. }) => Some(key.as_str()),
            _ => None,
        })
        .collect();
    for (key, value) in stores.control_db().export()?.secrets {
        if value.is_some() && !named.contains(key.as_str()) {
            use gunnar_store::SecretStore as _;
            stores
                .raw_secrets()
                .delete(&key)
                .with_context(|| format!("removing {key}, which the snapshot does not carry"))?;
        }
    }
    for rel in txlog::FRAMED_FILES {
        if !carried.contains(*rel) {
            match std::fs::remove_file(root.join(rel)) {
                Ok(()) => {}
                Err(e) if not_found(&e) => {}
                Err(e) => {
                    return Err(e).with_context(|| {
                        format!("removing {rel}, which the snapshot does not carry")
                    })
                }
            }
        }
    }

    let mut buf: Vec<u8> = Vec::with_capacity(INLINE_BYTES as usize);
    for (item, meta) in snap.items.iter().zip(&metas) {
        match (Kind::from_u16(item.kind), meta) {
            (Some(Kind::SnapshotAt), _) => {}
            // **An item this build cannot install is not a warning, it is the end of the
            // install** (2026-09-16, lane AUDIT-TWIN). `apply_raw` answers `Applied::Unknown`
            // for a kind a newer box invented, and this loop threw that answer away. On the
            // FOLLOWING path an unknown frame costs nothing because the twin's log still holds
            // it — but a SNAPSHOT item is never in any log: its `seq` is 0, `restart_after`
            // moves the twin's log past it, and the bytes are only in a staging file that is
            // deleted at the end of this function. So a twin taking a snapshot from a newer
            // primary silently and PERMANENTLY dropped every item of a kind it did not know,
            // then reported itself caught up. The install refuses instead: the marker stays,
            // the twin asks for the snapshot again, and it says which kind stopped it.
            (None, _) => anyhow::bail!(
                "the snapshot carries an item of kind {}, which this build cannot install — the \
                 source is a NEWER gunnar. A snapshot item is in no log, so installing the rest \
                 would lose it silently; this box installs nothing and stays where it was.",
                item.kind
            ),
            // The snapshot's pointer, whatever this box's was: a force, not a replayed claim.
            (Some(Kind::ControlDoc), Some(Meta::ControlDoc { name, new, .. })) => {
                let new = BlobId::from_hex(new).ok_or_else(|| {
                    anyhow!("snapshot: document {name} points at {new:?}, not hex")
                })?;
                let raw = stores.raw_control_store();
                let current = raw.read_doc(name)?;
                if current != Some(new) {
                    anyhow::ensure!(
                        raw.commit_doc(name, current, new)?,
                        "snapshot: document {name} moved during the install"
                    );
                }
            }
            // A store-only twin applies the control plane only; the data items stay in the
            // staging file, which is KEPT below as this twin's base (ruling 2026-09-22).
            (_, _) if twin && !super::twin_follows(item.kind, &item.meta) => {}
            (kind, _) => {
                let blob = blob_of(&snap.file, item, &mut buf)?;
                super::apply_raw(stores, 0, item.kind, &item.meta, &blob)
                    .with_context(|| format!("installing a snapshot item ({kind:?})"))?;
            }
        }
    }
    // The identity core's tables are the snapshot's document now (RA5-10), or none.
    if let Some(identity) = stores.identity() {
        crate::identity::follow(root, &identity)
            .map_err(|why| anyhow!("installing a snapshot: {why}"))?;
    }
    if twin {
        keep_as_base(root, snap)?;
    }
    stores.txlog().restart_after(snap.at)?;
    let _ = std::fs::remove_file(root.join(super::FRESH_DOCS));
    // The copy is whole again: the give-up and the too-small note are over. Before the install
    // marker, so a crash between leaves the marker that asks for the snapshot again.
    for done in [DROPPED, TOO_SMALL] {
        match std::fs::remove_file(root.join(done)) {
            Ok(()) => {}
            Err(e) if not_found(&e) => {}
            Err(e) => return Err(e).with_context(|| format!("removing {done} after the install")),
        }
    }
    std::fs::remove_file(root.join(INSTALLING))?;
    if was_dropped {
        tracing::warn!(target: gunnar_core::observe::TARGET_TWIN, at = snap.at, items = snap.items.len(), bytes = snap.bytes, "installed a snapshot; following resumes after it — RESOLVED: the full resync landed and this twin holds a whole copy again");
    } else {
        tracing::warn!(target: gunnar_core::observe::TARGET_TWIN, at = snap.at, items = snap.items.len(), bytes = snap.bytes, "installed a snapshot; following resumes after it");
    }
    Ok(())
}

/// Is an install in progress (a crash mid-install) under `root` — or has a twin given its stale
/// copy up for a resync that has not landed ([`DROPPED`])? Either way only a whole snapshot
/// makes the box a copy again.
pub fn installing(root: &Path) -> bool {
    root.join(INSTALLING).exists() || root.join(DROPPED).exists()
}

/// **One staging slot per DATA ROOT, not one per process.**
///
/// A box has one data root, so in the appliance a single slot and this map
/// behave identically — this changes nothing an appliance does. It matters for
/// a process driving two roots, which is the test binary: with one slot,
/// [`restore_piece`] replaced a stager whose `dir` did not match, and replacing
/// it DROPS it, which deletes the other root's staging directory under a stream
/// that is still arriving. Latent rather than observed — the suite flake of
/// 2026-09-16 that prompted the look turned out to be an unindexed pack, not
/// this — but a per-root resource in a per-process slot is wrong on its own,
/// and it is wrong in the twin's most important path.
static RESTORE_STAGERS: Mutex<BTreeMap<PathBuf, Stager>> = Mutex::new(BTreeMap::new());

/// **Forget whatever a broken `Bootstrap.Restore` left staged** — called by
/// [`super::restore_begin`], so every restore stream stages from nothing.
///
/// [`restore_piece`] keeps the stager in a process-global slot that only the terminator or an
/// error inside [`Stager::feed`] empties. A stream that BREAKS empties neither: the half that
/// arrived stayed in the slot, and the resent stream's pieces were appended to it, so [`install`]
/// ran over the union of two snapshots. Mostly that is the same data twice; where the two differ
/// it is silent divergence, because `install` builds its `carried` and `named` sets from every
/// item it holds — a framed file or a secret the SOURCE deleted between the attempts was
/// "carried" by the stale half and kept, with its stale bytes, on a box whose whole contract is
/// that it holds exactly what the source holds. Installed whole, never merged.
///
/// Dropping the stager takes its staging directory with it ([`Stager`]'s `Drop`).
///
/// Scoped to `root` by [`RESTORE_STAGERS`]' key: a process driving two roots (the tests) must
/// not have one root's `restore_begin` — or one root's first piece — throw the other's stream
/// away mid-flight.
pub fn forget_restore_stager(root: &Path) {
    let mut slots = RESTORE_STAGERS.lock().unwrap_or_else(|p| p.into_inner());
    let _ = slots.remove(root);
}

/// **`Bootstrap.Restore`, a snapshot piece** (`seq` 0): staged on disk until the `SnapshotAt`
/// terminator, which installs the snapshot and answers `Some(at)` — the restore then continues
/// with the log from `at + 1`.
pub fn restore_piece(stores: &ServerStores, p: Piece<'_>) -> anyhow::Result<Option<u64>> {
    let root = stores.root().to_path_buf();
    let mut slots = RESTORE_STAGERS.lock().unwrap_or_else(|p| p.into_inner());
    // This root's stager, begun if this is the stream's first piece. Another root's stager is
    // left exactly where it is: it belongs to a stream that is still arriving.
    if !slots.contains_key(&root) {
        slots.insert(root.clone(), Stager::begin(&root)?);
    }
    let fed = slots.get_mut(&root).expect("stager").feed(p);
    if let Err(err) = fed {
        slots.remove(&root);
        return Err(err);
    }
    if slots.get(&root).expect("stager").terminated().is_none() {
        return Ok(None);
    }
    let staged = slots.remove(&root).expect("stager").finish()?;
    install(stores, &staged)?;
    // Whatever a resumable stream had staged of a twin's base is no part of this box now.
    forget_restore_snapshot_or_warn(&root);
    Ok(Some(staged.at))
}

// ── a twin's base, staged on the box being restored, across streams ─────────────────────────

/// `<data-root>/txlog-restore-snapshot/`: **the twin's base a box being restored holds part of**
/// (restore-after-snapshot, 2026-10-07) — `blobs` (its items' bytes, in order) and [`RESTORE_RECORD`]
/// (which base, the items staged whole, the open item and the bytes of it that are DURABLE). Per
/// box, never framed. Gone when the snapshot is installed, when the restore finishes, and when a
/// fresh box begins one.
pub const RESTORE_SNAPSHOT: &str = "txlog-restore-snapshot";
const RESTORE_RECORD: &str = "restore.rec";
const RESTORE_MAGIC: &[u8] = b"gunnar-restore-snapshot v1\n";

/// Bytes staged between two durable commits of a restore's snapshot.
pub const SNAPSHOT_COMMIT_EVERY: u64 = 8 << 20;

/// **What a box being restored holds of a twin's base**, as its record says: what
/// `Bootstrap.Status.restore_snapshot_*` answers, and where the next stream resumes it.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct HeldSnapshot {
    /// The base ([`base_id`] on the twin).
    pub id: String,
    /// Items staged whole.
    pub items: u64,
    /// Bytes of the next item staged.
    pub bytes: u64,
    /// The terminator is staged: the whole base is here, waiting to be installed.
    pub whole: bool,
}

/// The restore's record, decoded.
struct Record {
    id: String,
    off: u64,
    at: Option<u64>,
    items: Vec<StagedItem>,
    crcs: Vec<u32>,
    open: Option<(StagedItem, u32)>,
}

fn encode_record(id: &str, st: &Stager) -> Vec<u8> {
    let mut out = RESTORE_MAGIC.to_vec();
    out.extend_from_slice(&(id.len() as u32).to_le_bytes());
    out.extend_from_slice(id.as_bytes());
    out.extend_from_slice(&st.off.to_le_bytes());
    out.push(u8::from(st.at.is_some()));
    out.extend_from_slice(&st.at.unwrap_or(0).to_le_bytes());
    let items = encode_items(&st.items);
    out.extend_from_slice(&(items.len() as u64).to_le_bytes());
    out.extend_from_slice(&items);
    out.extend_from_slice(&(st.crcs.len() as u32).to_le_bytes());
    for c in &st.crcs {
        out.extend_from_slice(&c.to_le_bytes());
    }
    match &st.open {
        Some(o) => {
            out.push(1);
            out.extend_from_slice(&encode_items(std::slice::from_ref(o)));
            out.extend_from_slice(&st.open_crc.state().to_le_bytes());
        }
        None => out.push(0),
    }
    out
}

fn decode_record(b: &[u8]) -> anyhow::Result<Record> {
    fn take<'a>(b: &mut &'a [u8], n: usize) -> anyhow::Result<&'a [u8]> {
        anyhow::ensure!(b.len() >= n, "the restore's snapshot record is truncated");
        let (h, t) = b.split_at(n);
        *b = t;
        Ok(h)
    }
    let mut b = b;
    anyhow::ensure!(
        take(&mut b, RESTORE_MAGIC.len())? == RESTORE_MAGIC,
        "not a restore's snapshot record"
    );
    let n = u32::from_le_bytes(take(&mut b, 4)?.try_into()?) as usize;
    let id = String::from_utf8(take(&mut b, n)?.to_vec())?;
    let off = u64::from_le_bytes(take(&mut b, 8)?.try_into()?);
    let has_at = take(&mut b, 1)?[0] == 1;
    let at = u64::from_le_bytes(take(&mut b, 8)?.try_into()?);
    let n = u64::from_le_bytes(take(&mut b, 8)?.try_into()?) as usize;
    let items = decode_items(take(&mut b, n)?)?;
    let n = u32::from_le_bytes(take(&mut b, 4)?.try_into()?) as usize;
    let mut crcs = Vec::with_capacity(n);
    for _ in 0..n {
        crcs.push(u32::from_le_bytes(take(&mut b, 4)?.try_into()?));
    }
    let open = if take(&mut b, 1)?[0] == 1 {
        anyhow::ensure!(b.len() >= 4, "the restore's open item is truncated");
        let item = decode_items(&b[..b.len() - 4])?
            .pop()
            .ok_or_else(|| anyhow!("the restore's open item is missing"))?;
        let state = u32::from_le_bytes(b[b.len() - 4..].try_into()?);
        Some((item, state))
    } else {
        None
    };
    anyhow::ensure!(
        crcs.len() == items.len(),
        "the restore's snapshot record has {} crcs for {} items",
        crcs.len(),
        items.len()
    );
    let end = open
        .as_ref()
        .map(|(o, _)| o.off + o.len)
        .or_else(|| items.last().map(|i| i.off + i.len))
        .unwrap_or(0);
    anyhow::ensure!(
        end == off,
        "the restore's snapshot record counts {off} bytes and its items end at {end}"
    );
    Ok(Record {
        id,
        off,
        at: has_at.then_some(at),
        items,
        crcs,
        open,
    })
}

/// **What this box holds of a twin's base**, out of its record (the durable part, which is what
/// the next stream finds once the staging is cut back to it). `None`: nothing staged.
pub fn held_snapshot(root: &Path) -> Option<HeldSnapshot> {
    let r = decode_record(&std::fs::read(root.join(RESTORE_SNAPSHOT).join(RESTORE_RECORD)).ok()?)
        .ok()?;
    Some(HeldSnapshot {
        id: r.id,
        items: r.items.len() as u64,
        bytes: r.open.as_ref().map(|(o, _)| o.len).unwrap_or(0),
        whole: r.at.is_some(),
    })
}

/// **Throw a restore's staged snapshot away, durably** (the directory, then the data root's entry).
/// Nothing staged is `Ok`; a removal or a directory fsync that failed is the caller's error.
pub fn forget_restore_snapshot(root: &Path) -> anyhow::Result<()> {
    let dir = root.join(RESTORE_SNAPSHOT);
    match std::fs::remove_dir_all(&dir) {
        Ok(()) => {}
        Err(e) if not_found(&e) => return Ok(()),
        Err(e) => return Err(e).with_context(|| format!("removing {}", dir.display())),
    }
    crate::durable::sync_dir(root).with_context(|| {
        format!(
            "flushing the removal of {RESTORE_SNAPSHOT} from {}",
            root.display()
        )
    })
}

/// [`forget_restore_snapshot`] where what it throws away is already no part of the box (the
/// snapshot installed, the restore finished): a failure is said, by name, and is not the
/// caller's — the next restore start throws it away again.
pub fn forget_restore_snapshot_or_warn(root: &Path) {
    if let Err(err) = forget_restore_snapshot(root) {
        tracing::warn!(target: gunnar_core::observe::TARGET_TWIN, err = %format!("{err:#}"), "a restore's staged snapshot could not be thrown away durably; the next restore start throws it away again");
    }
}

/// **A twin's base staged on the box being restored, ACROSS streams** (restore-after-snapshot,
/// 2026-10-07). The [`Stager`] a twin's own resync uses, in [`RESTORE_SNAPSHOT`], kept on drop,
/// with its state written to [`RESTORE_RECORD`] after the bytes it counts were synced — every
/// [`SNAPSHOT_COMMIT_EVERY`], when a stream ends, and on drop. A crash between a write and its
/// commit leaves bytes nobody counted, and [`RestoreSnapshot::resume`] cuts them off.
pub struct RestoreSnapshot {
    stager: Stager,
    id: String,
    /// Blob bytes durable (the record's count).
    committed: u64,
    /// Items in the last record written.
    committed_items: usize,
}

/// A staged item that does not match the crc the twin vouched for.
#[derive(Debug)]
pub struct ItemCrcMismatch {
    pub item: u64,
    pub got: u32,
    pub want: u32,
}

impl std::fmt::Display for ItemCrcMismatch {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(
            f,
            "item {} of the snapshot was put together as crc {:08x} and the twin kept it as {:08x}",
            self.item, self.got, self.want
        )
    }
}

impl std::error::Error for ItemCrcMismatch {}

impl RestoreSnapshot {
    /// **What a previous stream staged, made trustworthy**: the staging file cut back to the
    /// record's durable count. `None`: nothing (or nothing trustworthy — a record that does not
    /// parse, or a file SHORTER than its count — which is then thrown away).
    pub fn resume(root: &Path) -> anyhow::Result<Option<RestoreSnapshot>> {
        let dir = root.join(RESTORE_SNAPSHOT);
        let rec = match std::fs::read(dir.join(RESTORE_RECORD)) {
            Ok(b) => b,
            Err(e) if not_found(&e) => {
                forget_restore_snapshot(root)?;
                return Ok(None);
            }
            Err(e) => return Err(e).context("reading the restore's snapshot record"),
        };
        let r = match decode_record(&rec) {
            Ok(r) => r,
            Err(err) => {
                tracing::warn!(target: gunnar_core::observe::TARGET_TWIN, %err, "the restore's staged snapshot has a record that does not parse; it is dropped and the snapshot starts again");
                forget_restore_snapshot(root)?;
                return Ok(None);
            }
        };
        let file = match std::fs::OpenOptions::new()
            .read(true)
            .write(true)
            .open(dir.join("blobs"))
        {
            Ok(f) => f,
            Err(_) => {
                forget_restore_snapshot(root)?;
                return Ok(None);
            }
        };
        let len = file
            .metadata()
            .context("reading the restore's staged snapshot")?
            .len();
        if len < r.off {
            tracing::warn!(target: gunnar_core::observe::TARGET_TWIN, id = %r.id, durable = r.off, on_disk = len, "the restore's staged snapshot is SHORTER than its record says was made durable; it is not trusted and starts again");
            drop(file);
            forget_restore_snapshot(root)?;
            return Ok(None);
        }
        if len > r.off {
            file.set_len(r.off)
                .context("cutting the restore's staged snapshot back to its durable count")?;
            file.sync_all()
                .context("flushing the restore's staged snapshot")?;
        }
        let (open, open_crc) = match r.open {
            Some((o, state)) => (Some(o), Crc32c::resume(state)),
            None => (None, Crc32c::new()),
        };
        let items = r.items.len();
        let stager = Stager {
            dir,
            file: Some(file),
            off: r.off,
            items: r.items,
            crcs: r.crcs,
            open,
            open_crc,
            at: r.at,
            volume: std::sync::Arc::new(Disk),
            room: None,
            keep: true,
        };
        Ok(Some(RestoreSnapshot {
            stager,
            id: r.id,
            committed: r.off,
            committed_items: items,
        }))
    }

    /// **Stage snapshot `id` from its first item**: whatever was staged before goes.
    pub fn start(root: &Path, id: &str) -> anyhow::Result<RestoreSnapshot> {
        forget_restore_snapshot(root)?;
        let dir = root.join(RESTORE_SNAPSHOT);
        std::fs::create_dir_all(&dir).with_context(|| format!("creating {}", dir.display()))?;
        let file = std::fs::OpenOptions::new()
            .create_new(true)
            .write(true)
            .read(true)
            .open(dir.join("blobs"))
            .context("creating the restore's snapshot staging file")?;
        let stager = Stager {
            dir,
            file: Some(file),
            off: 0,
            items: Vec::new(),
            crcs: Vec::new(),
            open: None,
            open_crc: Crc32c::new(),
            at: None,
            volume: std::sync::Arc::new(Disk),
            room: None,
            keep: true,
        };
        let mut s = RestoreSnapshot {
            stager,
            id: id.to_owned(),
            committed: 0,
            committed_items: 0,
        };
        s.write_record()?;
        crate::durable::sync_dir(root)
            .with_context(|| format!("flushing {} into {}", RESTORE_SNAPSHOT, root.display()))?;
        Ok(s)
    }

    /// Which base.
    pub fn id(&self) -> &str {
        &self.id
    }

    /// `(items staged whole, bytes of the next one staged)` — in memory, the next byte wanted.
    pub fn position(&self) -> (u64, u64) {
        (
            self.stager.items.len() as u64,
            self.stager.open.as_ref().map(|o| o.len).unwrap_or(0),
        )
    }

    /// The terminator is staged: the snapshot's `at`.
    pub fn whole(&self) -> Option<u64> {
        self.stager.at
    }

    /// One piece (already cut to the bytes not held). `want_crc`: the twin's crc of the item this
    /// piece ends — checked before the item counts as staged ([`ItemCrcMismatch`]).
    pub fn feed(&mut self, p: Piece<'_>, want_crc: Option<u32>) -> anyhow::Result<()> {
        let before = self.stager.items.len();
        self.stager.feed(p)?;
        if self.stager.items.len() > before {
            if let (Some(want), Some(&got)) = (want_crc, self.stager.crcs.last()) {
                if got != want {
                    return Err(ItemCrcMismatch {
                        item: before as u64,
                        got,
                        want,
                    }
                    .into());
                }
            }
        }
        if self.stager.off - self.committed >= SNAPSHOT_COMMIT_EVERY || self.stager.at.is_some() {
            self.commit()?;
        }
        Ok(())
    }

    /// **Make what is staged durable and count it**: the file synced, then the record.
    pub fn commit(&mut self) -> anyhow::Result<()> {
        if self.stager.off == self.committed
            && self.stager.items.len() == self.committed_items
            && self.stager.at.is_none()
        {
            return Ok(());
        }
        self.stager
            .file
            .as_ref()
            .expect("staging file")
            .sync_data()
            .context("fdatasync of the restore's staged snapshot")?;
        self.write_record()
    }

    fn write_record(&mut self) -> anyhow::Result<()> {
        crate::durable::write_atomic_unframed(
            &self.stager.dir,
            RESTORE_RECORD,
            &encode_record(&self.id, &self.stager),
        )?;
        self.committed = self.stager.off;
        self.committed_items = self.stager.items.len();
        Ok(())
    }

    /// The whole base, ready to install. Dropped after the install, it takes the staging with it.
    pub fn finish(mut self) -> anyhow::Result<Staged> {
        self.commit()?;
        let mut stager = std::mem::replace(
            &mut self.stager,
            Stager {
                dir: PathBuf::new(),
                file: None,
                off: 0,
                items: Vec::new(),
                crcs: Vec::new(),
                open: None,
                open_crc: Crc32c::new(),
                at: None,
                volume: std::sync::Arc::new(Disk),
                room: None,
                keep: true,
            },
        );
        stager.keep = false;
        stager.finish()
    }
}

impl Drop for RestoreSnapshot {
    /// A stream that never reached its end (dropped, refused, broken) keeps every byte it staged.
    fn drop(&mut self) {
        if self.stager.dir.as_os_str().is_empty() {
            return;
        }
        if let Err(err) = self.commit() {
            tracing::warn!(target: gunnar_core::observe::TARGET_TWIN, id = %self.id, %err, "the restore's staged snapshot could not be committed as its stream ended; the next stream resumes from its last commit");
        }
    }
}