crates/oxy-core/src/store.rs at 17f94b13ae54 · andymccall/oxygen

# 17f94b13ae54 oxygen / crates / oxy-core / src / store.rs
65.7 KiB · Rust · last changed Store a large change's texts a batch at a time 1 month ago source blame history raw
//! Object store (spec §13.1, §13.3, §13.4): loose objects, one file per
//! object under `objects/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.
//!
//! 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`.

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 rand::RngCore;
use std::cell::RefCell;
use std::fs;
use std::io::{self, Read, Seek, SeekFrom, Write};
use std::path::{Path, PathBuf};

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;
const ZSTD_LEVEL: i32 = 3;
/// 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>);

#[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.
#[derive(Debug)]
struct PackIndex {
    path: PathBuf,
    refs: Vec<ObjectRef>,
    entries: Vec<PackEntry>,
}

impl PackIndex {
    fn load(idx_path: &Path, pack_path: &Path) -> Result<PackIndex> {
        let data = fs::read(idx_path)?;
        let bad =
            |m: &str| StoreError::Corrupt(ObjectRef::ZERO, format!("{}: {m}", idx_path.display()));
        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 = fs::metadata(pack_path)?.len();
        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!(
                        "{}: the index claims an object of {} bytes ({} stored), more than the {MAX_OBJECT_LEN} an object may be",
                        idx_path.display(),
                        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);
        }
        Ok(PackIndex {
            path: pack_path.to_path_buf(),
            refs,
            entries,
        })
    }

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

    fn read(&self, r: &ObjectRef, e: PackEntry) -> Result<(ObjectType, Vec<u8>)> {
        let mut f = fs::File::open(&self.path)?;
        f.seek(SeekFrom::Start(e.offset))?;
        let mut stored = vec![0u8; e.stored_len as usize];
        f.read_exact(&mut stored)?;
        let body = 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))
    }
}

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>;

#[derive(Debug)]
pub struct Store {
    dir: PathBuf,
    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>,
}

/// 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)
}

impl Store {
    pub fn open(dir: impl Into<PathBuf>) -> io::Result<Store> {
        let dir = dir.into();
        fs::create_dir_all(dir.join("loose"))?;
        let store = Store {
            dir,
            packs: RefCell::new(Vec::new()),
            pending: RefCell::new(None),
            batch_depth: std::cell::Cell::new(0),
        };
        store
            .reload_packs()
            .map_err(|e| io::Error::other(e.to_string()))?;
        Ok(store)
    }

