znippy_plugin_git/

archive_file.rs

//! **A repository's stored bytes, in ONE FILE.** `archive_out` writes it,
//! `archive_in` puts it back.
//!
//! This is the file-shaped sibling of [`crate::replicate`], and it exists
//! because that module's decisions were the right ones and its *transport* was
//! the part nobody had settled. A file needs no transport: no socket, no
//! protocol, no peer, no handshake — and a file can be copied to another
//! appliance, to object storage, to a laptop or to a tape by whatever already
//! moves files. **Nothing here opens a socket either**, for exactly the reason
//! [`crate::replicate`] gives.
//!
//! Every decision below is [`crate::replicate`]'s, restated because it now has
//! a body:
//!
//! * **STORE-side, one repository at a time.** The unit is one repository's
//!   bytes between two watermarks. It lands whole or it does not land.
//! * **`from_generation` is a WATERMARK, not a diff.** [`ArchiveWrite`] only
//!   ever *appends* to `objects.pack`, so everything above a byte offset is new
//!   and nothing below it has changed. That is why the parameter is one `u64`
//!   and not a manifest — and why an incremental archive is a `pread` of a
//!   suffix rather than a comparison of two trees.
//! * **NO INDEX TRAVELS.** `objects.tail` (redb) and `objects.exploded` (Arrow)
//!   are *derivations*, and shipping a derivation is how two copies of a
//!   repository come to disagree about the same bytes. The receiver rebuilds
//!   its own: [`crate::git_ops::GitStore::open`] diffs the journal it is handed
//!   against the index rows it has (none, in a fresh store) and re-queues every
//!   pack for indexing. **A restore is therefore three files and an `open`.**
//!
//! # What is truth, and what is not
//!
//! From the table in [`crate::git_ops::GitStore`]:
//!
//! | file | what it is | travels? |
//! |---|---|---|
//! | `objects.pack` | the verbatim pushed pack bytes | **yes** — the truth (§14) |
//! | `objects.pack.journal` | the durable extent rows behind the ack | **yes** — the indexed bit is derived from it |
//! | `refs.log` | the ref transaction log, one Arrow IPC frame per push | **yes** — refs are not derivable from objects |
//! | `objects.tail` | the un-sealed index tail (redb) | no — derivation |
//! | `objects.exploded` | §14's resolved content cache | no — derivation, explicitly droppable |
//!
//! Measured on a one-commit repository: the three that travel are **4 274
//! bytes**; the directory on disk is **1.59 MiB**, because `objects.tail` is a
//! redb file with redb's own floor. A block-level copy moves the second number.
//!
//! # The journal and the ref log ship WHOLE, and that is deliberate
//!
//! Only `objects.pack` is watermarked. `objects.pack.journal` is 24-byte rows
//! and `refs.log` is one small Arrow frame per push: at a thousand pushes they
//! are together a fraction of one pack. Cutting them too would buy little and
//! would need three watermarks where the settled signature has one. The format
//! nevertheless records `from_offset` **per segment**, so a later version can
//! watermark them without a format change — the reader already honours it.
//!
//! # A GC invalidates a watermark, and the file says so rather than lying
//!
//! [`crate::gc`] compacts into a *new generation*: after it, byte 0 of
//! `objects.pack` may not be the byte 0 the previous archive measured its
//! watermark against. So every archive records `base_prefix_sha256` — the
//! SHA-256 of the first [`PREFIX_BYTES`] bytes of `objects.pack` at the instant
//! of the cut — and an incremental cut checks it against the base's. A
//! mismatch is refused **by name** (`the base has been rewritten…`) and the
//! answer is a full archive, not a silent half one.
//!
//! # The file
//!
//! ```text
//! magic "GUNNARAF"          8
//! version u32 LE            4   = FORMAT_VERSION
//! flags   u32 LE            4   = 0
//! from_generation u64 LE    8   the objects.pack watermark this file starts above
//! to_generation   u64 LE    8   objects.pack length at the cut
//! base_prefix_sha256       32   SHA-256 of objects.pack[0 .. min(len, PREFIX_BYTES)]
//! segment_count u32 LE      4
//! reserved      u32 LE      4
//! per segment:  name_len u16 LE, name bytes, from_offset u64 LE, len u64 LE, sha256 [32]
//! payloads, back to back, in the order the segments were listed
//! trailer_sha256           32   SHA-256 of every byte before it
//! ```
//!
//! Names are **relative to the repository directory** (`OBJECT_FORMAT`,
//! `gunnar-git.znippy/objects.pack`, …), so a restore reconstructs the shape
//! rather than guessing it, and a reader can say what a file contains without
//! knowing which engine wrote it.

