crates/oxy-core/src/store.rs at 30d5fac7560d · andymccall/oxygen

# 30d5fac7560d oxygen / crates / oxy-core / src / store.rs
92.6 KiB · Rust · last changed The object store runs on any Store: a directory today, a bucket next 1 month ago source blame history raw
//! Object store (spec §13.1, §13.3, §13.4): loose objects, one per key
//! under `loose/<hh>/<rest>`, and packs, `pack-<id>.pack` with a sorted
//! `pack-<id>.idx`, holding zstd-compressed objects. Reads check loose
//! objects first, then every pack; every read re-hashes.
//!
//! Everything is kept on a [`crate::strata::Store`] (docs/stores.md): a
//! directory, which is what `.oxy/objects` is, or a bucket, which is how a
//! repository lives in its owner's region with the hub as one client of
//! it. The keys are the same wherever the store is, so a bucket holds
//! exactly what a directory would, and a directory store is as fast as
//! the files were before there was a trait: a loose object is one file, a
//! packed one is one ranged read of the pack. What must be local, the
//! pack being streamed out and the derived `metas` cache, lives in a
//! scratch directory.
//!
//! Pack file: `OXYPACK1` then entries of
//! `type u8 | flags u8 (bit 0: zstd) | stored_len u32 | raw_len u32 | ref 32 | body`.
//! Index file: `OXYIDX01` then `count u32` then sorted entries of
//! `ref 32 | offset u64 | stored_len u32 | raw_len u32 | type u8 | flags u8`.
//!
//! Beside the objects: `nocompress/<hex>` marks a chunk packs keep raw;
//! `sig/<hh>/<rest>/<key>` holds one signature over an object, written
//! once per signer; `seal/<hh>/<rest>/<attest>` and
//! `receipt/<hh>/<rest>/<attest>` index the seals covering a change and
//! the receipts on a state. Older stores hold these as one list per
//! object (`sigs/`, `seals/`, `receipts/`), which are still read.

use crate::attest::Attest;
use crate::hash::{hash_object, header, ObjectRef, ObjectType};
use crate::identity::{Identity, Signature};
use crate::objects::{Blob, Change, ObjectError, StateObject};
use crate::receipt::Receipt;
use crate::strata::{self, Error as BackendError, LocalStore, MemoryStore};
use rand::RngCore;
use std::cell::RefCell;
use std::fs;
use std::io::{self, Write};
use std::path::{Path, PathBuf};
use std::sync::Arc;

const PACK_MAGIC: &[u8; 8] = b"OXYPACK1";
const IDX_MAGIC: &[u8; 8] = b"OXYIDX01";
const IDX_ENTRY: usize = 32 + 8 + 4 + 4 + 1 + 1;
const FLAG_ZSTD: u8 = 1;
/// Compressed with the pack's trained dictionary (`pack-ID.dict`), which a
/// full repack makes from a sample of the objects it packs: small files
/// share most of their vocabulary, which each one alone cannot use.
const FLAG_DICT: u8 = 2;
/// The size of a pack's dictionary, and the least a repack must pack for
/// one to pay.
const DICT_BYTES: usize = 110 << 10;
const DICT_MIN_OBJECTS: usize = 1000;
/// Objects up to this size are compressed with the dictionary.
const DICT_OBJECT_MAX: usize = 64 << 10;
const ZSTD_LEVEL: i32 = 3;
/// The level of a full repack (`oxy gc`): slower, once, for a smaller store.
pub const REPACK_ZSTD_LEVEL: i32 = 9;
/// Largest object a pack entry may claim; anything bigger is corruption,
/// and refusing it keeps a bad index from driving huge allocations.
const MAX_OBJECT_LEN: u32 = 1 << 30;

/// An object read from the store: its type and body.
pub type Object = (ObjectType, Vec<u8>);

/// Where a long operation reports what it is doing (oxygen#66): a phase,
/// or a count within one, as a line for a spinner or a log.
pub type Progress<'a> = &'a mut dyn FnMut(&str);

/// The storage under a store: objects by key, see [`crate::strata`].
pub type Backend = dyn strata::Store;

fn silent() -> impl FnMut(&str) {
    |_| {}
}

/// A backend error as a store error: what is not found or already there
/// is handled where it matters; the rest is an I/O failure.
fn backend_err(e: BackendError) -> StoreError {
    match e {
        BackendError::Io(e) => StoreError::Io(e),
        other => StoreError::Io(io::Error::other(other.to_string())),
    }
}

/// `put_new` where an object already there is what we wanted.
fn put_new_or_existing(backend: &Backend, key: &str, bytes: &[u8]) -> Result<()> {
    match backend.put_new(key, bytes) {
        Ok(()) | Err(BackendError::Exists(_)) => Ok(()),
        Err(e) => Err(backend_err(e)),
    }
}

/// `get` as an option: `None` when there is no such key.
fn get_optional(backend: &Backend, key: &str) -> Result<Option<Vec<u8>>> {
    match backend.get(key) {
        Ok(b) => Ok(Some(b)),
        Err(BackendError::NotFound(_)) => Ok(None),
        Err(e) => Err(backend_err(e)),
    }
}

fn delete(backend: &Backend, key: &str) -> Result<()> {
    backend.delete(key).map_err(backend_err)
}

/// `loose/<hh>/<rest>`.
fn loose_key(r: &ObjectRef) -> String {
    let h = r.to_hex();
    format!("loose/{}/{}", &h[..2], &h[2..])
}

/// The object a loose key names, if it is one.
fn loose_ref(key: &str) -> Option<ObjectRef> {
    let rest = key.strip_prefix("loose/")?;
    let (hh, tail) = rest.split_once('/')?;
    if hh.len() != 2 || tail.contains('/') {
        return None;
    }
    ObjectRef::from_hex(&format!("{hh}{tail}"))
}

/// `<index>/<hh>/<rest>/` for the entries kept under an object.
fn fanout_prefix(index: &str, r: &ObjectRef) -> String {
    let h = r.to_hex();
    format!("{index}/{}/{}/", &h[..2], &h[2..])
}

/// `<index>/<hh>/<rest>`: the one-file-per-object layout of older stores.
fn legacy_key(index: &str, r: &ObjectRef) -> String {
    let h = r.to_hex();
    format!("{index}/{}/{}", &h[..2], &h[2..])
}

fn now_ms() -> u64 {
    std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map(|d| d.as_millis() as u64)
        .unwrap_or(0)
}

#[derive(Clone, Copy, Debug)]
struct PackEntry {
    offset: u64,
    stored_len: u32,
    raw_len: u32,
    ty: ObjectType,
    flags: u8,
}

/// One pack's index, fully in memory and sorted by ObjectRef.
struct PackIndex {
    /// The pack's key, `pack-<id>.pack`.
    key: String,
    refs: Vec<ObjectRef>,
    entries: Vec<PackEntry>,
    /// The pack's dictionary, prepared once, when it has one.
    dict: Option<Arc<zstd::dict::DecoderDictionary<'static>>>,
}

impl std::fmt::Debug for PackIndex {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("PackIndex")
            .field("key", &self.key)
            .field("objects", &self.refs.len())
            .field("dict", &self.dict.is_some())
            .finish()
    }
}

/// Decode a stored pack entry and check it against its reference.
fn decode_entry(
    r: &ObjectRef,
    e: PackEntry,
    stored: Vec<u8>,
    dict: Option<&Arc<zstd::dict::DecoderDictionary<'static>>>,
) -> Result<Object> {
    let body = if e.flags & FLAG_DICT != 0 {
        let dict = dict.ok_or_else(|| {
            StoreError::Corrupt(*r, "packed with a dictionary the pack lacks".into())
        })?;
        zstd::bulk::Decompressor::with_prepared_dictionary(dict)
            .and_then(|mut d| d.decompress(&stored, e.raw_len as usize))
            .map_err(|err| StoreError::Corrupt(*r, format!("zstd: {err}")))?
    } else if e.flags & FLAG_ZSTD != 0 {
        zstd::bulk::decompress(&stored, e.raw_len as usize)
            .map_err(|err| StoreError::Corrupt(*r, format!("zstd: {err}")))?
    } else {
        stored
    };
    if body.len() != e.raw_len as usize || hash_object(e.ty, &body) != *r {
        return Err(StoreError::Corrupt(
            *r,
            "packed object does not match its reference".into(),
        ));
    }
    Ok((e.ty, body))
}

impl PackIndex {
    fn load(backend: &Backend, idx_key: &str, pack_key: &str) -> Result<PackIndex> {
        let data = backend.get(idx_key).map_err(backend_err)?;
        let bad = |m: &str| StoreError::Corrupt(ObjectRef::ZERO, format!("{idx_key}: {m}"));
        if data.len() < 12 || &data[..8] != IDX_MAGIC {
            return Err(bad("bad index header"));
        }
        let count = u32::from_le_bytes(data[8..12].try_into().unwrap()) as usize;
        if data.len() != 12 + count * IDX_ENTRY {
            return Err(bad("index length mismatch"));
        }
        let pack_len = backend
            .stat(pack_key)
            .map_err(backend_err)?
            .ok_or_else(|| bad("the index names a pack that is not there"))?
            .size;
        let mut refs = Vec::with_capacity(count);
        let mut entries = Vec::with_capacity(count);
        for i in 0..count {
            let e = &data[12 + i * IDX_ENTRY..12 + (i + 1) * IDX_ENTRY];
            let r = ObjectRef::from_slice(&e[..32]).unwrap();
            if let Some(prev) = refs.last() {
                if r <= *prev {
                    return Err(bad("index not sorted"));
                }
            }
            refs.push(r);
            let entry = PackEntry {
                offset: u64::from_le_bytes(e[32..40].try_into().unwrap()),
                stored_len: u32::from_le_bytes(e[40..44].try_into().unwrap()),
                raw_len: u32::from_le_bytes(e[44..48].try_into().unwrap()),
                ty: ObjectType::ALL
                    .get(e[48] as usize)
                    .copied()
                    .ok_or_else(|| bad("unknown object type"))?,
                flags: e[49],
            };
            if entry.stored_len > MAX_OBJECT_LEN || entry.raw_len > MAX_OBJECT_LEN {
                return Err(StoreError::Corrupt(
                    r,
                    format!(
                        "{idx_key}: the index claims an object of {} bytes ({} stored), more than the {MAX_OBJECT_LEN} an object may be",
                        entry.raw_len, entry.stored_len
                    ),
                ));
            }
            if entry.offset.saturating_add(entry.stored_len as u64) > pack_len {
                return Err(bad("index entry points outside the pack"));
            }
            entries.push(entry);
        }
        let dict_key = format!("{}.dict", pack_key.trim_end_matches(".pack"));
        let dict = get_optional(backend, &dict_key)?
            .map(|d| Arc::new(zstd::dict::DecoderDictionary::copy(&d)));
        Ok(PackIndex {
            key: pack_key.to_string(),
            refs,
            entries,
            dict,
        })
    }

    fn find(&self, r: &ObjectRef) -> Option<PackEntry> {
        self.refs.binary_search(r).ok().map(|i| self.entries[i])
    }

    /// Read one entry from the pack: one ranged read, then decode.
    fn fetch(&self, backend: &Backend, r: &ObjectRef, e: PackEntry) -> Result<Object> {
        let stored = backend
            .get_range(&self.key, e.offset..e.offset + e.stored_len as u64)
            .map_err(|err| match err {
                BackendError::NotFound(_) => StoreError::NotFound(*r),
                other => backend_err(other),
            })?;
        decode_entry(r, e, stored, self.dict.as_ref())
    }
}

fn type_index(ty: ObjectType) -> u8 {
    ObjectType::ALL.iter().position(|t| *t == ty).unwrap() as u8
}