    fn reload_packs(&self) -> Result<()> {
        let mut packs = Vec::new();
        let mut names: Vec<PathBuf> = fs::read_dir(&self.dir)?
            .filter_map(|e| e.ok().map(|e| e.path()))
            .filter(|p| {
                p.extension().is_some_and(|x| x == "idx")
                    && p.file_name()
                        .is_some_and(|n| n.to_string_lossy().starts_with("pack-"))
            })
            .collect();
        names.sort();
        for idx in names {
            let pack = idx.with_extension("pack");
            if pack.is_file() {
                packs.push(PackIndex::load(&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 on disk, with their paths.
    fn loose_objects(&self) -> io::Result<Vec<(ObjectRef, PathBuf)>> {
        let mut out = Vec::new();
        let loose = self.dir.join("loose");
        for d in fs::read_dir(&loose)? {
            let d = d?;
            if !d.file_type()?.is_dir() {
                continue;
            }
            let prefix = d.file_name().to_string_lossy().to_string();
            for f in fs::read_dir(d.path())? {
                let f = f?;
                let name = f.file_name().to_string_lossy().to_string();
                if let Some(r) = ObjectRef::from_hex(&format!("{prefix}{name}")) {
                    out.push((r, f.path()));
                }
            }
        }
        out.sort();
        Ok(out)
    }

    pub fn loose_count(&self) -> io::Result<usize> {
        Ok(self.loose_objects()?.len())
    }

    /// Move every loose object into a new pack (spec §13.4). Loose files 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(|(_, p)| p.clone()).collect::<Vec<_>>(),
            &[],
        )
    }

    /// 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, paths): (Vec<ObjectRef>, Vec<PathBuf>) = {
            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.path.clone()).collect(),
            )
        };
        let n = paths.len();
        self.write_pack(&refs, &[], &paths)?;
        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. Signature sidecars and seal-index entries
    /// of deleted objects go too.
    pub fn prune(
        &self,
        keep: &std::collections::HashSet<ObjectRef>,
        grace: std::time::Duration,
    ) -> Result<PruneReport> {
        let now = std::time::SystemTime::now();
        let old_enough = |m: &fs::Metadata| {
            m.modified()
                .map(|t| now.duration_since(t).unwrap_or_default() >= grace)
                .unwrap_or(true)
        };
        let mut report = PruneReport::default();
        // Loose objects.
        for (r, path) in self.loose_objects()? {
            if keep.contains(&r) {
                continue;
            }
            let meta = fs::metadata(&path)?;
            if !old_enough(&meta) {
                report.deferred += 1;
                continue;
            }
            report.bytes += meta.len();
            report.objects += 1;
            fs::remove_file(&path)?;
            let _ = fs::remove_dir(path.parent().unwrap());
            self.remove_sidecars(&r)?;
        }
        // Packs: rewrite those that hold unreachable objects.
        let mut keep_refs: Vec<ObjectRef> = Vec::new();
        let mut old_paths: Vec<PathBuf> = Vec::new();
        let mut rewrite = false;
        for p in self.packs.borrow().iter() {
            let meta = fs::metadata(&p.path)?;
            let unreachable = p.refs.iter().filter(|r| !keep.contains(r)).count();
            if unreachable == 0 {
                continue;
            }
            if !old_enough(&meta) {
                report.deferred += unreachable;
                continue;
            }
            rewrite = true;
            old_paths.push(p.path.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_paths.contains(&p.path))
                    .flat_map(|p| p.refs.iter().copied())
                    .filter(|r| !keep.contains(r))
                    .collect()
            };
            self.write_pack(&keep_refs, &[], &old_paths)?;
            for r in dropped {
                self.remove_sidecars(&r)?;
            }
        }
        Ok(report)
    }

    fn remove_sidecars(&self, r: &ObjectRef) -> Result<()> {
        let sidecars = [
            self.sig_path(r),
            self.index_path("seals", r),
            self.index_path("receipts", r),
        ];
        for path in sidecars {
            match fs::remove_file(&path) {
                Ok(()) => {
                    let _ = fs::remove_dir(path.parent().unwrap());
                }
                Err(e) if e.kind() == io::ErrorKind::NotFound => {}
                Err(e) => return Err(e.into()),
            }
        }
        Ok(())
    }

    /// Write `refs` into a new pack, then delete `loose_paths` and `old_packs`.
    /// Write the index for a complete, synced temporary pack file and move
    /// both into place; `index` must be sorted by ObjectRef.
    fn finish_pack_files(
        &self,
        id: &str,
        tmp_pack: &Path,
        index: &[(ObjectRef, PackEntry)],
    ) -> Result<()> {
        let pack_path = self.dir.join(format!("pack-{id}.pack"));
        let idx_path = self.dir.join(format!("pack-{id}.idx"));
        let tmp_idx = self.dir.join(format!("tmp-{id}.idx"));
        {
            let mut f = io::BufWriter::new(fs::File::create(&tmp_idx)?);
            f.write_all(IDX_MAGIC)?;
            f.write_all(&(index.len() as u32).to_le_bytes())?;
            for (r, e) in index {
                f.write_all(&r.0)?;
                f.write_all(&e.offset.to_le_bytes())?;
                f.write_all(&e.stored_len.to_le_bytes())?;
                f.write_all(&e.raw_len.to_le_bytes())?;
                f.write_all(&[type_index(e.ty), e.flags])?;
            }
            f.flush()?;
            f.get_ref().sync_all()?;
        }
        // The pack first: an index never names a pack that is not there.
        crate::durable::rename(tmp_pack, &pack_path)?;
        crate::durable::rename(&tmp_idx, &idx_path)?;
        self.reload_packs()
    }

    fn write_pack(
        &self,
        refs: &[ObjectRef],
        loose_paths: &[PathBuf],
        old_packs: &[PathBuf],
    ) -> Result<PackReport> {
        if refs.is_empty() {
            for p in old_packs {
                fs::remove_file(p)?;
                fs::remove_file(p.with_extension("idx"))?;
            }
            self.reload_packs()?;
            return Ok(PackReport::default());
        }
        let mut refs: Vec<ObjectRef> = refs.to_vec();
        refs.sort();
        refs.dedup();
        let mut nonce = [0u8; 8];
        rand::rngs::OsRng.fill_bytes(&mut nonce);
        let id = hex::encode(nonce);
        let tmp_pack = self.dir.join(format!("tmp-{id}.pack"));
        let mut report = PackReport::default();
        let mut index: Vec<(ObjectRef, PackEntry)> = Vec::with_capacity(refs.len());
        {
            let mut f = io::BufWriter::new(fs::File::create(&tmp_pack)?);
            f.write_all(PACK_MAGIC)?;
            let mut offset = PACK_MAGIC.len() as u64;
            for r in &refs {
                let (ty, body) = self.get(r)?;
                let (flags, stored) = if self.is_nocompress(r) {
                    (0, body.clone())
                } 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.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()?;
        }
        self.finish_pack_files(&id, &tmp_pack, &index)?;
        for p in old_packs {
            fs::remove_file(p)?;
            fs::remove_file(p.with_extension("idx"))?;
        }
        self.reload_packs()?;
        for path in loose_paths {
            fs::remove_file(path)?;
            if let Some(dir) = path.parent() {
                let _ = fs::remove_dir(dir);
            }
        }
        Ok(report)
    }

    pub fn dir(&self) -> &Path {
        &self.dir
    }

    pub(crate) fn path_for(&self, r: &ObjectRef) -> PathBuf {
        let h = r.to_hex();
        self.dir.join("loose").join(&h[..2]).join(&h[2..])
    }

    pub fn has(&self, r: &ObjectRef) -> bool {
        self.path_for(r).is_file()
            || 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 => {
                for (ty, body) in &pending.buffered {
                    self.put_loose(*ty, body)?;
                }
                Ok(())
            }
            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<(ObjectType, Vec<u8>)>> {
        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)));
                };
                match pack.writer.flush() {
                    Ok(()) => PackIndex {
                        path: pack.tmp_path.clone(),
                        refs: vec![],
                        entries: vec![],
                    }
                    .read(r, e),
                    Err(err) => Err(err.into()),
                }
            }
        })
    }

    /// 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 mut nonce = [0u8; 8];
            rand::rngs::OsRng.fill_bytes(&mut nonce);
            let id = hex::encode(nonce);
            let tmp_path = self.dir.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.path_for(r).is_file()
                && self.find_packed(r).is_none()
            {
                todo.push(i);
            }
        }
        let raw: std::collections::HashSet<ObjectRef> =
            match fs::read_dir(self.dir.join("nocompress")) {
                Ok(rd) => rd
                    .flatten()
                    .filter_map(|e| ObjectRef::from_hex(&e.file_name().to_string_lossy()))
                    .collect(),
                Err(_) => std::collections::HashSet::new(),
            };
        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);
        let path = self.path_for(&r);
        if path.is_file() || self.find_packed(&r).is_some() {
            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 atomically with its own `fsync`.
    fn put_loose(&self, ty: ObjectType, body: &[u8]) -> Result<ObjectRef> {
        let r = hash_object(ty, body);
        let path = self.path_for(&r);
        if path.is_file() {
            return Ok(r);
        }
        let fanout = path.parent().unwrap();
        if !fanout.is_dir() {
            fs::create_dir_all(fanout)?;
            crate::durable::sync_dir(&self.dir.join("loose"))?;
        }
        let mut nonce = [0u8; 8];
        rand::rngs::OsRng.fill_bytes(&mut nonce);
        let tmp = self
            .dir
            .join("loose")
            .join(format!("tmp-{}", hex::encode(nonce)));
        crate::durable::step("write loose object")?;
        {
            let mut f = fs::File::create(&tmp)?;
            f.write_all(&header(ty, body.len()))?;
            f.write_all(body)?;
            f.sync_all()?;
        }
        crate::durable::rename(&tmp, &path)?;
        Ok(r)
    }

    /// Read and verify an object, loose, packed, or pending in a batch.
    pub fn get(&self, r: &ObjectRef) -> Result<(ObjectType, Vec<u8>)> {
        match read_loose_at(&self.path_for(r), r) {
            Some(found) => found,
            None => match self.find_packed(r) {
                Some((i, e)) => self.packs.borrow()[i].read(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,
    /// the rest (packed, or pending in a batch) one by one. Results are in
    /// the order of `refs`.
    pub fn get_many(&self, refs: &[ObjectRef]) -> Vec<Result<Object>> {
        use rayon::prelude::*;
        let paths: Vec<(ObjectRef, PathBuf)> =
            refs.iter().map(|r| (*r, self.path_for(r))).collect();
        let mut found: Vec<Option<Result<Object>>> =
            paths.par_iter().map(|(r, p)| read_loose_at(p, r)).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].read(&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| match self.read_text(text) {
            Ok((blob, content)) => {
                parts.push(*text);
                parts.extend(blob.chunks);
                crate::objects::text_lines(&content, count)
            }
            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).
    pub fn get_change_meta(&self, r: &ObjectRef) -> Result<crate::state::ChangeMeta> {
        Ok(crate::state::ChangeMeta::of(&self.get_change_lazy(r)?.0))
    }

    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);
        let dir = self.dir.join("nocompress");
        fs::create_dir_all(&dir)?;
        let _ = fs::File::create(dir.join(r.to_hex()));
        self.put(ObjectType::Chunk, bytes)
    }

    fn is_nocompress(&self, r: &ObjectRef) -> bool {
        self.dir.join("nocompress").join(r.to_hex()).exists()
    }

    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)?)?)
    }

    fn index_path(&self, index: &str, key: &ObjectRef) -> PathBuf {
        let h = key.to_hex();
        self.dir.join(index).join(&h[..2]).join(&h[2..])
    }

    /// Add `value` to the sorted list kept for `key` in `index`.
    fn index_add(&self, index: &str, key: &ObjectRef, value: &ObjectRef) -> Result<()> {
        let mut values = self.index_get(index, key)?;
        if values.contains(value) {
            return Ok(());
        }
        values.push(*value);
        values.sort();
        let path = self.index_path(index, key);
        fs::create_dir_all(path.parent().unwrap())?;
        let text: String = values.iter().map(|s| format!("{}\n", s.to_hex())).collect();
        crate::durable::write_atomic(&path, text.as_bytes())?;
        Ok(())
    }

    fn index_get(&self, index: &str, key: &ObjectRef) -> Result<Vec<ObjectRef>> {
        match fs::read_to_string(self.index_path(index, key)) {
            Ok(t) => Ok(t.lines().filter_map(ObjectRef::from_hex).collect()),
            Err(e) if e.kind() == io::ErrorKind::NotFound => Ok(vec![]),
            Err(e) => Err(e.into()),
        }
    }

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

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

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

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

    /// Every state with a receipt, for garbage collection.
    pub fn receipt_states(&self) -> Result<Vec<ObjectRef>> {
        let mut out = Vec::new();
        let dir = self.dir.join("receipts");
        let top = match fs::read_dir(&dir) {
            Ok(d) => d,
            Err(e) if e.kind() == io::ErrorKind::NotFound => return Ok(out),
            Err(e) => return Err(e.into()),
        };
        for d in top {
            let d = d?;
            let prefix = d.file_name().to_string_lossy().into_owned();
            for f in fs::read_dir(d.path())? {
                let name = f?.file_name().to_string_lossy().into_owned();
                if let Some(r) = ObjectRef::from_hex(&format!("{prefix}{name}")) {
                    out.push(r);
                }
            }
        }
        out.sort();
        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)?)?)
    }

    fn sig_path(&self, r: &ObjectRef) -> PathBuf {
        let h = r.to_hex();
        self.dir.join("sigs").join(&h[..2]).join(&h[2..])
    }

    /// Signatures stored alongside object `r` (spec §5.4). Empty if none.
    pub fn signatures(&self, r: &ObjectRef) -> Result<Vec<Signature>> {
        match fs::read(self.sig_path(r)) {
            Ok(b) => Ok(Signature::decode_list(&b)?),
            Err(e) if e.kind() == io::ErrorKind::NotFound => Ok(vec![]),
            Err(e) => Err(e.into()),
        }
    }

    /// Attach a signature to object `r`, replacing any earlier one by the same key.
    pub fn put_signature(&self, r: &ObjectRef, sig: &Signature) -> Result<()> {
        let mut sigs = self.signatures(r)?;
        sigs.retain(|s| s.key != sig.key);
        sigs.push(sig.clone());
        sigs.sort_by_key(|s| s.key);
        let path = self.sig_path(r);
        fs::create_dir_all(path.parent().unwrap())?;
        let mut nonce = [0u8; 8];
        rand::rngs::OsRng.fill_bytes(&mut nonce);
        let tmp = self.dir.join(format!("sigs-tmp-{}", hex::encode(nonce)));
        {
            let mut f = fs::File::create(&tmp)?;
            f.write_all(&Signature::encode_list(&sigs))?;
            f.sync_all()?;
        }
        crate::durable::rename(&tmp, &path)?;
        Ok(())
    }

    /// 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()?.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 at `path`, verified against `r`; `None` when there is no