use std::fs::File;
use std::io::{BufWriter, Read, Seek, SeekFrom, Write};
use std::path::{Path, PathBuf};

use anyhow::{bail, Context, Result};
use sha2::{Digest, Sha256};

/// The first eight bytes of every archive file.
pub const MAGIC: &[u8; 8] = b"GUNNARAF";

/// Bumped when the layout above changes. A reader refuses a version it does not
/// know rather than reading the bytes it recognises out of a file it does not.
pub const FORMAT_VERSION: u32 = 1;

/// How much of `objects.pack` the base fingerprint covers. One 64 KiB `pread`,
/// which is cheap enough to do on every cut and long enough that two different
/// bases do not collide by accident.
pub const PREFIX_BYTES: u64 = 65_536;

/// The file that is watermarked. Append-only; see the module docs.
pub const PACK: &str = "objects.pack";
/// The durable extent rows. The `indexed` bit is derived from this.
pub const JOURNAL: &str = "objects.pack.journal";
/// The ref transaction log.
pub const REFS: &str = "refs.log";
/// The hash-kind marker, one level up from the store directory.
pub const OBJECT_FORMAT: &str = "OBJECT_FORMAT";

/// Copy buffer. One page-aligned megabyte: big enough that a multi-gigabyte
/// pack is not a syscall storm, small enough to sit on the stack of nobody.
const COPY_BUF: usize = 1 << 20;

/// One segment as the header describes it.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Segment {
    /// Path relative to the repository directory.
    pub name: String,
    /// The offset in the source file this segment starts at. `0` = whole file.
    pub from_offset: u64,
    /// How many bytes travel.
    pub len: u64,
    /// SHA-256 of exactly those bytes.
    pub sha256: [u8; 32],
}

/// What a completed [`archive_out`] did.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ArchiveOutReceipt {
    /// The watermark this archive started from.
    pub from_generation: u64,
    /// The watermark to pass as `from_generation` next time.
    pub to_generation: u64,
    /// Bytes of payload — what actually moved, not the file's size.
    pub bytes_out: u64,
    /// The segments, in the order they were written.
    pub segments: Vec<Segment>,
    /// SHA-256 of the whole file.
    pub file_sha256: [u8; 32],
}

impl ArchiveOutReceipt {
    /// Zero payload is a legitimate, successful outcome: the destination was
    /// already current. It is reported rather than turned into an error.
    pub fn is_empty(&self) -> bool {
        self.bytes_out == 0
    }
}

/// What a completed [`archive_in`] did.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ArchiveInReceipt {
    /// Where the repository was written.
    pub repo_dir: PathBuf,
    /// The watermark the restored `objects.pack` now stands at.
    pub to_generation: u64,
    /// Bytes written into the store.
    pub bytes_in: u64,
    /// Segment names restored, in order.
    pub restored: Vec<String>,
}

/// The header, as read back.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ArchiveHeader {
    pub version: u32,
    pub flags: u32,
    pub from_generation: u64,
    pub to_generation: u64,
    pub base_prefix_sha256: [u8; 32],
    pub segments: Vec<Segment>,
    /// Byte offset in the file where the first payload starts.
    pub payload_start: u64,
}

/// **Find the directory that holds `objects.pack`.**
///
/// A repository directory either *is* the store root or holds exactly one
/// subdirectory that is (gunnar's engine names it `gunnar-git.znippy`). This is
/// the only place that shape is interpreted, and it is interpreted by looking
/// for the file rather than by knowing the name — a store laid down by a
/// differently-named engine restores the same way.
pub fn find_store_dir(repo_dir: &Path) -> Result<PathBuf> {
    if repo_dir.join(PACK).is_file() {
        return Ok(repo_dir.to_path_buf());
    }
    let mut found = None;
    for entry in std::fs::read_dir(repo_dir)
        .with_context(|| format!("reading the repository directory {}", repo_dir.display()))?
    {
        let entry = entry?;
        if entry.path().join(PACK).is_file() {
            if found.is_some() {
                bail!(
                    "{} holds more than one store directory; refusing to guess which one is the \
                     repository",
                    repo_dir.display()
                );
            }
            found = Some(entry.path());
        }
    }
    found.ok_or_else(|| {
        anyhow::anyhow!(
            "no {PACK} under {} — that file is the truth an archive is made of, and a directory \
             without one is not a store this can archive",
            repo_dir.display()
        )
    })
}