/// What `pack_loose` did.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct PackReport {
    pub objects: usize,
    pub raw_bytes: u64,
    pub packed_bytes: u64,
}

/// Packs tolerated before `merge_packs` folds them into one (spec §13.4).
pub const MAX_PACKS: usize = 16;

/// What `prune` did.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct PruneReport {
    pub objects: usize,
    pub bytes: u64,
    /// Unreachable objects left alone because they are younger than the grace period.
    pub deferred: usize,
}

#[derive(Debug, thiserror::Error)]
pub enum StoreError {
    #[error("io: {0}")]
    Io(#[from] io::Error),
    #[error("corrupt object {0}: {1}")]
    Corrupt(ObjectRef, String),
    #[error("object {0}")]
    Object(#[from] ObjectError),
    #[error("object {0} is a {1}, expected {2}")]
    WrongType(ObjectRef, &'static str, &'static str),
    #[error("object {0} not found")]
    NotFound(ObjectRef),
    #[error("a {kind} object of {size} bytes is larger than the {limit} bytes an object may be; nothing was stored. A change this large adds or touches too many files at once: commit it in parts (a directory at a time)")]
    TooLarge {
        kind: &'static str,
        size: usize,
        limit: u32,
    },
}

pub type Result<T> = std::result::Result<T, StoreError>;

pub struct Store {
    backend: Arc<Backend>,
    /// Temporary files (a pack being streamed) and derived caches (`metas`).
    /// For a store in a directory it is that directory.
    scratch: PathBuf,
    /// The backend is that directory, so the old file paths still hold.
    local: bool,
    packs: RefCell<Vec<PackIndex>>,
    /// An open write batch (spec §13.4): objects go here instead of to loose
    /// files, and are written out with one `fsync` when the batch finishes.
    pending: RefCell<Option<Pending>>,
    batch_depth: std::cell::Cell<u32>,
    /// Change metadata already read, from the `metas` cache file.
    metas: RefCell<Option<crate::fasthash::FastMap<ObjectRef, crate::state::ChangeMeta>>>,
}

impl std::fmt::Debug for Store {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("Store")
            .field("backend", &self.backend.describe())
            .field("scratch", &self.scratch)
            .field("packs", &self.packs.borrow().len())
            .finish()
    }
}

/// Objects written since `begin_batch`. Small batches stay in memory and
/// become loose objects; once `SPILL_OBJECTS` or `SPILL_BYTES` is exceeded
/// the batch streams into a pack file instead.
#[derive(Debug)]
struct Pending {
    buffered: Vec<(ObjectType, Vec<u8>)>,
    buffered_bytes: usize,
    lookup: std::collections::HashMap<ObjectRef, PendingEntry>,
    pack: Option<PendingPack>,
    report: PackReport,
}

#[derive(Debug, Clone, Copy)]
enum PendingEntry {
    Buffered(usize),
    Packed(PackEntry),
}

#[derive(Debug)]
struct PendingPack {
    id: String,
    tmp_path: PathBuf,
    writer: io::BufWriter<fs::File>,
    offset: u64,
    index: Vec<(ObjectRef, PackEntry)>,
}

const SPILL_OBJECTS: usize = 256;
const SPILL_BYTES: usize = 8 << 20;

/// Finishes the enclosing batch when dropped; call `finish` to see errors.
pub struct Batch<'a> {
    store: &'a Store,
    done: bool,
}

impl Batch<'_> {
    pub fn finish(mut self) -> Result<()> {
        self.done = true;
        self.store.finish_batch()
    }
}

impl Drop for Batch<'_> {
    fn drop(&mut self) {
        if !self.done {
            let _ = self.store.finish_batch();
        }
    }
}

/// Append one stored object to a pending pack and index it.
fn write_pack_entry(
    pack: &mut PendingPack,
    r: &ObjectRef,
    ty: ObjectType,
    raw_len: usize,
    flags: u8,
    stored: &[u8],
    report: &mut PackReport,
) -> Result<PackEntry> {
    let header_len = 1 + 1 + 4 + 4 + 32;
    pack.writer.write_all(&[type_index(ty), flags])?;
    pack.writer
        .write_all(&(stored.len() as u32).to_le_bytes())?;
    pack.writer.write_all(&(raw_len as u32).to_le_bytes())?;
    pack.writer.write_all(&r.0)?;
    pack.writer.write_all(stored)?;
    let entry = PackEntry {
        offset: pack.offset + header_len,
        stored_len: stored.len() as u32,
        raw_len: raw_len as u32,
        ty,
        flags,
    };
    pack.offset += header_len + stored.len() as u64;
    pack.index.push((*r, entry));
    report.objects += 1;
    report.raw_bytes += raw_len as u64;
    report.packed_bytes += stored.len() as u64;
    Ok(entry)
}

/// A random name for a temporary file or a pack.
fn fresh_id() -> String {
    let mut nonce = [0u8; 8];
    rand::rngs::OsRng.fill_bytes(&mut nonce);
    hex::encode(nonce)
}

/// Under `OXY_STORE_BACKEND=memory` every store opens on a store in
/// memory shared by directory, so the whole test suite can run without a
/// directory store, as it would over a bucket. Objects live for the
/// process.
fn memory_backend_for(dir: &Path) -> Option<Arc<Backend>> {
    use std::collections::HashMap;
    use std::sync::{Mutex, OnceLock};
    if std::env::var_os("OXY_STORE_BACKEND").as_deref() != Some(std::ffi::OsStr::new("memory")) {
        return None;
    }
    static STORES: OnceLock<Mutex<HashMap<PathBuf, Arc<MemoryStore>>>> = OnceLock::new();
    let key = dir.canonicalize().unwrap_or_else(|_| dir.to_path_buf());
    let mut map = STORES
        .get_or_init(Default::default)
        .lock()
        .unwrap_or_else(|p| p.into_inner());
    let store = map
        .entry(key)
        .or_insert_with(|| Arc::new(MemoryStore::new()))
        .clone();
    Some(store)
}

impl Store {
    /// Open the store in `dir`, which is also its scratch directory.
    pub fn open(dir: impl Into<PathBuf>) -> io::Result<Store> {
        let dir = dir.into();
        fs::create_dir_all(dir.join("loose"))?;
        if let Some(backend) = memory_backend_for(&dir) {
            return Store::open_on(backend, dir);
        }
        let backend: Arc<Backend> =
            Arc::new(LocalStore::open(&dir).map_err(|e| io::Error::other(e.to_string()))?);
        let mut store = Store::open_on(backend, dir)?;
        store.local = true;
        Ok(store)
    }

    /// Open the store kept on `backend`, with `scratch` for what must be
    /// on this disk: a pack while it is being written, and caches.
    pub fn open_on(backend: Arc<Backend>, scratch: impl Into<PathBuf>) -> io::Result<Store> {
        let scratch = scratch.into();
        fs::create_dir_all(&scratch)?;
        let store = Store {
            backend,
            scratch,
            local: false,
            packs: RefCell::new(Vec::new()),
            pending: RefCell::new(None),
            batch_depth: std::cell::Cell::new(0),
            metas: RefCell::new(None),
        };
        store
            .reload_packs()
            .map_err(|e| io::Error::other(e.to_string()))?;
        Ok(store)
    }

    /// The storage under this store.
    pub fn backend(&self) -> &Arc<Backend> {
        &self.backend
    }

    /// Whether the store is the directory `dir()` names, where every key
    /// is a file.
    pub fn is_local(&self) -> bool {
        self.local
    }

    fn reload_packs(&self) -> Result<()> {
        let listed = self.backend.list("pack-").map_err(backend_err)?;
        let mut idx_keys: Vec<&str> = listed
            .iter()
            .map(|e| e.key.as_str())
            .filter(|k| k.ends_with(".idx"))
            .collect();
        idx_keys.sort();
        let mut packs = Vec::new();
        for idx in idx_keys {
            let pack = format!("{}.pack", idx.trim_end_matches(".idx"));
            if listed.iter().any(|e| e.key == pack) {
                packs.push(PackIndex::load(self.backend.as_ref(), idx, &pack)?);
            }
        }
        *self.packs.borrow_mut() = packs;
        Ok(())
    }

    pub fn pack_count(&self) -> usize {
        self.packs.borrow().len()
    }

    fn find_packed(&self, r: &ObjectRef) -> Option<(usize, PackEntry)> {
        self.packs
            .borrow()
            .iter()
            .enumerate()
            .find_map(|(i, p)| p.find(r).map(|e| (i, e)))
    }

    /// Loose objects, with their keys and ages.
    fn loose_objects(&self) -> Result<Vec<(ObjectRef, strata::Entry)>> {
        let mut out: Vec<(ObjectRef, strata::Entry)> = self
            .backend
            .list("loose/")
            .map_err(backend_err)?
            .into_iter()
            .filter_map(|e| loose_ref(&e.key).map(|r| (r, e)))
            .collect();
        out.sort_by_key(|(r, _)| *r);
        Ok(out)
    }

    pub fn loose_count(&self) -> io::Result<usize> {
        self.loose_objects()
            .map(|l| l.len())
            .map_err(|e| io::Error::other(e.to_string()))
    }

    /// Move every loose object into a new pack (spec §13.4). Loose objects
    /// are deleted only after the pack and its index are durable.
    pub fn pack_loose(&self) -> Result<PackReport> {
        let loose = self.loose_objects()?;
        self.write_pack(
            &loose.iter().map(|(r, _)| *r).collect::<Vec<_>>(),
            &loose.iter().map(|(_, e)| e.key.clone()).collect::<Vec<_>>(),
            &[],
            ZSTD_LEVEL,
            &mut silent(),
        )
    }

    /// Fold every pack and loose object into one pack compressed at
    /// `level` (spec §13.4): what an explicit `oxy gc` does, as `git gc`
    /// repacks. Returns the report and how many packs were folded in.
    pub fn repack_all(&self, level: i32) -> Result<(PackReport, usize)> {
        self.repack_all_with(level, &mut silent())
    }

    /// [`Store::repack_all`], reporting progress.
    pub fn repack_all_with(&self, level: i32, progress: Progress) -> Result<(PackReport, usize)> {
        let loose = self.loose_objects()?;
        let (mut refs, keys): (Vec<ObjectRef>, Vec<String>) = {
            let packs = self.packs.borrow();
            (
                packs.iter().flat_map(|p| p.refs.iter().copied()).collect(),
                packs.iter().map(|p| p.key.clone()).collect(),
            )
        };
        if loose.is_empty() && keys.len() <= 1 && level == ZSTD_LEVEL {
            return Ok((PackReport::default(), 0));
        }
        refs.extend(loose.iter().map(|(r, _)| *r));
        let loose_keys: Vec<String> = loose.into_iter().map(|(_, e)| e.key).collect();
        let n = keys.len();
        let report = self.write_pack(&refs, &loose_keys, &keys, level, progress)?;
        Ok((report, n))
    }

    /// Fold every pack into one when there are more than `MAX_PACKS`
    /// (spec §13.4). Returns how many packs were merged.
    pub fn merge_packs(&self) -> Result<usize> {
        let (refs, keys): (Vec<ObjectRef>, Vec<String>) = {
            let packs = self.packs.borrow();
            if packs.len() <= MAX_PACKS {
                return Ok(0);
            }
            (
                packs.iter().flat_map(|p| p.refs.iter().copied()).collect(),
                packs.iter().map(|p| p.key.clone()).collect(),
            )
        };
        let n = keys.len();
        self.write_pack(&refs, &[], &keys, ZSTD_LEVEL, &mut silent())?;
        Ok(n)
    }