/// such file (it may be packed).
fn read_loose_at(path: &Path, r: &ObjectRef) -> Option<Result<Object>> {
    let data = match fs::read(path) {
        Ok(d) => d,
        Err(e) if e.kind() == io::ErrorKind::NotFound => return None,
        Err(e) => return Some(Err(e.into())),
    };
    Some((|| {
        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()))
    })())
}

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

    #[test]
    fn put_get_verify() {
        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() {
        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();
            (packs[i].path.clone(), 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 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].read(&raw, e).unwrap().1, body);
    }

    #[test]
    fn many_packs_merge_into_one() {
        let dir = tempfile::tempdir().unwrap();
        let s = Store::open(dir.path()).unwrap();
        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);
        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());
        }
        assert_eq!(Store::open(dir.path()).unwrap().pack_count(), 1);
    }

    #[test]
    fn prune_removes_unreachable_loose_and_packed_objects() {
        use std::collections::HashSet;
        let dir = tempfile::tempdir().unwrap();
        let s = Store::open(dir.path()).unwrap();
        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));
        let r = s.prune(&keep, std::time::Duration::ZERO).unwrap();
        assert_eq!(r.objects, 2, "{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();
        let s = Store::open(dir.path()).unwrap();
        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]);
        // The object itself is untouched by its signatures.
        assert_eq!(s.get(&r).unwrap().1, b"x");
    }

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

    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() {
        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();
        let s = Store::open(dir.path().join("objects")).unwrap();
        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,
                "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(dir.path().join("objects"))
            .unwrap()
            .flatten()
            .all(|e| !e.file_name().to_string_lossy().starts_with("tmp-")));
        let s2 = Store::open(dir.path().join("objects")).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();
        let s = Store::open(dir.path().join("objects")).unwrap();
        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()),
                "{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() {
        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