/// SHA-256 of `objects.pack`'s first [`PREFIX_BYTES`] bytes: the base
/// fingerprint an incremental cut is checked against. See the module docs on
/// GC.
pub fn base_prefix(store_dir: &Path) -> Result<[u8; 32]> {
    let path = store_dir.join(PACK);
    let mut f = File::open(&path).with_context(|| format!("opening {}", path.display()))?;
    let len = f.metadata()?.len().min(PREFIX_BYTES);
    let mut buf = vec![0u8; len as usize];
    f.read_exact(&mut buf)
        .with_context(|| format!("reading the base prefix of {}", path.display()))?;
    let mut h = Sha256::new();
    h.update(&buf);
    Ok(h.finalize().into())
}

/// **Write one repository's bytes above `from_generation` into `out`.**
///
/// `repo_dir` is the repository directory (the one holding `OBJECT_FORMAT` and
/// the store directory). `from_generation` is an `objects.pack` byte offset:
/// `0` for a full archive, a previous receipt's `to_generation` for an
/// incremental one.
///
/// The pack is read while the server may still be appending to it. That is safe
/// and it is the whole point of an append-only file: the cut is taken at a
/// length read once, up front, and every byte below that length is already
/// durable (`ArchiveWrite::append` fsyncs the blob, then the journal row, then
/// fsyncs that). Bytes appended *after* the cut simply belong to the next
/// archive. **There is no torn state to quiesce for.**
///
/// This is [`archive_out_to`] plus a file: the create, the `flush` and the
/// `sync_all` that a *file* destination owes and a socket does not.
pub fn archive_out(repo_dir: &Path, out: &Path, from_generation: u64) -> Result<ArchiveOutReceipt> {
    if let Some(parent) = out.parent() {
        std::fs::create_dir_all(parent).ok();
    }
    let file = File::create(out).with_context(|| format!("creating {}", out.display()))?;
    let mut w = BufWriter::new(file);
    let receipt = archive_out_to(repo_dir, &mut w, from_generation)
        .with_context(|| format!("writing the archive {}", out.display()))?;
    w.flush()
        .with_context(|| format!("flushing {}", out.display()))?;
    w.into_inner()
        .map_err(|e| anyhow::anyhow!("flushing {}: {e}", out.display()))?
        .sync_all()
        .with_context(|| format!("fsyncing {}", out.display()))?;
    Ok(receipt)
}