    /// Delete every object not in `keep` (spec §21). Loose objects and packs
    /// younger than `grace` are left alone, since a concurrent writer may not
    /// have recorded its references yet. Packs holding unreachable objects
    /// are rewritten without them. Signatures, seal and receipt entries of
    /// deleted objects go too.
    pub fn prune(
        &self,
        keep: &std::collections::HashSet<ObjectRef>,
        grace: std::time::Duration,
    ) -> Result<PruneReport> {
        self.prune_with(keep, grace, &mut silent())
    }

    /// [`Store::prune`], reporting progress.
    pub fn prune_with(
        &self,
        keep: &std::collections::HashSet<ObjectRef>,
        grace: std::time::Duration,
        progress: Progress,
    ) -> Result<PruneReport> {
        progress("pruning loose objects");
        let now = now_ms();
        let grace_ms = grace.as_millis() as u64;
        // An object whose age the store does not know is old enough.
        let old_enough =
            |modified: Option<u64>| modified.is_none_or(|t| now.saturating_sub(t) >= grace_ms);
        let mut report = PruneReport::default();
        // Loose objects.
        for (r, entry) in self.loose_objects()? {
            if keep.contains(&r) {
                continue;
            }
            if !old_enough(entry.modified_ms) {
                report.deferred += 1;
                continue;
            }
            report.bytes += entry.size;
            report.objects += 1;
            delete(self.backend.as_ref(), &entry.key)?;
            self.remove_sidecars(&r)?;
        }
        // Packs: rewrite those that hold unreachable objects.
        let mut keep_refs: Vec<ObjectRef> = Vec::new();
        let mut old_keys: Vec<String> = Vec::new();
        let mut rewrite = false;
        progress("pruning packs");
        for p in self.packs.borrow().iter() {
            let unreachable = p.refs.iter().filter(|r| !keep.contains(r)).count();
            if unreachable == 0 {
                continue;
            }
            let modified = self
                .backend
                .stat(&p.key)
                .map_err(backend_err)?
                .and_then(|e| e.modified_ms);
            if !old_enough(modified) {
                report.deferred += unreachable;
                continue;
            }
            rewrite = true;
            old_keys.push(p.key.clone());
            for (r, e) in p.refs.iter().zip(p.entries.iter()) {
                if keep.contains(r) {
                    keep_refs.push(*r);
                } else {
                    report.objects += 1;
                    report.bytes += e.stored_len as u64;
                }
            }
        }
        if rewrite {
            let dropped: Vec<ObjectRef> = {
                let packs = self.packs.borrow();
                packs
                    .iter()
                    .filter(|p| old_keys.contains(&p.key))
                    .flat_map(|p| p.refs.iter().copied())
                    .filter(|r| !keep.contains(r))
                    .collect()
            };
            self.write_pack(&keep_refs, &[], &old_keys, ZSTD_LEVEL, progress)?;
            for r in dropped {
                self.remove_sidecars(&r)?;
            }
        }
        Ok(report)
    }

    /// Everything kept beside an object: its signatures and the seal and
    /// receipt entries under it, in both layouts.
    fn remove_sidecars(&self, r: &ObjectRef) -> Result<()> {
        for index in ["sigs", "seals", "receipts"] {
            delete(self.backend.as_ref(), &legacy_key(index, r))?;
        }
        for index in ["sig", "seal", "receipt"] {
            for e in self
                .backend
                .list(&fanout_prefix(index, r))
                .map_err(backend_err)?
            {
                delete(self.backend.as_ref(), &e.key)?;
            }
        }
        Ok(())
    }

    /// Put a complete, synced temporary pack file and its index into the
    /// store; `index` must be sorted by ObjectRef. The pack goes first: an
    /// index never names a pack that is not there.
    fn finish_pack_files(
        &self,
        id: &str,
        tmp_pack: &Path,
        index: &[(ObjectRef, PackEntry)],
    ) -> Result<()> {
        let mut idx = Vec::with_capacity(12 + index.len() * IDX_ENTRY);
        idx.extend_from_slice(IDX_MAGIC);
        idx.extend_from_slice(&(index.len() as u32).to_le_bytes());
        for (r, e) in index {
            idx.extend_from_slice(&r.0);
            idx.extend_from_slice(&e.offset.to_le_bytes());
            idx.extend_from_slice(&e.stored_len.to_le_bytes());
            idx.extend_from_slice(&e.raw_len.to_le_bytes());
            idx.extend_from_slice(&[type_index(e.ty), e.flags]);
        }
        let put = self
            .backend
            .put_new_file(&format!("pack-{id}.pack"), tmp_pack)
            .map_err(backend_err);
        // A store on another disk copied it; the one in this directory
        // moved it and this finds nothing.
        let _ = fs::remove_file(tmp_pack);
        put?;
        self.backend
            .put_new(&format!("pack-{id}.idx"), &idx)
            .map_err(backend_err)?;
        self.reload_packs()
    }

    /// Remove a pack with its index and dictionary.
    fn remove_pack(&self, pack_key: &str) -> Result<()> {
        let stem = pack_key.trim_end_matches(".pack");
        delete(self.backend.as_ref(), pack_key)?;
        delete(self.backend.as_ref(), &format!("{stem}.idx"))?;
        delete(self.backend.as_ref(), &format!("{stem}.dict"))
    }

    /// Write `refs` into a new pack, then delete `loose_keys` and `old_packs`.
    fn write_pack(
        &self,
        refs: &[ObjectRef],
        loose_keys: &[String],
        old_packs: &[String],
        level: i32,
        progress: Progress,
    ) -> Result<PackReport> {
        if refs.is_empty() {
            for p in old_packs {
                self.remove_pack(p)?;
            }
            self.reload_packs()?;
            return Ok(PackReport::default());
        }
        let mut refs: Vec<ObjectRef> = refs.to_vec();
        refs.sort();
        refs.dedup();
        let id = fresh_id();
        let tmp_pack = self.scratch.join(format!(".tmp-{id}.pack"));
        let mut report = PackReport::default();
        let mut index: Vec<(ObjectRef, PackEntry)> = Vec::with_capacity(refs.len());
        // A full repack of enough objects trains a dictionary for them.
        let dict = if level == REPACK_ZSTD_LEVEL && refs.len() >= DICT_MIN_OBJECTS {
            let _t = crate::timing::phase("pack: train a dictionary");
            progress(&format!("training a dictionary on {} objects", refs.len()));
            self.train_dictionary(&refs)
        } else {
            None
        };
        let encoder = dict
            .as_ref()
            .map(|d| zstd::dict::EncoderDictionary::copy(d, level));
        let nocompress = self.nocompress_set()?;
        {
            let mut f = io::BufWriter::new(fs::File::create(&tmp_pack)?);
            f.write_all(PACK_MAGIC)?;
            let mut offset = PACK_MAGIC.len() as u64;
            // Read a batch (the store is not shared between threads),
            // compress it on every core, write it in order.
            let mut batches = refs.chunks(1024);
            let mut compressed_batch = Vec::new();
            let total = refs.len();
            let mut reported = usize::MAX;
            loop {
                if compressed_batch.is_empty() {
                    let Some(batch) = batches.next() else { break };
                    // Every tenth, and the first batch.
                    let tenth = report.objects * 10 / total.max(1);
                    if tenth != reported {
                        reported = tenth;
                        progress(&format!("packing {} of {total} objects", report.objects));
                    }
                    let mut read: Vec<(ObjectRef, ObjectType, Vec<u8>, bool)> =
                        Vec::with_capacity(batch.len());
                    for r in batch {
                        let (ty, body) = self.get(r)?;
                        read.push((*r, ty, body, nocompress.contains(r)));
                    }
                    use rayon::prelude::*;
                    compressed_batch = read
                        .into_par_iter()
                        .map_init(
                            || {
                                encoder.as_ref().and_then(|e| {
                                    zstd::bulk::Compressor::with_prepared_dictionary(e).ok()
                                })
                            },
                            |with_dict, (r, ty, body, raw)| {
                                let packed = if raw {
                                    None
                                } else {
                                    // Small objects with the dictionary;
                                    // a large one has context enough.
                                    let dicted = with_dict
                                        .as_mut()
                                        .filter(|_| body.len() <= DICT_OBJECT_MAX)
                                        .and_then(|c| c.compress(&body).ok())
                                        .map(|c| (FLAG_ZSTD | FLAG_DICT, c));
                                    dicted
                                        .or_else(|| {
                                            zstd::bulk::compress(&body, level)
                                                .ok()
                                                .map(|c| (FLAG_ZSTD, c))
                                        })
                                        .filter(|(_, c)| c.len() < body.len())
                                };
                                (r, ty, body, packed)
                            },
                        )
                        .collect::<Vec<_>>();
                    compressed_batch.reverse();
                }
                let Some((r, ty, body, packed)) = compressed_batch.pop() else {
                    continue;
                };
                let r = &r;
                let (flags, stored) = match packed {
                    Some(c) => c,
                    None => (0, body.clone()),
                };
                let header_len = 1 + 1 + 4 + 4 + 32;
                f.write_all(&[type_index(ty), flags])?;
                f.write_all(&(stored.len() as u32).to_le_bytes())?;
                f.write_all(&(body.len() as u32).to_le_bytes())?;
                f.write_all(&r.0)?;
                f.write_all(&stored)?;
                index.push((
                    *r,
                    PackEntry {
                        offset: offset + header_len,
                        stored_len: stored.len() as u32,
                        raw_len: body.len() as u32,
                        ty,
                        flags,
                    },
                ));
                offset += header_len + stored.len() as u64;
                report.objects += 1;
                report.raw_bytes += body.len() as u64;
                report.packed_bytes += stored.len() as u64;
            }
            f.flush()?;
            f.get_ref().sync_all()?;
        }
        // The dictionary before the pack that needs it becomes visible.
        if let Some(d) = &dict {
            self.backend
                .put_new(&format!("pack-{id}.dict"), d)
                .map_err(backend_err)?;
        }
        progress(&format!("writing the index of {} objects", index.len()));
        self.finish_pack_files(&id, &tmp_pack, &index)?;
        for p in old_packs {
            self.remove_pack(p)?;
        }
        self.reload_packs()?;
        for key in loose_keys {
            delete(self.backend.as_ref(), key)?;
        }
        Ok(report)
    }

    /// A dictionary for these objects, trained on a sample of them (at
    /// most 10 000, 32 MB); `None` when training fails.
    fn train_dictionary(&self, refs: &[ObjectRef]) -> Option<Vec<u8>> {
        let step = (refs.len() / 10_000).max(1);
        let mut samples: Vec<Vec<u8>> = Vec::new();
        let mut total = 0usize;
        for r in refs.iter().step_by(step) {
            if total > 32 << 20 {
                break;
            }
            let Ok((ty, body)) = self.get(r) else {
                continue;
            };
            if matches!(
                ty,
                ObjectType::Chunk | ObjectType::Blob | ObjectType::Change
            ) && !body.is_empty()
                && body.len() < 1 << 20
            {
                total += body.len();
                samples.push(body);
            }
        }
        zstd::dict::from_samples(&samples, DICT_BYTES).ok()
    }

    /// The scratch directory: for a store in a directory, that directory.
    pub fn dir(&self) -> &Path {
        &self.scratch
    }

    /// Where a loose object is kept, for a store in a directory (tests
    /// tamper with and remove objects there).
    #[cfg(test)]
    pub(crate) fn path_for(&self, r: &ObjectRef) -> PathBuf {
        self.scratch.join(loose_key(r))
    }

    /// Remove a loose object, whatever the store is on (tests make an
    /// object go missing).
    #[cfg(test)]
    pub(crate) fn remove_loose_for_test(&self, r: &ObjectRef) -> Result<()> {
        delete(self.backend.as_ref(), &loose_key(r))
    }

    pub fn has(&self, r: &ObjectRef) -> bool {
        self.backend.has(&loose_key(r)).unwrap_or(false)
            || self.find_packed(r).is_some()
            || self
                .pending
                .borrow()
                .as_ref()
                .is_some_and(|p| p.lookup.contains_key(r))
    }

    /// Start a write batch: until the matching `finish`, `put` collects
    /// objects and writes them out together, with one `fsync` for a large
    /// batch (a pack) instead of one per object. Batches nest; only the
    /// outermost does the work. Objects are readable while pending.
    pub fn begin_batch(&self) -> Result<Batch<'_>> {
        let depth = self.batch_depth.get();
        if depth == 0 {
            *self.pending.borrow_mut() = Some(Pending {
                buffered: vec![],
                buffered_bytes: 0,
                lookup: Default::default(),
                pack: None,
                report: PackReport::default(),
            });
        }
        self.batch_depth.set(depth + 1);
        Ok(Batch {
            store: self,
            done: false,
        })
    }

    fn finish_batch(&self) -> Result<()> {
        let depth = self.batch_depth.get();
        if depth == 0 {
            return Ok(());
        }
        self.batch_depth.set(depth - 1);
        if depth > 1 {
            return Ok(());
        }
        let Some(pending) = self.pending.borrow_mut().take() else {
            return Ok(());
        };
        self.write_pending(pending)
    }

    /// Make every object written so far durable, even inside an open
    /// batch, which stays open. Call before a reference that names them
    /// is moved (spec §13.6): the operation head, a bookmark pointer.
    pub fn flush(&self) -> Result<()> {
        let Some(pending) = self.pending.borrow_mut().take() else {
            return Ok(());
        };
        self.write_pending(pending)?;
        *self.pending.borrow_mut() = Some(Pending {
            buffered: vec![],
            buffered_bytes: 0,
            lookup: Default::default(),
            pack: None,
            report: PackReport::default(),
        });
        Ok(())
    }

    fn write_pending(&self, pending: Pending) -> Result<()> {
        match pending.pack {
            None => {
                let objects: Vec<(ObjectType, &[u8])> = pending
                    .buffered
                    .iter()
                    .map(|(ty, body)| (*ty, body.as_slice()))
                    .collect();
                self.put_loose_many(&objects)
            }
            Some(mut pack) => {
                pack.writer.flush()?;
                pack.writer.get_ref().sync_all()?;
                drop(pack.writer);
                pack.index.sort_by_key(|(r, _)| *r);
                self.finish_pack_files(&pack.id, &pack.tmp_path, &pack.index)?;
                Ok(())
            }
        }
    }

    /// Whether a batch is open and holds `r`.
    fn pending_get(&self, r: &ObjectRef) -> Option<Result<Object>> {
        let mut guard = self.pending.borrow_mut();
        let pending = guard.as_mut()?;
        let entry = *pending.lookup.get(r)?;
        Some(match entry {
            PendingEntry::Buffered(i) => {
                let (ty, body) = &pending.buffered[i];
                Ok((*ty, body.clone()))
            }
            PendingEntry::Packed(e) => {
                let Some(pack) = pending.pack.as_mut() else {
                    return Some(Err(StoreError::NotFound(*r)));
                };
                // Read from the pack file being written, on this disk.
                (|| {
                    pack.writer.flush()?;
                    use std::io::{Read, Seek, SeekFrom};
                    let mut f = fs::File::open(&pack.tmp_path)?;
                    f.seek(SeekFrom::Start(e.offset))?;
                    let mut stored = vec![0u8; e.stored_len as usize];
                    f.read_exact(&mut stored)?;
                    decode_entry(r, e, stored, None)
                })()
            }
        })
    }

    /// Append one object to the pending pack file.
    fn pending_pack_write(
        &self,
        pack: &mut PendingPack,
        r: &ObjectRef,
        ty: ObjectType,
        body: &[u8],
        report: &mut PackReport,
    ) -> Result<PackEntry> {
        let (flags, stored) = if self.is_nocompress(r) {
            (0, body.to_vec())
        } else {
            let compressed = zstd::bulk::compress(body, ZSTD_LEVEL)
                .map_err(|e| StoreError::Corrupt(*r, format!("zstd: {e}")))?;
            if compressed.len() < body.len() {
                (FLAG_ZSTD, compressed)
            } else {
                (0, body.to_vec())
            }
        };
        write_pack_entry(pack, r, ty, body.len(), flags, &stored, report)
    }

    fn put_pending(&self, r: ObjectRef, ty: ObjectType, body: &[u8]) -> Result<()> {
        let mut guard = self.pending.borrow_mut();
        let pending = guard.as_mut().expect("batch open");
        if pending.lookup.contains_key(&r) {
            return Ok(());
        }
        if pending.pack.is_none()
            && pending.buffered.len() < SPILL_OBJECTS
            && pending.buffered_bytes + body.len() <= SPILL_BYTES
        {
            pending.buffered.push((ty, body.to_vec()));
            pending.buffered_bytes += body.len();
            pending
                .lookup
                .insert(r, PendingEntry::Buffered(pending.buffered.len() - 1));
            return Ok(());
        }
        self.ensure_pack(pending)?;
        let mut report = std::mem::take(&mut pending.report);
        let pack = pending.pack.as_mut().unwrap();
        let e = self.pending_pack_write(pack, &r, ty, body, &mut report)?;
        pending.report = report;
        pending.lookup.insert(r, PendingEntry::Packed(e));
        Ok(())
    }

    /// Open the pending batch's pack, moving the objects buffered so far
    /// into it (a spill).
    fn ensure_pack(&self, pending: &mut Pending) -> Result<()> {
        if pending.pack.is_none() {
            // Spill: open the pack and move the buffered objects into it.
            let id = fresh_id();
            let tmp_path = self.scratch.join(format!(".tmp-{id}.pack"));
            let mut writer = io::BufWriter::new(fs::File::create(&tmp_path)?);
            writer.write_all(PACK_MAGIC)?;
            let mut pack = PendingPack {
                id,
                tmp_path,
                writer,
                offset: PACK_MAGIC.len() as u64,
                index: vec![],
            };
            let buffered = std::mem::take(&mut pending.buffered);
            pending.buffered_bytes = 0;
            let mut report = std::mem::take(&mut pending.report);
            for (bty, bbody) in &buffered {
                let br = hash_object(*bty, bbody);
                let e = self.pending_pack_write(&mut pack, &br, *bty, bbody, &mut report)?;
                pending.lookup.insert(br, PendingEntry::Packed(e));
            }
            pending.report = report;
            pending.pack = Some(pack);
        }
        Ok(())
    }

    /// Store many objects at once. Inside a large write batch they are
    /// compressed on every core and appended in order; otherwise each is
    /// stored as [`Store::put`] does.
    pub fn put_many(&self, objects: &[(ObjectType, &[u8])]) -> Result<Vec<ObjectRef>> {
        use rayon::prelude::*;
        let total: usize = objects.iter().map(|(_, b)| b.len()).sum();
        if self.pending.borrow().is_none() || total < SPILL_BYTES {
            return objects.iter().map(|(ty, b)| self.put(*ty, b)).collect();
        }
        for (ty, b) in objects {
            if b.len() > MAX_OBJECT_LEN as usize {
                return Err(StoreError::TooLarge {
                    kind: ty.name(),
                    size: b.len(),
                    limit: MAX_OBJECT_LEN,
                });
            }
        }
        let refs: Vec<ObjectRef> = objects
            .par_iter()
            .map(|(ty, b)| hash_object(*ty, b))
            .collect();
        // What is new: not stored already, not earlier in this call.
        let mut seen: std::collections::HashSet<ObjectRef> = std::collections::HashSet::new();
        let mut todo: Vec<usize> = Vec::new();
        for (i, r) in refs.iter().enumerate() {
            let known = self
                .pending
                .borrow()
                .as_ref()
                .is_some_and(|p| p.lookup.contains_key(r));
            if seen.insert(*r)
                && !known
                && self.find_packed(r).is_none()
                && !self.backend.has(&loose_key(r)).unwrap_or(false)
            {
                todo.push(i);
            }
        }
        let raw = self.nocompress_set()?;
        let packed: Vec<(u8, Vec<u8>)> = todo
            .par_iter()
            .map(|i| {
                let body = objects[*i].1;
                if raw.contains(&refs[*i]) {
                    return (0, body.to_vec());
                }
                match zstd::bulk::compress(body, ZSTD_LEVEL) {
                    Ok(c) if c.len() < body.len() => (FLAG_ZSTD, c),
                    _ => (0, body.to_vec()),
                }
            })
            .collect();
        let mut guard = self.pending.borrow_mut();
        let pending = guard.as_mut().expect("batch open");
        self.ensure_pack(pending)?;
        let mut report = std::mem::take(&mut pending.report);
        let pack = pending.pack.as_mut().unwrap();
        for (i, (flags, stored)) in todo.iter().zip(packed) {
            let (ty, body) = objects[*i];
            let e = write_pack_entry(pack, &refs[*i], ty, body.len(), flags, &stored, &mut report)?;
            pending.lookup.insert(refs[*i], PendingEntry::Packed(e));
        }
        pending.report = report;
        Ok(refs)
    }

    /// Store `body` as an object of type `ty`; idempotent.
    pub fn put(&self, ty: ObjectType, body: &[u8]) -> Result<ObjectRef> {
        // Refuse what the store could not read back (spec §13.2), before
        // anything is written: a pack must never hold an entry its index
        // would reject.
        if body.len() > MAX_OBJECT_LEN as usize {
            return Err(StoreError::TooLarge {
                kind: ty.name(),
                size: body.len(),
                limit: MAX_OBJECT_LEN,
            });
        }
        let r = hash_object(ty, body);
        if self.find_packed(&r).is_some() || self.backend.has(&loose_key(&r)).unwrap_or(false) {
            return Ok(r);
        }
        if self.pending.borrow().is_some() {
            self.put_pending(r, ty, body)?;
            return Ok(r);
        }
        self.put_loose(ty, body)?;
        Ok(r)
    }

    /// Write one loose object, durable when this returns.
    fn put_loose(&self, ty: ObjectType, body: &[u8]) -> Result<ObjectRef> {
        let r = hash_object(ty, body);
        let mut bytes = header(ty, body.len());
        bytes.extend_from_slice(body);
        crate::durable::step("write loose object")?;
        put_new_or_existing(self.backend.as_ref(), &loose_key(&r), &bytes)?;
        Ok(r)
    }

    /// Write several loose objects with as few waits on the disk as the
    /// guarantees allow (a directory store writes them all, then syncs
    /// each directory once). Durable when this returns, as `put_loose`
    /// one at a time would be (spec §13.6).
    fn put_loose_many(&self, objects: &[(ObjectType, &[u8])]) -> Result<()> {
        if objects.len() <= 1 {
            for (ty, body) in objects {
                self.put_loose(*ty, body)?;
            }
            return Ok(());
        }
        let mut seen = std::collections::HashSet::new();
        let items: Vec<(String, Vec<u8>)> = objects
            .iter()
            .filter_map(|(ty, body)| {
                let r = hash_object(*ty, body);
                seen.insert(r).then(|| {
                    let mut bytes = header(*ty, body.len());
                    bytes.extend_from_slice(body);
                    (loose_key(&r), bytes)
                })
            })
            .collect();
        self.backend.put_many_new(&items).map_err(backend_err)
    }

    /// Read and verify an object, loose, packed, or pending in a batch.
    pub fn get(&self, r: &ObjectRef) -> Result<Object> {
        match get_optional(self.backend.as_ref(), &loose_key(r))? {
            Some(bytes) => parse_loose(&bytes, r),
            None => match self.find_packed(r) {
                Some((i, e)) => self.packs.borrow()[i].fetch(self.backend.as_ref(), r, e),
                None => self.pending_get(r).unwrap_or(Err(StoreError::NotFound(*r))),
            },
        }
    }

    /// Many objects at once: loose ones are read and verified in parallel,
    /// then packed ones in parallel, then what is pending in a batch.
    /// Results are in the order of `refs`.
    pub fn get_many(&self, refs: &[ObjectRef]) -> Vec<Result<Object>> {
        use rayon::prelude::*;
        let backend = self.backend.as_ref();
        let mut found: Vec<Option<Result<Object>>> = refs
            .par_iter()
            .map(|r| match get_optional(backend, &loose_key(r)) {
                Ok(Some(bytes)) => Some(parse_loose(&bytes, r)),
                Ok(None) => None,
                Err(e) => Some(Err(e)),
            })
            .collect();
        // Packed ones: look them up in order, then read them in parallel.
        let packs = self.packs.borrow();
        let packed: Vec<(usize, usize, PackEntry)> = found
            .iter()
            .enumerate()
            .filter(|(_, f)| f.is_none())
            .filter_map(|(i, _)| {
                packs
                    .iter()
                    .enumerate()
                    .find_map(|(p, pack)| pack.find(&refs[i]).map(|e| (i, p, e)))
            })
            .collect();
        let indexes: &[PackIndex] = &packs;
        let read: Vec<(usize, Result<Object>)> = packed
            .par_iter()
            .map(|(i, p, e)| (*i, indexes[*p].fetch(backend, &refs[*i], *e)))
            .collect();
        drop(packs);
        for (i, r) in read {
            found[i] = Some(r);
        }
        found
            .into_iter()
            .zip(refs)
            .map(|(f, r)| match f {
                Some(f) => f,
                None => self.get(r),
            })
            .collect()
    }

    pub fn get_typed(&self, r: &ObjectRef, want: ObjectType) -> Result<Vec<u8>> {
        let (ty, body) = self.get(r)?;
        if ty != want {
            return Err(StoreError::WrongType(*r, ty.name(), want.name()));
        }
        Ok(body)
    }

    /// Store a change and the text blobs (and their chunks) its large
    /// inserts are stored as (spec §6.1).
    pub fn put_change(&self, c: &Change) -> Result<ObjectRef> {
        let bytes = c.encode_streaming(|batch| self.put_texts(&batch))?;
        self.put(ObjectType::Change, &bytes)
    }

    /// Store text blobs and their chunks: every chunk at once (compressed
    /// on every core in a large batch), then the blobs.
    pub fn put_texts(&self, texts: &[crate::objects::StoredText]) -> Result<()> {
        let mut chunks: Vec<(ObjectType, &[u8])> = Vec::new();
        for t in texts {
            if !self.has(&t.blob.hash()) {
                chunks.extend(
                    crate::chunk::chunks(&t.content)
                        .into_iter()
                        .map(|c| (ObjectType::Chunk, c)),
                );
            }
        }
        self.put_many(&chunks)?;
        for t in texts {
            self.put(ObjectType::Blob, &t.blob.encode())?;
        }
        Ok(())
    }

    pub fn get_change(&self, r: &ObjectRef) -> Result<Change> {
        Ok(self.get_change_parts(r)?.0)
    }

    /// A change and the objects its stored inserts are made of: each text
    /// blob and its chunks, which travel and are kept with it.
    pub fn get_change_parts(&self, r: &ObjectRef) -> Result<(Change, Vec<ObjectRef>)> {
        let bytes = self.get_typed(r, ObjectType::Change)?;
        let mut parts = Vec::new();
        let mut failed = None;
        let decoded = Change::decode_with(&bytes, &mut |text, count, form| match self
            .read_text(text)
        {
            Ok((blob, content)) => {
                parts.push(*text);
                parts.extend(blob.chunks);
                crate::objects::text_lines(&content, count, form)
            }
            Err(e) => {
                let why = e.to_string();
                failed = Some(e);
                Err(ObjectError::Schema(format!(
                    "text blob {}: {why}",
                    text.short(12)
                )))
            }
        });
        match (decoded, failed) {
            (Ok(c), _) => Ok((c, parts)),
            (Err(_), Some(StoreError::NotFound(t))) => Err(StoreError::Corrupt(
                *r,
                format!("its text blob {} is missing", t.short(12)),
            )),
            (Err(e), _) => Err(StoreError::Corrupt(*r, e.to_string())),
        }
    }

    /// A change with every insert left without its lines (empty
    /// placeholders), and where each one's lines are, in order: for
    /// history-only states, which read text only for the graphs they build
    /// (see [`Change::decode_skeleton`]).
    pub fn get_change_lazy(&self, r: &ObjectRef) -> Result<(Change, Vec<crate::objects::TextRef>)> {
        let bytes = self.get_typed(r, ObjectType::Change)?;
        Change::decode_skeleton(&bytes, *r).map_err(|e| StoreError::Corrupt(*r, e.to_string()))
    }

    /// A change's metadata (id, dependencies, author, clock, message, kind)
    /// without its contents: read from its skeleton, so neither its text
    /// blobs nor its inline lines are read (spec §11).
    ///
    /// Served from the `metas` cache when this store read it before (a
    /// change object can be as large as the tree it adds), as long as the
    /// store still has the change.
    pub fn get_change_meta(&self, r: &ObjectRef) -> Result<crate::state::ChangeMeta> {
        if self.metas.borrow().is_none() {
            *self.metas.borrow_mut() = Some(self.read_metas_cache());
        }
        let cached = self.metas.borrow().as_ref().and_then(|m| m.get(r).cloned());
        if let Some(meta) = cached {
            if self.has(r) {
                #[cfg(test)]
                assert_eq!(
                    meta,
                    Change::decode_meta(&self.get_typed(r, ObjectType::Change)?).unwrap(),
                    "the metadata cache holds a change's metadata"
                );
                return Ok(meta);
            }
        }
        let bytes = self.get_typed(r, ObjectType::Change)?;
        let meta =
            Change::decode_meta(&bytes).map_err(|e| StoreError::Corrupt(*r, e.to_string()))?;
        self.remember_meta(r, &meta);
        Ok(meta)
    }

    /// The metadata cache: records of a change hash, its metadata in the
    /// flat layout, and the first 8 bytes of BLAKE3 over both, each
    /// appended in one write. Derived only: a record that does not check
    /// ends the reading (a torn write), and the cache can be deleted. It
    /// lives in the scratch directory, on this disk.
    fn read_metas_cache(&self) -> crate::fasthash::FastMap<ObjectRef, crate::state::ChangeMeta> {
        let mut out = crate::fasthash::FastMap::default();
        let Ok(bytes) = fs::read(self.scratch.join("metas")) else {
            return out;
        };
        let mut r = crate::flat::Reader::new(&bytes);
        let mut good = 0;
        while !r.at_end() {
            let mut record = || -> std::result::Result<_, ObjectError> {
                let h = r.object()?;
                let body = r.bytes()?;
                let check = r.fixed::<8>()?;
                Ok((h, body, check))
            };
            let Ok((h, body, check)) = record() else {
                break;
            };
            let mut sum = blake3::Hasher::new();
            sum.update(&h.0).update(body);
            if sum.finalize().as_bytes()[..8] != check {
                break;
            }
            match crate::state::ChangeMeta::read_flat(&mut crate::flat::Reader::new(body)) {
                Ok(m) => out.insert(h, m),
                Err(_) => break,
            };
            good += 32 + 4 + body.len() + 8;
        }
        // Cut a bad tail, so records appended later are read.
        if good < bytes.len() {
            let _ = fs::OpenOptions::new()
                .write(true)
                .open(self.scratch.join("metas"))
                .and_then(|f| f.set_len(good as u64));
        }
        out
    }

    /// Add a change's metadata to the cache; quietly nothing when it
    /// cannot be written (it is only a cache).
    fn remember_meta(&self, r: &ObjectRef, meta: &crate::state::ChangeMeta) {
        let mut body = crate::flat::Writer::default();
        meta.write_flat(&mut body);
        let mut w = crate::flat::Writer::default();
        w.object(r);
        w.bytes(&body.0);
        let mut sum = blake3::Hasher::new();
        sum.update(&r.0).update(&body.0);
        w.fixed(&sum.finalize().as_bytes()[..8]);
        let _ = fs::OpenOptions::new()
            .create(true)
            .append(true)
            .open(self.scratch.join("metas"))
            .and_then(|mut f| f.write_all(&w.0));
        if let Some(m) = self.metas.borrow_mut().as_mut() {
            m.insert(*r, meta.clone());
        }
    }

    fn read_text(&self, r: &ObjectRef) -> Result<(Blob, Vec<u8>)> {
        let blob = Blob::decode(&self.get_typed(r, ObjectType::Blob)?)?;
        let mut content = Vec::with_capacity(blob.length as usize);
        for c in &blob.chunks {
            content.extend(self.get_typed(c, ObjectType::Chunk)?);
        }
        Ok((blob, content))
    }

    pub fn put_chunk(&self, bytes: &[u8]) -> Result<ObjectRef> {
        self.put(ObjectType::Chunk, bytes)
    }

    /// Store a chunk that packs will keep uncompressed (`nocompress`, spec §13.3).
    pub fn put_chunk_nocompress(&self, bytes: &[u8]) -> Result<ObjectRef> {
        let r = hash_object(ObjectType::Chunk, bytes);
        put_new_or_existing(
            self.backend.as_ref(),
            &format!("nocompress/{}", r.to_hex()),
            b"",
        )?;
        self.put(ObjectType::Chunk, bytes)
    }

    fn is_nocompress(&self, r: &ObjectRef) -> bool {
        self.backend
            .has(&format!("nocompress/{}", r.to_hex()))
            .unwrap_or(false)
    }

    /// Every chunk marked to be kept raw.
    fn nocompress_set(&self) -> Result<std::collections::HashSet<ObjectRef>> {
        Ok(self
            .backend
            .list("nocompress/")
            .map_err(backend_err)?
            .into_iter()
            .filter_map(|e| ObjectRef::from_hex(e.key.trim_start_matches("nocompress/")))
            .collect())
    }

    pub fn put_state(&self, st: &StateObject) -> Result<ObjectRef> {
        self.put(ObjectType::State, &st.encode())
    }

    pub fn get_state(&self, r: &ObjectRef) -> Result<StateObject> {
        Ok(StateObject::decode(&self.get_typed(r, ObjectType::State)?)?)
    }

    pub fn put_attest(&self, a: &Attest) -> Result<ObjectRef> {
        self.put(ObjectType::Attest, &a.encode())
    }

    pub fn get_attest(&self, r: &ObjectRef) -> Result<Attest> {
        Ok(Attest::decode(&self.get_typed(r, ObjectType::Attest)?)?)
    }

    pub fn put_receipt(&self, r: &Receipt) -> Result<ObjectRef> {
        self.put(ObjectType::Receipt, &r.encode())
    }

    pub fn get_receipt(&self, r: &ObjectRef) -> Result<Receipt> {
        Ok(Receipt::decode(&self.get_typed(r, ObjectType::Receipt)?)?)
    }

    /// Record `value` under `key` in `index`: one write-once entry, so two
    /// writers of a shared store never lose each other's.
    fn index_add(&self, index: &str, key: &ObjectRef, value: &ObjectRef) -> Result<()> {
        let entry = format!("{}{}", fanout_prefix(index, key), value.to_hex());
        put_new_or_existing(self.backend.as_ref(), &entry, b"")
    }

    /// The values recorded under `key` in `index`, with those an older
    /// store kept as one list in `legacy`, sorted.
    fn index_get(&self, index: &str, legacy: &str, key: &ObjectRef) -> Result<Vec<ObjectRef>> {
        let mut values: Vec<ObjectRef> = self
            .backend
            .list(&fanout_prefix(index, key))
            .map_err(backend_err)?
            .into_iter()
            .filter_map(|e| e.key.rsplit('/').next().and_then(ObjectRef::from_hex))
            .collect();
        if let Some(bytes) = get_optional(self.backend.as_ref(), &legacy_key(legacy, key))? {
            values.extend(
                String::from_utf8_lossy(&bytes)
                    .lines()
                    .filter_map(ObjectRef::from_hex),
            );
        }
        values.sort();
        values.dedup();
        Ok(values)
    }

    /// Record that seal attestation `attest` covers `change`.
    pub fn index_seal(&self, change: &ObjectRef, attest: &ObjectRef) -> Result<()> {
        self.index_add("seal", change, attest)
    }

    /// Seal attestations known to cover `change`.
    pub fn seals_for(&self, change: &ObjectRef) -> Result<Vec<ObjectRef>> {
        self.index_get("seal", "seals", change)
    }

    /// Record receipt attestation `attest` for `state` (spec §14.6).
    pub fn index_receipt(&self, state: &ObjectRef, attest: &ObjectRef) -> Result<()> {
        self.index_add("receipt", state, attest)
    }

    /// Receipt attestations recorded for `state`.
    pub fn receipts_for(&self, state: &ObjectRef) -> Result<Vec<ObjectRef>> {
        self.index_get("receipt", "receipts", state)
    }

    /// Every state with a receipt, for garbage collection.
    pub fn receipt_states(&self) -> Result<Vec<ObjectRef>> {
        let mut out: Vec<ObjectRef> = Vec::new();
        for e in self.backend.list("receipt/").map_err(backend_err)? {
            // receipt/<hh>/<rest>/<attest>
            let mut parts = e.key.splitn(4, '/');
            let (Some(_), Some(hh), Some(rest), Some(_)) =
                (parts.next(), parts.next(), parts.next(), parts.next())
            else {
                continue;
            };
            out.extend(ObjectRef::from_hex(&format!("{hh}{rest}")));
        }
        for e in self.backend.list("receipts/").map_err(backend_err)? {
            let mut parts = e.key.splitn(3, '/');
            let (Some(_), Some(hh), Some(rest)) = (parts.next(), parts.next(), parts.next()) else {
                continue;
            };
            out.extend(ObjectRef::from_hex(&format!("{hh}{rest}")));
        }
        out.sort();
        out.dedup();
        Ok(out)
    }

    pub fn put_identity(&self, id: &Identity) -> Result<ObjectRef> {
        self.put(ObjectType::Identity, &id.encode())
    }

    pub fn get_identity(&self, r: &ObjectRef) -> Result<Identity> {
        Ok(Identity::decode(&self.get_typed(r, ObjectType::Identity)?)?)
    }

    /// Signatures stored alongside object `r` (spec §5.4), one per signing
    /// key, sorted by key. Empty if none.
    pub fn signatures(&self, r: &ObjectRef) -> Result<Vec<Signature>> {
        let mut sigs: Vec<Signature> = Vec::new();
        for e in self
            .backend
            .list(&fanout_prefix("sig", r))
            .map_err(backend_err)?
        {
            let bytes = self.backend.get(&e.key).map_err(backend_err)?;
            sigs.extend(Signature::decode_list(&bytes)?);
        }
        // An older store's one list per object; a signer kept there and
        // here counts once, as written here.
        if let Some(bytes) = get_optional(self.backend.as_ref(), &legacy_key("sigs", r))? {
            for sig in Signature::decode_list(&bytes)? {
                if !sigs.iter().any(|s| s.key == sig.key) {
                    sigs.push(sig);
                }
            }
        }
        sigs.sort_by_key(|s| s.key);
        sigs.dedup_by_key(|s| s.key);
        Ok(sigs)
    }

    /// Attach a signature to object `r`, replacing any earlier one by the
    /// same key: one entry per signer, so the latest by a key is the one
    /// kept, and two signers never touch each other's.
    pub fn put_signature(&self, r: &ObjectRef, sig: &Signature) -> Result<()> {
        let key = format!("{}{}", fanout_prefix("sig", r), sig.key.to_hex());
        let bytes = Signature::encode_list(std::slice::from_ref(sig));
        if get_optional(self.backend.as_ref(), &key)?.as_deref() == Some(bytes.as_slice()) {
            return Ok(());
        }
        delete(self.backend.as_ref(), &key)?;
        put_new_or_existing(self.backend.as_ref(), &key, &bytes)
    }

    /// Every object reference in the store, loose and packed (slow; for gc and tests).
    pub fn list(&self) -> io::Result<Vec<ObjectRef>> {
        let mut out: Vec<ObjectRef> = self
            .loose_objects()
            .map_err(|e| io::Error::other(e.to_string()))?
            .into_iter()
            .map(|(r, _)| r)
            .collect();
        for p in self.packs.borrow().iter() {
            out.extend(p.refs.iter().copied());
        }
        out.sort();
        out.dedup();
        Ok(out)
    }
}