/// **The same bytes, into any [`Write`] — a file, a socket, a gRPC stream.**
///
/// [`archive_out`] is this function with a `File` under it. The split exists
/// because `archive-out` that can only write a path **on the appliance** is
/// useless for off-box backup, and gunnar's box has no shell to copy the file
/// out with: the control plane streams these bytes instead
/// (`Repos.ArchiveOut`). Nothing here opens a socket — the caller owns the
/// transport, exactly as `crate::replicate` says it must.
///
/// It writes strictly forward and never seeks, so a consumer may hand it a
/// pipe. It does **not** flush or fsync: those belong to whoever owns the
/// destination, and a network sink has neither.
pub fn archive_out_to<W: Write>(
    repo_dir: &Path,
    out: &mut W,
    from_generation: u64,
) -> Result<ArchiveOutReceipt> {
    let store = find_store_dir(repo_dir)?;
    let pack = store.join(PACK);
    let pack_len = std::fs::metadata(&pack)
        .with_context(|| format!("measuring {}", pack.display()))?
        .len();
    if from_generation > pack_len {
        bail!(
            "the watermark {from_generation} is above {}'s length {pack_len} — the base has been \
             rewritten (a gc compacts into a new generation, see the module docs). Take a full \
             archive with --from-generation 0.",
            pack.display()
        );
    }
    let base_prefix_sha256 = base_prefix(&store)?;

    // The segments, in the order they are written. `OBJECT_FORMAT` first
    // because a restore needs the hash kind before anything else means
    // anything; the pack next because it is the big one and a reader that
    // streams gets the expensive part flowing early.
    let rel_store = store
        .strip_prefix(repo_dir)
        .map(|p| p.to_path_buf())
        .unwrap_or_default();
    let mut planned: Vec<(PathBuf, String, u64)> = Vec::new();
    let of = repo_dir.join(OBJECT_FORMAT);
    if of.is_file() {
        planned.push((of, OBJECT_FORMAT.to_string(), 0));
    }
    let rel = |name: &str| -> String {
        if rel_store.as_os_str().is_empty() {
            name.to_string()
        } else {
            format!("{}/{}", rel_store.display(), name)
        }
    };
    planned.push((pack.clone(), rel(PACK), from_generation));
    for name in [JOURNAL, REFS] {
        let p = store.join(name);
        if p.is_file() {
            planned.push((p, rel(name), 0));
        }
    }

    // Two passes: measure and hash, then write. The header carries every
    // segment's length and digest, so a reader knows what it is getting before
    // it has it — which is what makes the file self-describing rather than a
    // tar that has to be unpacked to be understood.
    let mut segments = Vec::with_capacity(planned.len());
    for (path, name, from) in &planned {
        let (len, sha256) = measure(path, *from, if name.ends_with(PACK) { Some(pack_len) } else { None })?;
        segments.push(Segment {
            name: name.clone(),
            from_offset: *from,
            len,
            sha256,
        });
    }

    let mut w = HashingWriter::new(out);

    w.write_all(MAGIC)?;
    w.write_all(&FORMAT_VERSION.to_le_bytes())?;
    w.write_all(&0u32.to_le_bytes())?;
    w.write_all(&from_generation.to_le_bytes())?;
    w.write_all(&pack_len.to_le_bytes())?;
    w.write_all(&base_prefix_sha256)?;
    w.write_all(&(segments.len() as u32).to_le_bytes())?;
    w.write_all(&0u32.to_le_bytes())?;
    for s in &segments {
        let nb = s.name.as_bytes();
        w.write_all(&(nb.len() as u16).to_le_bytes())?;
        w.write_all(nb)?;
        w.write_all(&s.from_offset.to_le_bytes())?;
        w.write_all(&s.len.to_le_bytes())?;
        w.write_all(&s.sha256)?;
    }
    let mut bytes_out = 0u64;
    for ((path, _, from), s) in planned.iter().zip(&segments) {
        copy_range(path, *from, s.len, &mut w)?;
        bytes_out += s.len;
    }
    let trailer = w.digest();
    let inner = w.into_inner();
    inner.write_all(&trailer)?;

    Ok(ArchiveOutReceipt {
        from_generation,
        to_generation: pack_len,
        bytes_out,
        segments,
        file_sha256: trailer,
    })
}

/// Read an archive's header without reading its payload.
pub fn read_header(path: &Path) -> Result<ArchiveHeader> {
    let mut f = File::open(path).with_context(|| format!("opening {}", path.display()))?;
    let mut fixed = [0u8; 72];
    f.read_exact(&mut fixed)
        .with_context(|| format!("{} is too short to be an archive", path.display()))?;
    if &fixed[0..8] != MAGIC {
        bail!(
            "{} does not start with {} — it is not a gunnar repository archive",
            path.display(),
            String::from_utf8_lossy(MAGIC)
        );
    }
    let version = u32::from_le_bytes(fixed[8..12].try_into().unwrap());
    if version != FORMAT_VERSION {
        bail!(
            "{} is format version {version}; this build reads {FORMAT_VERSION} and will not guess \
             at the difference",
            path.display()
        );
    }
    let flags = u32::from_le_bytes(fixed[12..16].try_into().unwrap());
    let from_generation = u64::from_le_bytes(fixed[16..24].try_into().unwrap());
    let to_generation = u64::from_le_bytes(fixed[24..32].try_into().unwrap());
    let mut base_prefix_sha256 = [0u8; 32];
    base_prefix_sha256.copy_from_slice(&fixed[32..64]);
    let count = u32::from_le_bytes(fixed[64..68].try_into().unwrap());
    let mut segments = Vec::with_capacity(count as usize);
    let mut at = 72u64;
    for _ in 0..count {
        let mut nl = [0u8; 2];
        f.read_exact(&mut nl)?;
        let nl = u16::from_le_bytes(nl) as usize;
        let mut nb = vec![0u8; nl];
        f.read_exact(&mut nb)?;
        let name = String::from_utf8(nb).context("a segment name that is not UTF-8")?;
        if name.contains("..") || name.starts_with('/') {
            bail!("refusing the segment name {name:?}: an archive may not write outside the repository directory");
        }
        let mut o = [0u8; 8];
        f.read_exact(&mut o)?;
        let mut l = [0u8; 8];
        f.read_exact(&mut l)?;
        let mut d = [0u8; 32];
        f.read_exact(&mut d)?;
        at += 2 + nl as u64 + 8 + 8 + 32;
        segments.push(Segment {
            name,
            from_offset: u64::from_le_bytes(o),
            len: u64::from_le_bytes(l),
            sha256: d,
        });
    }
    Ok(ArchiveHeader {
        version,
        flags,
        from_generation,
        to_generation,
        base_prefix_sha256,
        segments,
        payload_start: at,
    })
}