/// A loose object's bytes, verified against `r`.
fn parse_loose(data: &[u8], r: &ObjectRef) -> Result<Object> {
    let nul = data
        .iter()
        .position(|&b| b == 0)
        .ok_or_else(|| StoreError::Corrupt(*r, "no header".into()))?;
    let head = std::str::from_utf8(&data[..nul])
        .map_err(|_| StoreError::Corrupt(*r, "header not UTF-8".into()))?;
    let mut parts = head.split(' ');
    let (Some("oxy1"), Some(ty), Some(len), None) =
        (parts.next(), parts.next(), parts.next(), parts.next())
    else {
        return Err(StoreError::Corrupt(*r, format!("bad header {head:?}")));
    };
    let ty = ObjectType::from_name(ty)
        .ok_or_else(|| StoreError::Corrupt(*r, format!("unknown type {ty}")))?;
    let len: usize = len
        .parse()
        .map_err(|_| StoreError::Corrupt(*r, "bad length".into()))?;
    let body = &data[nul + 1..];
    if body.len() != len {
        return Err(StoreError::Corrupt(
            *r,
            format!("length {} != header {len}", body.len()),
        ));
    }
    if hash_object(ty, body) != *r {
        return Err(StoreError::Corrupt(*r, "hash mismatch".into()));
    }
    Ok((ty, body.to_vec()))
}

/// Whether the objects are files that a test may tamper with: not under
/// `OXY_STORE_BACKEND=memory`.
#[cfg(test)]
fn files_on_disk() -> bool {
    std::env::var_os("OXY_STORE_BACKEND").is_none()
}

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

    /// The store under test twice: in a directory, and on a store in
    /// memory with a scratch directory, as over a bucket.
    fn each_store(dir: &Path, f: impl Fn(&str, Store)) {
        f("directory", Store::open(dir.join("objects")).unwrap());
        f(
            "memory",
            Store::open_on(Arc::new(MemoryStore::new()), dir.join("scratch")).unwrap(),
        );
    }

    #[test]
    fn put_get_verify() {
        if !super::files_on_disk() {
            return;
        }
        let dir = tempfile::tempdir().unwrap();
        let s = Store::open(dir.path().join("objects")).unwrap();
        let r = s.put(ObjectType::Chunk, b"hello").unwrap();
        assert!(s.has(&r));
        assert_eq!(s.get(&r).unwrap(), (ObjectType::Chunk, b"hello".to_vec()));
        assert_eq!(s.put(ObjectType::Chunk, b"hello").unwrap(), r);
        assert_eq!(s.list().unwrap(), vec![r]);
        assert!(matches!(
            s.get_typed(&r, ObjectType::Blob),
            Err(StoreError::WrongType(..))
        ));
        assert!(matches!(
            s.get(&ObjectRef([9; 32])),
            Err(StoreError::NotFound(_))
        ));

        // Tamper: flip a byte, read must fail.
        let path = s.path_for(&r);
        let mut data = fs::read(&path).unwrap();
        let last = data.len() - 1;
        data[last] ^= 1;
        fs::write(&path, data).unwrap();
        assert!(matches!(s.get(&r), Err(StoreError::Corrupt(..))));
    }

    #[test]
    fn packing_keeps_every_object_readable() {
        if !super::files_on_disk() {
            return;
        }
        let dir = tempfile::tempdir().unwrap();
        let s = Store::open(dir.path()).unwrap();
        let mut refs = Vec::new();
        for i in 0..300u32 {
            let body = format!("object {i} ").repeat(1 + (i % 7) as usize);
            refs.push((s.put(ObjectType::Chunk, body.as_bytes()).unwrap(), body));
        }
        let c = crate::objects::tests::sample_change();
        let cr = s.put_change(&c).unwrap();
        assert_eq!(s.loose_count().unwrap(), 301);
        let report = s.pack_loose().unwrap();
        assert_eq!(report.objects, 301);
        assert!(report.packed_bytes < report.raw_bytes, "{report:?}");
        assert_eq!(s.loose_count().unwrap(), 0);
        assert_eq!(s.pack_count(), 1);
        for (r, body) in &refs {
            assert!(s.has(r));
            assert_eq!(
                s.get(r).unwrap(),
                (ObjectType::Chunk, body.as_bytes().to_vec())
            );
        }
        assert_eq!(s.get_change(&cr).unwrap(), c);
        assert_eq!(s.list().unwrap().len(), 301);
        // Re-putting a packed object is a no-op; new objects go loose and a
        // second pack coexists with the first.
        assert_eq!(
            s.put(ObjectType::Chunk, refs[0].1.as_bytes()).unwrap(),
            refs[0].0
        );
        assert_eq!(s.loose_count().unwrap(), 0);
        let extra = s.put(ObjectType::Chunk, b"after packing").unwrap();
        s.pack_loose().unwrap();
        assert_eq!(s.pack_count(), 2);
        assert_eq!(s.get(&extra).unwrap().1, b"after packing");
        // A reopened store sees the packs.
        let s2 = Store::open(dir.path()).unwrap();
        assert_eq!(s2.pack_count(), 2);
        assert_eq!(s2.get(&refs[5].0).unwrap().1, refs[5].1.as_bytes());
        assert!(matches!(
            s2.get(&ObjectRef([9; 32])),
            Err(StoreError::NotFound(_))
        ));
        // Corrupting an object's body inside the pack is detected on read.
        let target = refs[5].0;
        let (pack_path, offset) = {
            let packs = s2.packs.borrow();
            let (i, e) = s2.find_packed(&target).unwrap();
            (dir.path().join(&packs[i].key), e.offset as usize)
        };
        let mut data = fs::read(&pack_path).unwrap();
        data[offset] ^= 0xff;
        fs::write(&pack_path, data).unwrap();
        let s3 = Store::open(dir.path()).unwrap();
        assert!(matches!(s3.get(&target), Err(StoreError::Corrupt(..))));
        assert_eq!(
            s3.get(&refs[6].0).unwrap().1,
            refs[6].1.as_bytes(),
            "other objects are unaffected"
        );
    }

    #[test]
    fn a_full_repack_trains_a_dictionary_and_reads_back() {
        if !super::files_on_disk() {
            return;
        }
        let dir = tempfile::tempdir().unwrap();
        let s = Store::open(dir.path()).unwrap();
        // Many small, alike files: what a dictionary is for.
        let bodies: Vec<Vec<u8>> = (0..1500)
            .map(|i| {
                format!(
                    "// SPDX-License-Identifier: GPL-2.0\n#include <linux/module.h>\n\
                     static int probe_{i}(struct device *dev)\n{{\n\treturn {i};\n}}\n"
                )
                .into_bytes()
            })
            .collect();
        let refs: Vec<ObjectRef> = bodies
            .iter()
            .map(|b| s.put(ObjectType::Chunk, b).unwrap())
            .collect();
        s.repack_all(REPACK_ZSTD_LEVEL).unwrap();
        let dicts = fs::read_dir(dir.path())
            .unwrap()
            .flatten()
            .filter(|e| e.path().extension().is_some_and(|x| x == "dict"))
            .count();
        assert_eq!(dicts, 1, "the pack has a dictionary");
        let with_dict = refs
            .iter()
            .filter(|r| s.find_packed(r).unwrap().1.flags & FLAG_DICT != 0)
            .count();
        assert!(with_dict > 1000, "most use it: {with_dict}");
        let again = Store::open(dir.path()).unwrap();
        for (r, b) in refs.iter().zip(&bodies) {
            assert_eq!(&again.get(r).unwrap().1, b);
        }
        // A repack after that replaces the pack and its dictionary.
        again.repack_all(REPACK_ZSTD_LEVEL).unwrap();
        let files: Vec<String> = fs::read_dir(dir.path())
            .unwrap()
            .flatten()
            .map(|e| e.file_name().to_string_lossy().into_owned())
            .filter(|n| n.starts_with("pack-"))
            .collect();
        assert_eq!(files.len(), 3, "one pack, index and dictionary: {files:?}");
        assert_eq!(&again.get(&refs[7]).unwrap().1, &bodies[7]);
    }

    #[test]
    fn nocompress_chunks_pack_raw() {
        let dir = tempfile::tempdir().unwrap();
        let s = Store::open(dir.path()).unwrap();
        let body = vec![b'a'; 10_000];
        let raw = s.put_chunk_nocompress(&body).unwrap();
        let compressible = s.put(ObjectType::Chunk, &vec![b'b'; 10_000]).unwrap();
        s.pack_loose().unwrap();
        let packs = s.packs.borrow();
        let (i, e) = s.find_packed(&raw).unwrap();
        assert_eq!((e.flags, e.stored_len as usize), (0, 10_000));
        let (_, c) = s.find_packed(&compressible).unwrap();
        assert_eq!(c.flags, FLAG_ZSTD);
        assert!(c.stored_len < 10_000);
        assert_eq!(packs[i].fetch(s.backend.as_ref(), &raw, e).unwrap().1, body);
    }

    #[test]
    fn many_packs_merge_into_one() {
        let dir = tempfile::tempdir().unwrap();
        each_store(dir.path(), |which, s| {
            let mut refs = Vec::new();
            for i in 0..(MAX_PACKS + 2) {
                refs.push(
                    s.put(ObjectType::Chunk, format!("pack {i}").as_bytes())
                        .unwrap(),
                );
                s.pack_loose().unwrap();
            }
            assert_eq!(s.pack_count(), MAX_PACKS + 2, "{which}");
            assert_eq!(s.merge_packs().unwrap(), MAX_PACKS + 2);
            assert_eq!(s.pack_count(), 1);
            assert_eq!(s.merge_packs().unwrap(), 0);
            for (i, r) in refs.iter().enumerate() {
                assert_eq!(s.get(r).unwrap().1, format!("pack {i}").as_bytes());
            }
            let again = Store::open_on(s.backend().clone(), s.dir()).unwrap();
            assert_eq!(again.pack_count(), 1, "{which}: reopened");
        });
    }

    #[test]
    fn prune_removes_unreachable_loose_and_packed_objects() {
        use std::collections::HashSet;
        let dir = tempfile::tempdir().unwrap();
        each_store(dir.path(), |which, s| {
            let keep_a = s.put(ObjectType::Chunk, b"keep a").unwrap();
            let drop_a = s.put(ObjectType::Chunk, b"drop a").unwrap();
            s.pack_loose().unwrap();
            let keep_b = s.put(ObjectType::Chunk, b"keep b").unwrap();
            let drop_b = s.put(ObjectType::Chunk, b"drop b").unwrap();
            let sk = crate::identity::SecretKey {
                ed_seed: [1; 32],
                ml_seed: [2; 32],
            };
            let id = sk.identity("a", "b", 0);
            let key = s.put_identity(&id).unwrap();
            s.put_signature(
                &drop_b,
                &sk.sign(crate::identity::Domain::Change, &key, &drop_b),
            )
            .unwrap();
            s.index_seal(&drop_a, &ObjectRef([3; 32])).unwrap();
            let keep: HashSet<ObjectRef> = [keep_a, keep_b, key].into_iter().collect();
            // Everything is too young under a grace period.
            let r = s
                .prune(&keep, std::time::Duration::from_secs(3600))
                .unwrap();
            assert_eq!((r.objects, r.deferred), (0, 2), "{which}");
            let r = s.prune(&keep, std::time::Duration::ZERO).unwrap();
            assert_eq!(r.objects, 2, "{which}: {r:?}");
            assert!(s.has(&keep_a) && s.has(&keep_b) && s.has(&key));
            assert!(!s.has(&drop_a) && !s.has(&drop_b));
            assert!(s.signatures(&drop_b).unwrap().is_empty());
            assert!(s.seals_for(&drop_a).unwrap().is_empty());
            assert_eq!(s.pack_count(), 1);
            assert_eq!(s.get(&keep_a).unwrap().1, b"keep a");
            assert_eq!(s.list().unwrap().len(), 3);
        });
    }

    #[test]
    fn signatures_are_stored_beside_objects() {
        use crate::identity::{Domain, SecretKey};
        let dir = tempfile::tempdir().unwrap();
        each_store(dir.path(), |which, s| {
            let r = s.put(ObjectType::Chunk, b"x").unwrap();
            assert!(s.signatures(&r).unwrap().is_empty());
            let sk = SecretKey {
                ed_seed: [1; 32],
                ml_seed: [2; 32],
            };
            let id = sk.identity("a", "b", 0);
            let key = s.put_identity(&id).unwrap();
            assert_eq!(s.get_identity(&key).unwrap(), id);
            let sig = sk.sign(Domain::Change, &key, &r);
            s.put_signature(&r, &sig).unwrap();
            s.put_signature(&r, &sig).unwrap();
            assert_eq!(s.signatures(&r).unwrap(), vec![sig.clone()], "{which}");
            // A second signer is kept beside the first, sorted by key.
            let sk2 = SecretKey {
                ed_seed: [3; 32],
                ml_seed: [4; 32],
            };
            let key2 = s.put_identity(&sk2.identity("c", "d", 0)).unwrap();
            let sig2 = sk2.sign(Domain::Change, &key2, &r);
            s.put_signature(&r, &sig2).unwrap();
            let mut both = vec![sig.clone(), sig2.clone()];
            both.sort_by_key(|s| s.key);
            assert_eq!(s.signatures(&r).unwrap(), both);
            // The object itself is untouched by its signatures.
            assert_eq!(s.get(&r).unwrap().1, b"x");
            // Seal and receipt entries, and the states with receipts.
            s.index_seal(&r, &ObjectRef([5; 32])).unwrap();
            s.index_seal(&r, &ObjectRef([4; 32])).unwrap();
            s.index_seal(&r, &ObjectRef([5; 32])).unwrap();
            assert_eq!(
                s.seals_for(&r).unwrap(),
                vec![ObjectRef([4; 32]), ObjectRef([5; 32])]
            );
            let st = ObjectRef([6; 32]);
            s.index_receipt(&st, &ObjectRef([7; 32])).unwrap();
            assert_eq!(s.receipts_for(&st).unwrap(), vec![ObjectRef([7; 32])]);
            assert_eq!(s.receipt_states().unwrap(), vec![st]);
        });
    }

    /// A store written before signatures, seals and receipts were one
    /// entry each (`sigs/`, `seals/`, `receipts/` lists) still reads, and
    /// what is written now sits beside what was there.
    #[test]
    fn an_older_store_s_sidecar_lists_are_still_read() {
        if !super::files_on_disk() {
            return;
        }
        use crate::identity::{Domain, SecretKey};
        let dir = tempfile::tempdir().unwrap();
        let s = Store::open(dir.path()).unwrap();
        let r = s.put(ObjectType::Chunk, b"old").unwrap();
        let h = r.to_hex();
        let sk = SecretKey {
            ed_seed: [1; 32],
            ml_seed: [2; 32],
        };
        let key = s.put_identity(&sk.identity("a", "b", 0)).unwrap();
        let sig = sk.sign(Domain::Change, &key, &r);
        let legacy = dir.path().join("sigs").join(&h[..2]);
        fs::create_dir_all(&legacy).unwrap();
        fs::write(
            legacy.join(&h[2..]),
            Signature::encode_list(std::slice::from_ref(&sig)),
        )
        .unwrap();
        let seals = dir.path().join("seals").join(&h[..2]);
        fs::create_dir_all(&seals).unwrap();
        fs::write(
            seals.join(&h[2..]),
            format!("{}\n", ObjectRef([9; 32]).to_hex()),
        )
        .unwrap();
        let receipts = dir.path().join("receipts").join(&h[..2]);
        fs::create_dir_all(&receipts).unwrap();
        fs::write(
            receipts.join(&h[2..]),
            format!("{}\n", ObjectRef([8; 32]).to_hex()),
        )
        .unwrap();

        assert_eq!(s.signatures(&r).unwrap(), vec![sig.clone()]);
        assert_eq!(s.seals_for(&r).unwrap(), vec![ObjectRef([9; 32])]);
        assert_eq!(s.receipts_for(&r).unwrap(), vec![ObjectRef([8; 32])]);
        assert_eq!(s.receipt_states().unwrap(), vec![r]);
        // New entries join the old ones; the same signer is not doubled.
        s.put_signature(&r, &sig).unwrap();
        s.index_seal(&r, &ObjectRef([1; 32])).unwrap();
        assert_eq!(s.signatures(&r).unwrap().len(), 1);
        assert_eq!(
            s.seals_for(&r).unwrap(),
            vec![ObjectRef([1; 32]), ObjectRef([9; 32])]
        );
        // Pruning the object takes both layouts with it.
        let r2 = s.put(ObjectType::Chunk, b"kept").unwrap();
        let keep = [r2, key].into_iter().collect();
        s.prune(&keep, std::time::Duration::ZERO).unwrap();
        assert!(s.signatures(&r).unwrap().is_empty());
        assert!(s.seals_for(&r).unwrap().is_empty());
        assert!(s.receipt_states().unwrap().is_empty());
    }

    #[test]
    fn change_round_trip_through_store() {
        let dir = tempfile::tempdir().unwrap();
        each_store(dir.path(), |which, s| {
            let c = crate::objects::tests::sample_change();
            let r = s.put_change(&c).unwrap();
            assert_eq!(r, c.hash(), "{which}");
            assert_eq!(s.get_change(&r).unwrap(), c);
            assert_eq!(s.get_change_meta(&r).unwrap().message, c.message);
        });
    }

    fn xorshift(state: &mut u64) -> u64 {
        *state ^= *state << 13;
        *state ^= *state >> 7;
        *state ^= *state << 17;
        *state
    }

    /// Random corruption of pack and index files never panics: reads fail
    /// with an error or the store refuses to open, and untouched objects
    /// stay readable when the index survives.
    #[test]
    fn corrupted_packs_and_indexes_fail_cleanly() {
        if !super::files_on_disk() {
            return;
        }
        let dir = tempfile::tempdir().unwrap();
        let objects = dir.path().join("objects");
        let s = Store::open(&objects).unwrap();
        let mut refs = Vec::new();
        let mut seed = 0x9e3779b97f4a7c15u64;
        for i in 0..40u32 {
            let len = (xorshift(&mut seed) % 5000) as usize;
            let mut body = vec![0u8; len];
            for b in body.iter_mut() {
                *b = if i % 2 == 0 {
                    (xorshift(&mut seed) & 0xff) as u8
                } else {
                    b'a' + (i % 26) as u8
                };
            }
            refs.push(s.put(ObjectType::Chunk, &body).unwrap());
        }
        s.pack_loose().unwrap();
        let pack = fs::read_dir(&objects)
            .unwrap()
            .filter_map(|e| e.ok().map(|e| e.path()))
            .find(|p| p.extension().is_some_and(|x| x == "pack"))
            .unwrap();
        let idx = pack.with_extension("idx");
        let pristine_pack = fs::read(&pack).unwrap();
        let pristine_idx = fs::read(&idx).unwrap();

        for round in 0..300u32 {
            let (path, pristine) = if round % 2 == 0 {
                (&pack, &pristine_pack)
            } else {
                (&idx, &pristine_idx)
            };
            let mut data = pristine.clone();
            match xorshift(&mut seed) % 4 {
                0 => {
                    let at = (xorshift(&mut seed) as usize) % data.len();
                    data[at] ^= 1 << (xorshift(&mut seed) % 8);
                }
                1 => {
                    let at = (xorshift(&mut seed) as usize) % data.len();
                    data[at] = (xorshift(&mut seed) & 0xff) as u8;
                }
                2 => {
                    let keep = (xorshift(&mut seed) as usize) % data.len();
                    data.truncate(keep);
                }
                _ => {
                    let at = (xorshift(&mut seed) as usize) % data.len();
                    let n = ((xorshift(&mut seed) % 8) + 1) as usize;
                    for k in 0..n.min(data.len() - at) {
                        data[at + k] = 0xff;
                    }
                }
            }
            fs::write(path, &data).unwrap();
            // Opening may refuse a broken index; a store that opens must
            // either return each object intact or report it corrupt/missing.
            if let Ok(s2) = Store::open(&objects) {
                for r in &refs {
                    match s2.get(r) {
                        Ok((ty, body)) => assert_eq!(
                            hash_object(ty, &body),
                            *r,
                            "a returned object always matches its reference"
                        ),
                        Err(StoreError::Corrupt(..))
                        | Err(StoreError::NotFound(_))
                        | Err(StoreError::Io(_)) => {}
                        Err(e) => panic!("unexpected error {e}"),
                    }
                }
            }
            fs::write(path, pristine).unwrap();
        }
        // Restored files read back in full.
        let s3 = Store::open(&objects).unwrap();
        for r in &refs {
            let (ty, body) = s3.get(r).unwrap();
            assert_eq!(hash_object(ty, &body), *r);
        }
    }

    /// A small batch becomes loose objects; a large one becomes a single
    /// pack. Objects are readable while pending, and batches nest.
    #[test]
    fn batches_write_loose_or_one_pack() {
        let dir = tempfile::tempdir().unwrap();
        each_store(dir.path(), |which, s| {
            let outer = s.begin_batch().unwrap();
            let a = s.put(ObjectType::Chunk, b"small a").unwrap();
            {
                let inner = s.begin_batch().unwrap();
                let b = s.put(ObjectType::Chunk, b"small b").unwrap();
                assert!(s.has(&b));
                assert_eq!(s.get(&b).unwrap().1, b"small b");
                inner.finish().unwrap();
                assert_eq!(
                    s.loose_count().unwrap(),
                    0,
                    "{which}: the inner batch defers to the outer one"
                );
            }
            assert_eq!(s.get(&a).unwrap().1, b"small a");
            assert_eq!(
                s.put(ObjectType::Chunk, b"small a").unwrap(),
                a,
                "idempotent while pending"
            );
            outer.finish().unwrap();
            assert_eq!(s.loose_count().unwrap(), 2);
            assert_eq!(s.pack_count(), 0);
            assert_eq!(s.get(&a).unwrap().1, b"small a");

            // Large: spills into one pack, readable during and after.
            let big = s.begin_batch().unwrap();
            let mut refs = Vec::new();
            let mut seed = 7u64;
            for i in 0..300u32 {
                let mut body = vec![b'z'; 40_000];
                if i % 3 == 0 {
                    for b in body.iter_mut() {
                        *b = (xorshift(&mut seed) & 0xff) as u8;
                    }
                }
                body[..4].copy_from_slice(&i.to_le_bytes());
                let r = if i % 5 == 0 {
                    s.put_chunk_nocompress(&body).unwrap()
                } else {
                    s.put(ObjectType::Chunk, &body).unwrap()
                };
                assert_eq!(
                    s.get(&r).unwrap().1,
                    body,
                    "readable while pending (object {i})"
                );
                refs.push((r, body));
            }
            assert!(s.has(&refs[0].0));
            big.finish().unwrap();
            assert_eq!(s.pack_count(), 1);
            assert_eq!(
                s.loose_count().unwrap(),
                2,
                "nothing from the big batch is loose"
            );
            assert!(fs::read_dir(s.dir())
                .unwrap()
                .flatten()
                .all(|e| !e.file_name().to_string_lossy().contains("tmp-")));
            let s2 = Store::open_on(s.backend().clone(), s.dir()).unwrap();
            for (r, body) in &refs {
                assert_eq!(s2.get(r).unwrap().1, *body);
            }
            // A dropped batch still writes its objects.
            {
                let _b = s2.begin_batch().unwrap();
                s2.put(ObjectType::Chunk, b"dropped").unwrap();
            }
            assert_eq!(
                s2.get(&hash_object(ObjectType::Chunk, b"dropped"))
                    .unwrap()
                    .1,
                b"dropped"
            );
        });
    }

    #[test]
    fn get_many_reads_loose_and_packed_objects_in_order() {
        let dir = tempfile::tempdir().unwrap();
        each_store(dir.path(), |which, s| {
            let packed: Vec<ObjectRef> = (0..20)
                .map(|i| {
                    s.put(ObjectType::Chunk, format!("packed {i}").as_bytes())
                        .unwrap()
                })
                .collect();
            s.pack_loose().unwrap();
            let loose: Vec<ObjectRef> = (0..20)
                .map(|i| {
                    s.put(ObjectType::Chunk, format!("loose {i}").as_bytes())
                        .unwrap()
                })
                .collect();
            let missing = hash_object(ObjectType::Chunk, b"never stored");
            let mut refs: Vec<ObjectRef> = Vec::new();
            for i in 0..20 {
                refs.push(loose[i]);
                refs.push(packed[i]);
            }
            refs.push(missing);
            let got = s.get_many(&refs);
            assert_eq!(got.len(), refs.len());
            for (i, r) in refs.iter().enumerate().take(40) {
                let want = if i % 2 == 0 {
                    format!("loose {}", i / 2)
                } else {
                    format!("packed {}", i / 2)
                };
                let (ty, body) = got[i].as_ref().unwrap();
                assert_eq!(
                    (*ty, body.as_slice()),
                    (ObjectType::Chunk, want.as_bytes()),
                    "{which}: {r:?}"
                );
            }
            assert!(matches!(got[40], Err(StoreError::NotFound(_))));
        });
    }
}

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

    #[test]
    fn a_large_batch_packs_every_object_and_indexes_them_all() {
        let dir = tempfile::tempdir().unwrap();
        let store = Store::open(dir.path()).unwrap();
        let mut refs = Vec::new();
        {
            let batch = store.begin_batch().unwrap();
            for i in 0..1000u32 {
                let body = format!("object {i} {}", "x".repeat(20_000 + i as usize)).into_bytes();
                refs.push(store.put(ObjectType::Blob, &body).unwrap());
            }
            batch.finish().unwrap();
        }
        let again = Store::open(dir.path()).unwrap();
        for r in &refs {
            again
                .get(r)
                .unwrap_or_else(|e| panic!("{}: {e}", r.short(12)));
        }
    }
}

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

    #[test]
    fn an_object_over_the_limit_is_refused_and_nothing_is_written() {
        let dir = tempfile::tempdir().unwrap();
        let store = Store::open(dir.path()).unwrap();
        let big = vec![b'x'; MAX_OBJECT_LEN as usize + 1];
        {
            let batch = store.begin_batch().unwrap();
            store.put(ObjectType::Blob, b"small").unwrap();
            let e = store.put(ObjectType::Change, &big).unwrap_err();
            assert!(
                matches!(e, StoreError::TooLarge { kind: "change", .. }),
                "{e}"
            );
            assert!(e.to_string().contains("commit it in parts"), "{e}");
            batch.finish().unwrap();
        }
        // The store still opens and reads what was written.
        let again = Store::open(dir.path()).unwrap();
        again.get(&hash_object(ObjectType::Blob, b"small")).unwrap();
    }
}