/// **Put an archive back into `repo_dir`.**
///
/// A full archive (`from_generation == 0`) needs an empty or absent directory:
/// restoring over a store that already has bytes is how a restore silently
/// half-happens. An incremental archive is *applied* to a directory whose
/// `objects.pack` is already exactly `from_generation` bytes long and whose
/// base fingerprint matches; anything else is refused by name.
///
/// **No index is written and none is expected.** The next
/// [`crate::git_ops::GitStore::open`] on this directory reads the journal, sees
/// index rows for none of it, and re-queues every pack — which is the same path
/// a crash mid-index takes, and it is the reason nothing derived travels.
pub fn archive_in(file: &Path, repo_dir: &Path) -> Result<ArchiveInReceipt> {
    let header = read_header(file)?;
    let mut f = File::open(file)?;

    // Verify the trailer before writing one byte into the store: an archive
    // that is short or corrupt must not become a half-restored repository.
    let file_len = f.metadata()?.len();
    if file_len < header.payload_start + 32 {
        bail!("{} is truncated: no room for its payload and trailer", file.display());
    }
    let mut hasher = Sha256::new();
    f.seek(SeekFrom::Start(0))?;
    let mut left = file_len - 32;
    let mut buf = vec![0u8; COPY_BUF];
    while left > 0 {
        let n = (left as usize).min(buf.len());
        f.read_exact(&mut buf[..n])?;
        hasher.update(&buf[..n]);
        left -= n as u64;
    }
    let want: [u8; 32] = hasher.finalize().into();
    let mut got = [0u8; 32];
    f.read_exact(&mut got)?;
    if want != got {
        bail!(
            "{} fails its own trailer digest — the file is corrupt or truncated and NOTHING has \
             been written",
            file.display()
        );
    }

    let incremental = header.from_generation > 0;
    if incremental {
        let store = find_store_dir(repo_dir).with_context(|| {
            format!(
                "an incremental archive is applied to an existing store, and {} is not one",
                repo_dir.display()
            )
        })?;
        let have = std::fs::metadata(store.join(PACK))?.len();
        if have != header.from_generation {
            bail!(
                "this archive starts at watermark {} but {}'s pack is {have} bytes: restore the \
                 base first, or take a full archive",
                header.from_generation,
                store.display()
            );
        }
        if base_prefix(&store)? != header.base_prefix_sha256 {
            bail!(
                "the base has been rewritten: {}'s pack prefix does not match the one this \
                 archive was cut against (a gc compacts into a new generation). Take a full \
                 archive with --from-generation 0.",
                store.display()
            );
        }
    } else if repo_dir.exists() {
        let empty = std::fs::read_dir(repo_dir)?.next().is_none();
        if !empty {
            bail!(
                "{} is not empty; a full restore refuses to write over an existing repository",
                repo_dir.display()
            );
        }
    }
    std::fs::create_dir_all(repo_dir)
        .with_context(|| format!("creating {}", repo_dir.display()))?;

    f.seek(SeekFrom::Start(header.payload_start))?;
    let mut bytes_in = 0u64;
    let mut restored = Vec::new();
    for s in &header.segments {
        let dest = repo_dir.join(&s.name);
        if let Some(parent) = dest.parent() {
            std::fs::create_dir_all(parent)?;
        }
        let mut out = if s.from_offset == 0 {
            File::create(&dest).with_context(|| format!("creating {}", dest.display()))?
        } else {
            let mut o = std::fs::OpenOptions::new()
                .write(true)
                .open(&dest)
                .with_context(|| format!("opening {} to append a watermarked segment", dest.display()))?;
            o.seek(SeekFrom::Start(s.from_offset))?;
            o
        };
        let mut left = s.len;
        let mut h = Sha256::new();
        while left > 0 {
            let n = (left as usize).min(buf.len());
            f.read_exact(&mut buf[..n])?;
            h.update(&buf[..n]);
            out.write_all(&buf[..n])?;
            left -= n as u64;
        }
        let d: [u8; 32] = h.finalize().into();
        if d != s.sha256 {
            bail!("segment {} does not match its digest", s.name);
        }
        out.sync_all()
            .with_context(|| format!("fsyncing {}", dest.display()))?;
        bytes_in += s.len;
        restored.push(s.name.clone());
    }
    Ok(ArchiveInReceipt {
        repo_dir: repo_dir.to_path_buf(),
        to_generation: header.to_generation,
        bytes_in,
        restored,
    })
}