#[cfg(test)]
mod text_tests {
    use super::*;
    use crate::objects::{AtomRef, Edit, CHANGE_VERSION};

    fn large_change() -> Change {
        let mut c = crate::objects::tests::sample_change();
        assert_eq!(c.version, CHANGE_VERSION);
        c.edits[0] = Edit::Insert {
            node: crate::ids::NodeId([7; 16]),
            after: AtomRef::Start,
            before: AtomRef::End,
            lines: (0..2000)
                .map(|i| format!("line {i}\n").into_bytes())
                .collect(),
        };
        c
    }

    #[test]
    fn a_large_insert_is_stored_beside_its_change_and_read_back() {
        let dir = tempfile::tempdir().unwrap();
        let s = Store::open(dir.path()).unwrap();
        let c = large_change();
        let r = s.put_change(&c).unwrap();
        assert_eq!(r, c.hash());
        let (_, body) = s.get(&r).unwrap();
        assert!(
            body.len() < 1024,
            "the change holds a reference, not {} bytes of lines",
            body.len()
        );
        let (back, parts) = s.get_change_parts(&r).unwrap();
        assert_eq!(back, c);
        // The text blob and its chunk.
        assert_eq!(parts.len(), 2);
        for p in &parts {
            assert!(s.has(p));
        }
        // Storing it again writes nothing new.
        let n = s.list().unwrap().len();
        s.put_change(&c).unwrap();
        assert_eq!(s.list().unwrap().len(), n);
    }

    #[test]
    fn a_change_whose_text_blob_is_missing_is_reported_not_misread() {
        if !super::files_on_disk() {
            return;
        }
        let dir = tempfile::tempdir().unwrap();
        let s = Store::open(dir.path()).unwrap();
        let c = large_change();
        let r = s.put_change(&c).unwrap();
        let (_, parts) = s.get_change_parts(&r).unwrap();
        fs::remove_file(s.path_for(&parts[0])).unwrap();
        let e = s.get_change(&r).unwrap_err();
        assert!(
            matches!(&e, StoreError::Corrupt(h, why) if *h == r && why.contains("missing")),
            "{e}"
        );
    }
}

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

    #[test]
    fn many_objects_in_a_batch_are_compressed_together_and_all_read_back() {
        let dir = tempfile::tempdir().unwrap();
        let s = Store::open(dir.path()).unwrap();
        let before = s.put(ObjectType::Chunk, b"already here").unwrap();
        let bodies: Vec<Vec<u8>> = (0..200u32)
            .map(|i| format!("chunk {i} ").repeat(6000).into_bytes())
            .collect();
        let mut objects: Vec<(ObjectType, &[u8])> = bodies
            .iter()
            .map(|b| (ObjectType::Chunk, b.as_slice()))
            .collect();
        objects.push((ObjectType::Chunk, bodies[0].as_slice()));
        objects.push((ObjectType::Chunk, b"already here"));
        let refs = {
            let batch = s.begin_batch().unwrap();
            let refs = s.put_many(&objects).unwrap();
            batch.finish().unwrap();
            refs
        };
        assert_eq!(refs.len(), objects.len());
        assert_eq!(refs[200], refs[0], "a repeat is the same object");
        assert_eq!(refs[201], before);
        let again = Store::open(dir.path()).unwrap();
        for (r, b) in refs.iter().zip(&objects) {
            assert_eq!(again.get(r).unwrap().1, b.1);
        }
    }
}
`/`	Search
`g` `c`	Code
`g` `h`	History
`g` `b`	Bookmarks
`g` `t`	Trust
`y`	Permalink to this state (files and trees)
`?`	This sheet
Keyboard