// ── plumbing ────────────────────────────────────────────────────────────────

/// Length and digest of `path[from..cap]`. `cap` pins the pack at the length
/// the cut was taken at, so a push landing mid-archive cannot lengthen the
/// segment out from under the header that already described it.
fn measure(path: &Path, from: u64, cap: Option<u64>) -> Result<(u64, [u8; 32])> {
    let mut f = File::open(path).with_context(|| format!("opening {}", path.display()))?;
    let end = cap.unwrap_or(f.metadata()?.len());
    let len = end.saturating_sub(from);
    f.seek(SeekFrom::Start(from))?;
    let mut h = Sha256::new();
    let mut buf = vec![0u8; COPY_BUF];
    let mut left = len;
    while left > 0 {
        let n = (left as usize).min(buf.len());
        f.read_exact(&mut buf[..n])
            .with_context(|| format!("reading {} at {from}", path.display()))?;
        h.update(&buf[..n]);
        left -= n as u64;
    }
    Ok((len, h.finalize().into()))
}

fn copy_range<W: Write>(path: &Path, from: u64, len: u64, w: &mut W) -> Result<()> {
    let mut f = File::open(path)?;
    f.seek(SeekFrom::Start(from))?;
    let mut buf = vec![0u8; COPY_BUF];
    let mut left = len;
    while left > 0 {
        let n = (left as usize).min(buf.len());
        f.read_exact(&mut buf[..n])?;
        w.write_all(&buf[..n])?;
        left -= n as u64;
    }
    Ok(())
}

/// Writes through and hashes everything that passes. The trailer is this
/// digest, so it covers the header as well as the payload.
struct HashingWriter<W: Write> {
    inner: W,
    hasher: Sha256,
}

impl<W: Write> HashingWriter<W> {
    fn new(inner: W) -> Self {
        Self {
            inner,
            hasher: Sha256::new(),
        }
    }
    fn digest(&self) -> [u8; 32] {
        self.hasher.clone().finalize().into()
    }
    fn into_inner(self) -> W {
        self.inner
    }
}

impl<W: Write> Write for HashingWriter<W> {
    fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
        let n = self.inner.write(buf)?;
        self.hasher.update(&buf[..n]);
        Ok(n)
    }
    fn flush(&mut self) -> std::io::Result<()> {
        self.inner.flush()
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    /// A store directory with the five files a real one has, three of them
    /// truth and two of them derivations.
    fn fixture(root: &Path, pack: &[u8]) -> PathBuf {
        let repo = root.join("r.git");
        let store = repo.join("gunnar-git.znippy");
        std::fs::create_dir_all(&store).unwrap();
        std::fs::write(repo.join(OBJECT_FORMAT), b"sha1").unwrap();
        std::fs::write(store.join(PACK), pack).unwrap();
        std::fs::write(store.join(JOURNAL), b"journal rows").unwrap();
        std::fs::write(store.join(REFS), b"arrow ref frames").unwrap();
        std::fs::write(store.join("objects.tail"), vec![0u8; 4096]).unwrap();
        std::fs::write(store.join("objects.exploded"), vec![1u8; 2048]).unwrap();
        repo
    }

    fn tmp(name: &str) -> PathBuf {
        let d = std::env::temp_dir().join(format!("laneAB-{name}-{}", std::process::id()));
        let _ = std::fs::remove_dir_all(&d);
        std::fs::create_dir_all(&d).unwrap();
        d
    }

    /// **The derivations do not travel.** This is the property the whole module
    /// exists for, so it is asserted rather than described.
    #[test]
    fn no_index_travels() {
        let d = tmp("noindex");
        let repo = fixture(&d, b"PACK-bytes-0123456789");
        let out = d.join("a.gaf");
        let r = archive_out(&repo, &out, 0).unwrap();
        let names: Vec<&str> = r.segments.iter().map(|s| s.name.as_str()).collect();
        assert!(names.iter().any(|n| n.ends_with(PACK)));
        assert!(names.iter().any(|n| n.ends_with(JOURNAL)));
        assert!(names.iter().any(|n| n.ends_with(REFS)));
        assert!(
            !names.iter().any(|n| n.contains("objects.tail")),
            "the redb index travelled: {names:?}"
        );
        assert!(
            !names.iter().any(|n| n.contains("objects.exploded")),
            "the exploded cache travelled: {names:?}"
        );
        let _ = std::fs::remove_dir_all(&d);
    }

    /// **The streamed bytes ARE the file's bytes.** `archive_out_to` is what a
    /// gRPC `Repos.ArchiveOut` writes down a socket, and the only thing that
    /// makes an off-box archive worth anything is that `archive_in` cannot tell
    /// the two apart. Asserted on the bytes and on the receipt, not on a
    /// round-trip of the receipt through itself.
    ///
    /// Seen red: with `archive_out_to` writing the trailer through the hashing
    /// writer instead of past it, `file_sha256` differs and this fails.
    #[test]
    fn a_streamed_archive_is_the_same_bytes_as_the_file() {
        let d = tmp("streamed");
        let pack: Vec<u8> = (0..300_000u32).map(|i| (i % 253) as u8).collect();
        let repo = fixture(&d, &pack);
        let out = d.join("f.gaf");
        let on_file = archive_out(&repo, &out, 0).unwrap();

        let mut streamed = Vec::new();
        let over_wire = archive_out_to(&repo, &mut streamed, 0).unwrap();

        assert_eq!(on_file, over_wire, "the receipts disagree");
        assert_eq!(
            streamed,
            std::fs::read(&out).unwrap(),
            "the streamed archive is not the file's bytes"
        );

        // …and it restores, which is the only claim that matters.
        let back = d.join("back");
        let r = archive_in(&out, &back).unwrap();
        assert_eq!(r.to_generation, pack.len() as u64);
        let wire = d.join("wire.gaf");
        std::fs::write(&wire, &streamed).unwrap();
        let back2 = d.join("back2");
        let r2 = archive_in(&wire, &back2).unwrap();
        assert_eq!(r.to_generation, r2.to_generation);
        let _ = std::fs::remove_dir_all(&d);
    }

    /// Full round trip, byte for byte.
    #[test]
    fn full_round_trip_is_byte_identical() {
        let d = tmp("full");
        let pack: Vec<u8> = (0..200_000u32).map(|i| (i % 251) as u8).collect();
        let repo = fixture(&d, &pack);
        let out = d.join("full.gaf");
        let r = archive_out(&repo, &out, 0).unwrap();
        assert_eq!(r.from_generation, 0);
        assert_eq!(r.to_generation, pack.len() as u64);

        let back = d.join("restored.git");
        let got = archive_in(&out, &back).unwrap();
        assert_eq!(got.to_generation, pack.len() as u64);
        let store = find_store_dir(&back).unwrap();
        assert_eq!(std::fs::read(store.join(PACK)).unwrap(), pack);
        assert_eq!(std::fs::read(store.join(JOURNAL)).unwrap(), b"journal rows");
        assert_eq!(std::fs::read(store.join(REFS)).unwrap(), b"arrow ref frames");
        assert_eq!(std::fs::read(back.join(OBJECT_FORMAT)).unwrap(), b"sha1");
        assert!(
            !store.join("objects.tail").exists(),
            "a restore must not fabricate an index"
        );
        let _ = std::fs::remove_dir_all(&d);
    }

    /// **The watermark is a watermark.** Append to the pack, cut from the
    /// previous `to_generation`, and only the new bytes move.
    #[test]
    fn incremental_moves_only_the_new_bytes() {
        let d = tmp("incr");
        let base: Vec<u8> = (0..100_000u32).map(|i| (i % 97) as u8).collect();
        let repo = fixture(&d, &base);
        let a = d.join("base.gaf");
        let r1 = archive_out(&repo, &a, 0).unwrap();

        let store = find_store_dir(&repo).unwrap();
        let extra: Vec<u8> = (0..30_000u32).map(|i| (i % 13) as u8).collect();
        {
            use std::io::Write as _;
            let mut f = std::fs::OpenOptions::new()
                .append(true)
                .open(store.join(PACK))
                .unwrap();
            f.write_all(&extra).unwrap();
        }
        let b = d.join("incr.gaf");
        let r2 = archive_out(&repo, &b, r1.to_generation).unwrap();
        let pack_seg = r2
            .segments
            .iter()
            .find(|s| s.name.ends_with(PACK))
            .unwrap();
        assert_eq!(
            pack_seg.len,
            extra.len() as u64,
            "an incremental cut must move only the appended bytes"
        );

        let back = d.join("restored.git");
        archive_in(&a, &back).unwrap();
        archive_in(&b, &back).unwrap();
        let rstore = find_store_dir(&back).unwrap();
        let mut want = base.clone();
        want.extend_from_slice(&extra);
        assert_eq!(std::fs::read(rstore.join(PACK)).unwrap(), want);
        let _ = std::fs::remove_dir_all(&d);
    }

    /// **A rewritten base is refused by name, not restored into a mixture.**
    #[test]
    fn a_rewritten_base_is_refused() {
        let d = tmp("gc");
        let base: Vec<u8> = (0..80_000u32).map(|i| (i % 89) as u8).collect();
        let repo = fixture(&d, &base);
        let a = d.join("base.gaf");
        let r1 = archive_out(&repo, &a, 0).unwrap();
        let store = find_store_dir(&repo).unwrap();
        {
            use std::io::Write as _;
            let mut f = std::fs::OpenOptions::new()
                .append(true)
                .open(store.join(PACK))
                .unwrap();
            f.write_all(b"more").unwrap();
        }
        let b = d.join("incr.gaf");
        archive_out(&repo, &b, r1.to_generation).unwrap();

        // A gc: the base is compacted into different bytes at the same length.
        let back = d.join("restored.git");
        archive_in(&a, &back).unwrap();
        let rstore = find_store_dir(&back).unwrap();
        let mut rewritten: Vec<u8> = base.clone();
        rewritten[0] ^= 0xff;
        std::fs::write(rstore.join(PACK), &rewritten).unwrap();
        let err = archive_in(&b, &back).expect_err("a rewritten base must be refused");
        assert!(
            format!("{err:#}").contains("the base has been rewritten"),
            "refused, but not by name: {err:#}"
        );
        let _ = std::fs::remove_dir_all(&d);
    }

    /// **A corrupt file writes NOTHING.** Seen RED by flipping the assertion:
    /// without the up-front trailer check the segments land and then the digest
    /// fails, leaving a half-restored repository.
    #[test]
    fn a_corrupt_archive_restores_nothing() {
        let d = tmp("corrupt");
        let repo = fixture(&d, b"PACK-abcdefghijklmnop");
        let out = d.join("a.gaf");
        archive_out(&repo, &out, 0).unwrap();
        let mut bytes = std::fs::read(&out).unwrap();
        let n = bytes.len();
        bytes[n - 40] ^= 0xff; // inside the payload, above the trailer
        std::fs::write(&out, &bytes).unwrap();
        let back = d.join("restored.git");
        let err = archive_in(&out, &back).expect_err("a corrupt archive must be refused");
        assert!(format!("{err:#}").contains("trailer digest"), "{err:#}");
        assert!(
            !back.exists() || std::fs::read_dir(&back).unwrap().next().is_none(),
            "a refused restore left bytes behind"
        );
        let _ = std::fs::remove_dir_all(&d);
    }

    /// A full restore never writes over a repository that already has bytes.
    #[test]
    fn a_full_restore_refuses_a_non_empty_directory() {
        let d = tmp("nonempty");
        let repo = fixture(&d, b"PACK-xyz");
        let out = d.join("a.gaf");
        archive_out(&repo, &out, 0).unwrap();
        let back = d.join("restored.git");
        std::fs::create_dir_all(&back).unwrap();
        std::fs::write(back.join("something"), b"in the way").unwrap();
        let err = archive_in(&out, &back).expect_err("must refuse");
        assert!(format!("{err:#}").contains("not empty"), "{err:#}");
        let _ = std::fs::remove_dir_all(&d);
    }

    /// A segment name may not escape the repository directory.
    #[test]
    fn a_segment_name_cannot_escape() {
        let d = tmp("escape");
        let repo = fixture(&d, b"PACK-1");
        let out = d.join("a.gaf");
        archive_out(&repo, &out, 0).unwrap();
        let mut bytes = std::fs::read(&out).unwrap();
        // The first segment is OBJECT_FORMAT (13 bytes); overwrite it with the
        // same-length traversal name.
        let at = 72 + 2;
        let evil = b"../../escaped";
        assert_eq!(evil.len(), OBJECT_FORMAT.len());
        bytes[at..at + evil.len()].copy_from_slice(evil);
        std::fs::write(&out, &bytes).unwrap();
        let err = read_header(&out).expect_err("traversal must be refused");
        assert!(format!("{err:#}").contains("outside the repository"), "{err:#}");
        let _ = std::fs::remove_dir_all(&d);
    }
}