//! 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.
//! `pack-retired` records the packs and loose objects a repack replaced,
//! which stay for readers until a collection an hour later (oxygenhub#249).
//!
//! A store that is not this disk's directory, where finding an object's
//! entries is a LIST request, also keeps every entry key in the
//! `pack-entries` journal and, once a collection folds them, in entry
//! lists (`pack-<id>.entries`), read once per open (spec §13.7, #116).

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;
/// What reading an entry of a pack whose dictionary is gone says. A pack
/// being deleted can lose its dictionary first, so a read that meets this
/// looks at the packs again before it is believed.
const NO_DICT: &str = "packed with a dictionary the pack lacks";

/// The record of what repacks replaced (oxygenhub#249): packs and loose
/// objects whose contents a newer pack holds, kept for readers that
/// still use them until garbage collection deletes them, at least
/// [`RETIRED_FOR`] later. One line per key, `<retired at, wall ms> <key>`.
/// It is a reference record (updated by compare-and-swap, so concurrent
/// repacks and collections never lose each other's lines), named under
/// `pack-` so the pack listing every open makes shows whether it holds
/// anything.
const RETIRED_KEY: &str = "pack-retired";
/// How long a retired pack or loose object stays in the store: a reader
/// that loaded its list of packs before a repack reads on from the old
/// packs this long, and after that finds the objects in the new pack by
/// looking again (see [`Store::get`]).
pub const RETIRED_FOR: std::time::Duration = std::time::Duration::from_secs(60 * 60);
/// Compare-and-swap rounds for the retired record before giving up.
const RETIRED_TRIES: usize = 32;
/// After a look at the pack listing found nothing new, how long a missed
/// [`Store::has`] answers without looking again: a writer probing for
/// thousands of objects it is about to store would otherwise list the
/// packs for each.
const HAS_RELIST_AFTER: std::time::Duration = std::time::Duration::from_secs(1);

/// The journal of signatures and index entries (spec §13.7): every entry
/// key a writer on a store that is not this disk's directory wrote since
/// the last fold, one line each, `<folded at, wall ms; 0 until folded>
/// <key>`, as the retired record's lines are. A reference record updated
/// by compare-and-swap, so concurrent writers never lose each other's
/// lines; read once per open, instead of a listing per lookup.
pub const JOURNAL_KEY: &str = "pack-entries";
/// A complete entry list (spec §13.7): every entry a fold found.
const LIST_SUFFIX: &str = ".entries";
/// A partial entry list: journal lines a writer folded out.
const PART_SUFFIX: &str = ".entries-part";
/// The first line of an entry list.
const LIST_MAGIC: &str = "oxy-entries 1";
/// Written once a collection has copied the older one-list keys into
/// per-object entries: readers stop looking for them.
const LEGACY_FOLDED_KEY: &str = "pack-legacy-folded";
/// Unfolded journal lines a writer lets the journal hold before it folds
/// them into a partial list.
const JOURNAL_FOLD_AT: usize = 1000;
/// The per-object entry indexes, each with the one-list name older
/// stores kept it under.
const ENTRY_INDEXES: [(&str, &str); 6] = [
    ("sig", "sigs"),
    ("seal", "seals"),
    ("receipt", "receipts"),
    ("approval", "approvals"),
    ("cert", "certs"),
    ("accepted", "accepted-lists"),
];

/// A pack listing as compared: each key with its size and time.
type Listed = Vec<(String, u64, Option<u64>)>;

/// The retired record's lines: each key with the time it was retired.
fn parse_retired(bytes: &[u8]) -> std::collections::BTreeMap<String, u64> {
    String::from_utf8_lossy(bytes)
        .lines()
        .filter_map(|l| {
            let (t, k) = l.split_once(' ')?;
            Some((k.to_string(), t.parse().ok()?))
        })
        .collect()
}

fn encode_retired(m: &std::collections::BTreeMap<String, u64>) -> Vec<u8> {
    m.iter()
        .map(|(k, t)| format!("{t} {k}\n"))
        .collect::<String>()
        .into_bytes()
}

/// What deleting retired packs and loose objects did.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct RetiredReport {
    /// Retired packs deleted, each with its index and dictionary.
    pub packs: usize,
    /// Retired loose objects deleted.
    pub loose: usize,
    /// Retired entries younger than [`RETIRED_FOR`], left for the next
    /// collection.
    pub kept: usize,
}

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

/// Whether a key names an entry list (spec §13.7), complete or partial.
fn is_entry_list(key: &str) -> bool {
    key.starts_with("pack-") && (key.ends_with(LIST_SUFFIX) || key.ends_with(PART_SUFFIX))
}

/// An entry list's bytes: zstd-compressed text, [`LIST_MAGIC`] and then
/// one key per line, sorted.
fn encode_entry_list<'a>(keys: impl Iterator<Item = &'a String>) -> Result<Vec<u8>> {
    let mut text = String::from(LIST_MAGIC);
    text.push('\n');
    for k in keys {
        text.push_str(k);
        text.push('\n');
    }
    Ok(zstd::bulk::compress(text.as_bytes(), ZSTD_LEVEL)?)
}

/// The keys an entry list holds.
fn decode_entry_list(key: &str, bytes: &[u8]) -> Result<Vec<String>> {
    let bad = |why: &str| StoreError::Io(io::Error::other(format!("{key}: {why}")));
    let text = zstd::stream::decode_all(bytes).map_err(|_| bad("not zstd"))?;
    let text = String::from_utf8(text).map_err(|_| bad("not UTF-8"))?;
    let mut lines = text.lines();
    if lines.next() != Some(LIST_MAGIC) {
        return Err(bad("not an entry list"));
    }
    Ok(lines
        .filter(|l| !l.is_empty())
        .map(str::to_string)
        .collect())
}

/// Drop journal lines folded at least [`RETIRED_FOR`] before `now`:
/// every reader that could need them has a list holding them by then.
fn expire_folded(m: &mut std::collections::BTreeMap<String, u64>, now: u64) {
    let hour = RETIRED_FOR.as_millis() as u64;
    m.retain(|_, t| *t == 0 || now.saturating_sub(*t) < hour);
}

/// `<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, NO_DICT.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 {
    /// Load a pack's index; `pack_len` is the pack's size, from the listing
    /// that found it (a stat of it would be a request of its own). `None`
    /// when the index is gone since the listing: the pack was deleted.
    fn load(
        backend: &Backend,
        idx_key: &str,
        pack_key: &str,
        pack_len: u64,
    ) -> Result<Option<PackIndex>> {
        let data = match backend.get(idx_key) {
            Ok(d) => d,
            Err(BackendError::NotFound(_)) => return Ok(None),
            Err(e) => return Err(backend_err(e)),
        };
        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 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(Some(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<Arc<PackIndex>>>,
    /// The pack listing `packs` was loaded from (key, size, time), to tell
    /// whether a look at the store would find anything new.
    packs_listed: RefCell<Listed>,
    /// When a look at the pack listing for a missed [`Store::has`] last
    /// found nothing new.
    has_relisted: std::cell::Cell<Option<std::time::Instant>>,
    /// 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>>>,
    /// The live entry lists (spec §13.7) the pack listing showed: listed
    /// and not retired.
    entry_lists: RefCell<Vec<String>>,
    /// Whether the store is folded: a live entry list is complete.
    folded: std::cell::Cell<bool>,
    /// Whether the pack listing showed `pack-legacy-folded`.
    legacy_folded: std::cell::Cell<bool>,
    /// The entry keys of a folded store, read from its lists and journal
    /// on the first lookup after the packs were loaded.
    entries: RefCell<Option<std::collections::BTreeSet<String>>>,
    /// Entry keys written in an open batch, added to the journal when the
    /// batch is flushed.
    journal_queue: RefCell<Vec<String>>,
}

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

#[cfg(test)]
thread_local! {
    /// A backend every store opened on this thread uses instead of its
    /// directory: tests of a repository kept on a store that is not this
    /// disk's directory, with several handles on one store.
    pub(crate) static TEST_BACKEND: RefCell<Option<Arc<Backend>>> = const { RefCell::new(None) };
}

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"))?;
        #[cfg(test)]
        if let Some(backend) = TEST_BACKEND.with(|t| t.borrow().clone()) {
            return Store::open_on(backend, dir);
        }
        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()),
            packs_listed: RefCell::new(Vec::new()),
            has_relisted: std::cell::Cell::new(None),
            pending: RefCell::new(None),
            batch_depth: std::cell::Cell::new(0),
            metas: RefCell::new(None),
            entry_lists: RefCell::new(Vec::new()),
            folded: std::cell::Cell::new(false),
            legacy_folded: std::cell::Cell::new(false),
            entries: RefCell::new(None),
            journal_queue: RefCell::new(Vec::new()),
        };
        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
    }

    /// Load the list of packs again, from a new listing: what other
    /// writers packed since becomes readable, and packs since retired or
    /// deleted are let go. Packs already loaded are kept, not read again.
    pub fn refresh(&self) -> Result<()> {
        *self.entries.borrow_mut() = None;
        self.reload_packs()
    }

    /// The pack listing, with how `packs_listed` records it.
    fn list_packs(&self) -> Result<(Vec<strata::Entry>, Listed)> {
        let listed = self.backend.list("pack-").map_err(backend_err)?;
        let sig = listed
            .iter()
            .map(|e| (e.key.clone(), e.size, e.modified_ms))
            .collect();
        Ok((listed, sig))
    }

    /// Load every pack the listing names that is not retired. An index
    /// that is gone by the time it is read (its pack deleted after the
    /// listing) is passed over, and the store listed once more, since
    /// what replaced it may be new since the first listing.
    fn reload_packs(&self) -> Result<()> {
        let (listed, sig) = self.list_packs()?;
        self.load_listed(listed, sig)
    }

    fn load_listed(&self, mut listed: Vec<strata::Entry>, mut sig: Listed) -> Result<()> {
        let had: std::collections::HashMap<String, Arc<PackIndex>> = self
            .packs
            .borrow()
            .iter()
            .map(|p| (p.key.clone(), p.clone()))
            .collect();
        let mut loaded: std::collections::HashMap<String, Arc<PackIndex>> = Default::default();
        for attempt in 0..2 {
            let retired = self.retired_in(&listed)?;
            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();
            let mut vanished = false;
            for idx in idx_keys {
                let pack = format!("{}.pack", idx.trim_end_matches(".idx"));
                if retired.contains_key(&pack) {
                    continue;
                }
                let Some(p) = listed.iter().find(|e| e.key == pack) else {
                    continue;
                };
                if let Some(kept) = had.get(&pack).or_else(|| loaded.get(&pack)) {
                    packs.push(kept.clone());
                    continue;
                }
                match PackIndex::load(self.backend.as_ref(), idx, &pack, p.size)? {
                    Some(index) => {
                        let index = Arc::new(index);
                        loaded.insert(pack, index.clone());
                        packs.push(index);
                    }
                    None => vanished = true,
                }
            }
            if vanished && attempt == 0 {
                (listed, sig) = self.list_packs()?;
                continue;
            }
            let lists: Vec<String> = listed
                .iter()
                .map(|e| e.key.as_str())
                .filter(|k| is_entry_list(k) && !retired.contains_key(*k))
                .map(str::to_string)
                .collect();
            self.folded
                .set(lists.iter().any(|k| k.ends_with(LIST_SUFFIX)));
            self.legacy_folded
                .set(listed.iter().any(|e| e.key == LEGACY_FOLDED_KEY));
            *self.entry_lists.borrow_mut() = lists;
            if *self.packs_listed.borrow() != sig {
                *self.entries.borrow_mut() = None;
            }
            *self.packs.borrow_mut() = packs;
            *self.packs_listed.borrow_mut() = sig;
            return Ok(());
        }
        unreachable!("the second round always returns")
    }

    /// Load the packs again when the pack listing differs from the one
    /// they were loaded from; whether it did. What a miss calls before it
    /// is believed (oxygenhub#249): another writer may have packed the
    /// object, or a repack moved it, since this store loaded its packs.
    fn reload_if_changed(&self) -> Result<bool> {
        let (listed, sig) = self.list_packs()?;
        if *self.packs_listed.borrow() == sig {
            return Ok(false);
        }
        self.load_listed(listed, sig)?;
        Ok(true)
    }

    /// The retired record, read when `listed` (a pack listing) shows it
    /// holds anything.
    fn retired_in(
        &self,
        listed: &[strata::Entry],
    ) -> Result<std::collections::BTreeMap<String, u64>> {
        match listed.iter().find(|e| e.key == RETIRED_KEY) {
            Some(e) if e.size > 0 => Ok(self.read_retired()?.0),
            _ => Ok(Default::default()),
        }
    }

    /// The retired record and its version (`None`: never written).
    fn read_retired(
        &self,
    ) -> Result<(
        std::collections::BTreeMap<String, u64>,
        Option<strata::Version>,
    )> {
        self.read_record(RETIRED_KEY)
    }

    /// A record of `<wall ms> <key>` lines (the retired record, the
    /// journal) and its version (`None`: never written).
    fn read_record(
        &self,
        key: &str,
    ) -> Result<(
        std::collections::BTreeMap<String, u64>,
        Option<strata::Version>,
    )> {
        match self.backend.read_ref(key).map_err(backend_err)? {
            Some((bytes, v)) => Ok((parse_retired(&bytes), Some(v))),
            None => Ok((Default::default(), None)),
        }
    }

    /// Change the retired record by compare-and-swap, reading it again
    /// whenever another writer changed it first. `edit` must give the
    /// same result however often it runs.
    fn update_retired(
        &self,
        edit: impl FnMut(&mut std::collections::BTreeMap<String, u64>),
    ) -> Result<()> {
        self.update_record(RETIRED_KEY, "the record of retired packs", edit)
    }

    /// Change a record of `<wall ms> <key>` lines by compare-and-swap, as
    /// [`Store::update_retired`] does; `what` names it in the error when
    /// other writers kept winning.
    fn update_record(
        &self,
        key: &str,
        what: &str,
        mut edit: impl FnMut(&mut std::collections::BTreeMap<String, u64>),
    ) -> Result<()> {
        for _ in 0..RETIRED_TRIES {
            let (mut m, version) = self.read_record(key)?;
            let before = m.clone();
            edit(&mut m);
            // Unchanged; a record never written stays unwritten.
            if m == before && (version.is_some() || m.is_empty()) {
                return Ok(());
            }
            match self
                .backend
                .write_ref(key, &encode_retired(&m), version.as_ref())
            {
                Ok(_) => return Ok(()),
                Err(BackendError::Conflict(_)) => continue,
                Err(e) => return Err(backend_err(e)),
            }
        }
        Err(StoreError::Io(io::Error::other(format!(
            "other writers kept changing {what}; run the command again"
        ))))
    }

    /// Record `keys` (packs, by their `.pack` key, and loose objects) as
    /// retired now: kept for the readers still using them, and deleted by
    /// a collection at least [`RETIRED_FOR`] later. A key already retired
    /// keeps its first time.
    fn retire(&self, keys: &[String]) -> Result<()> {
        if keys.is_empty() {
            return Ok(());
        }
        let now = now_ms();
        self.update_retired(|m| {
            for k in keys {
                m.entry(k.clone()).or_insert(now);
            }
        })
    }

    /// Delete what was retired at least `older_than` ago (a collection
    /// does, with [`RETIRED_FOR`]): each pack with its index and
    /// dictionary, and each loose object a pack now holds. A retired loose
    /// object no pack holds any more (the pack that had it was pruned, and
    /// the object stored again since) is kept, as an ordinary loose object.
    pub fn expire_retired(
        &self,
        older_than: std::time::Duration,
        progress: Progress,
    ) -> Result<RetiredReport> {
        let (listed, _) = self.list_packs()?;
        let retired = self.retired_in(&listed)?;
        let mut report = RetiredReport::default();
        if retired.is_empty() {
            return Ok(report);
        }
        let now = now_ms();
        let older_than = older_than.as_millis() as u64;
        let due: Vec<(String, u64)> = retired
            .iter()
            .filter(|(_, t)| now.saturating_sub(**t) >= older_than)
            .map(|(k, t)| (k.clone(), *t))
            .collect();
        report.kept = retired.len() - due.len();
        if due.is_empty() {
            return Ok(report);
        }
        progress(&format!("deleting {} retired files", due.len()));
        // Which objects the live packs hold.
        self.reload_packs()?;
        for (key, _) in &due {
            if key.ends_with(".pack") {
                self.remove_pack(key)?;
                report.packs += 1;
            } else if is_entry_list(key) {
                // An entry list a fold replaced (spec §13.7).
                delete(self.backend.as_ref(), key)?;
            } else if let Some(r) = loose_ref(key) {
                if self.find_packed(&r).is_some() {
                    delete(self.backend.as_ref(), key)?;
                    report.loose += 1;
                }
            }
        }
        self.update_retired(|m| {
            for (k, t) in &due {
                if m.get(k) == Some(t) {
                    m.remove(k);
                }
            }
        })?;
        self.reload_packs()?;
        Ok(report)
    }

    /// The loose objects retired and not yet deleted, by key.
    fn retired_loose(&self) -> Result<std::collections::HashSet<String>> {
        let (listed, _) = self.list_packs()?;
        Ok(self
            .retired_in(&listed)?
            .into_keys()
            .filter(|k| k.starts_with("loose/"))
            .collect())
    }

    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; not those retired, whose
    /// objects a pack holds.
    fn loose_objects(&self) -> Result<Vec<(ObjectRef, strata::Entry)>> {
        let retired = self.retired_loose()?;
        let mut out = self.loose_objects_all()?;
        out.retain(|(_, e)| !retired.contains(&e.key));
        Ok(out)
    }

    /// Every loose object, retired or not.
    fn loose_objects_all(&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, trained): (Vec<ObjectRef>, Vec<String>, bool) = {
            let packs = self.packs.borrow();
            (
                packs.iter().flat_map(|p| p.refs.iter().copied()).collect(),
                packs.iter().map(|p| p.key.clone()).collect(),
                packs.first().is_some_and(|p| p.dict.is_some()),
            )
        };
        // Nothing loose and one pack already made by a full repack (only
        // such a pack has a trained dictionary): rewriting it would give
        // the same pack. A hub's nightly gc rewrote andymccall/oxygen's
        // 48 000 objects (37 s under its write lock) every time.
        if loose.is_empty() && keys.len() <= 1 && (level == ZSTD_LEVEL || trained) {
            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();
        let mut gone: Vec<String> = Vec::new();
        // Loose objects. A retired one (its object packed) whose object is
        // unreachable goes with it, uncounted: the pack's copy is counted.
        let retired = self.retired_loose()?;
        let mut unretired: Vec<String> = Vec::new();
        for (r, entry) in self.loose_objects_all()? {
            if keep.contains(&r) {
                continue;
            }
            if retired.contains(&entry.key) {
                if old_enough(entry.modified_ms) {
                    delete(self.backend.as_ref(), &entry.key)?;
                    unretired.push(entry.key);
                }
                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, &mut gone)?;
        }
        if !unretired.is_empty() {
            self.update_retired(|m| {
                for k in &unretired {
                    m.remove(k);
                }
            })?;
        }
        // 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, &mut gone)?;
            }
        }
        self.journal_forget(&gone)?;
        Ok(report)
    }

    /// Everything kept beside an object: its signatures and the seal and
    /// receipt entries under it, in both layouts.
    /// The entry keys deleted are added to `gone`, for the journal.
    fn remove_sidecars(&self, r: &ObjectRef, gone: &mut Vec<String>) -> 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)?;
                gone.push(e.key);
            }
        }
        Ok(())
    }

    /// Take entry keys a collection deleted out of the journal, so the
    /// next complete fold leaves them out (spec §13.7).
    fn journal_forget(&self, gone: &[String]) -> Result<()> {
        if self.local || gone.is_empty() {
            return Ok(());
        }
        if let Some(set) = self.entries.borrow_mut().as_mut() {
            for k in gone {
                set.remove(k);
            }
        }
        self.update_record(JOURNAL_KEY, "the journal of entries", |m| {
            for k in gone {
                m.remove(k);
            }
        })
    }

    /// 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: the index first, so
    /// a store opened meanwhile does not load a pack that is going.
    fn remove_pack(&self, pack_key: &str) -> Result<()> {
        let stem = pack_key.trim_end_matches(".pack");
        delete(self.backend.as_ref(), &format!("{stem}.idx"))?;
        delete(self.backend.as_ref(), pack_key)?;
        delete(self.backend.as_ref(), &format!("{stem}.dict"))
    }

    /// Write `refs` into a new pack, then retire `loose_keys` and
    /// `old_packs`: they stay in the store, unused by stores opened from
    /// now on, until a collection [`RETIRED_FOR`] later deletes them, so a
    /// reader that loaded its packs before goes on reading (oxygenhub#249).
    fn write_pack(
        &self,
        refs: &[ObjectRef],
        loose_keys: &[String],
        old_packs: &[String],
        level: i32,
        progress: Progress,
    ) -> Result<PackReport> {
        if refs.is_empty() {
            self.retire(old_packs)?;
            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)?;
        let mut retired = old_packs.to_vec();
        retired.extend_from_slice(loose_keys);
        self.retire(&retired)?;
        self.reload_packs()?;
        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))
    }

    /// Whether the store holds `r`. A miss looks at the pack listing
    /// again before it is believed, as [`Store::get`]'s does; a writer
    /// asking about many objects it is about to store looks at most once
    /// in [`HAS_RELIST_AFTER`] while nothing new turns up.
    pub fn has(&self, r: &ObjectRef) -> bool {
        if self.has_now(r) {
            return true;
        }
        let recent = self
            .has_relisted
            .get()
            .is_some_and(|t| t.elapsed() < HAS_RELIST_AFTER);
        if recent {
            return false;
        }
        match self.reload_if_changed() {
            Ok(true) => {
                self.has_relisted.set(None);
                self.has_now(r)
            }
            _ => {
                self.has_relisted.set(Some(std::time::Instant::now()));
                false
            }
        }
    }

    /// [`Store::has`] against the packs as loaded.
    fn has_now(&self, r: &ObjectRef) -> bool {
        self.find_packed(r).is_some()
            || self
                .pending
                .borrow()
                .as_ref()
                .is_some_and(|p| p.lookup.contains_key(r))
            || self.backend.has(&loose_key(r)).unwrap_or(false)
    }

    /// 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 self.flush_journal();
        };
        self.write_pending(pending)?;
        self.flush_journal()
    }

    /// 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 self.flush_journal();
        };
        self.write_pending(pending)?;
        self.flush_journal()?;
        *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, packed, loose, or pending in a batch.
    /// The packs' indexes are in memory, so they are asked first: asking
    /// for a loose copy of a packed object costs a request that finds
    /// nothing on a store in a bucket, every time (oxygenhub#228).
    ///
    /// A miss is not believed at once (oxygenhub#249): when the pack it
    /// was in is gone (deleted after a repack retired it) the packs are
    /// loaded again, and when the object is nowhere the pack listing is
    /// looked at, and the packs loaded again if it changed (another
    /// writer packed it, or a repack moved it); then it is read once more.
    pub fn get(&self, r: &ObjectRef) -> Result<Object> {
        match self.get_now(r) {
            Err(e) if is_miss(&e) => {
                // Read from a pack this store loaded, which is gone.
                let packed_miss =
                    !matches!(e, StoreError::NotFound(_)) || self.find_packed(r).is_some();
                let changed = if packed_miss {
                    self.reload_packs()?;
                    true
                } else {
                    self.reload_if_changed()?
                };
                if changed {
                    self.get_now(r)
                } else {
                    Err(e)
                }
            }
            other => other,
        }
    }

    /// [`Store::get`] against the packs as loaded.
    fn get_now(&self, r: &ObjectRef) -> Result<Object> {
        if let Some((i, e)) = self.find_packed(r) {
            let pack = self.packs.borrow()[i].clone();
            return pack.fetch(self.backend.as_ref(), r, e);
        }
        match get_optional(self.backend.as_ref(), &loose_key(r))? {
            Some(bytes) => parse_loose(&bytes, r),
            None => self.pending_get(r).unwrap_or(Err(StoreError::NotFound(*r))),
        }
    }

    /// Many objects at once, read and verified in parallel: packed ones
    /// from their packs, the others loose (as [`Store::get`], packs first),
    /// then what is pending in a batch. Results are in the order of `refs`.
    ///
    /// Misses are looked for again as [`Store::get`]'s are, together.
    pub fn get_many(&self, refs: &[ObjectRef]) -> Vec<Result<Object>> {
        let mut out = self.get_many_now(refs);
        let missed: Vec<usize> = out
            .iter()
            .enumerate()
            .filter(|(_, g)| g.as_ref().is_err_and(is_miss))
            .map(|(i, _)| i)
            .collect();
        if missed.is_empty() {
            return out;
        }
        let packed_miss = missed.iter().any(|i| {
            !matches!(out[*i], Err(StoreError::NotFound(_)))
                || self.find_packed(&refs[*i]).is_some()
        });
        let changed = if packed_miss {
            self.reload_packs().map(|_| true)
        } else {
            self.reload_if_changed()
        };
        match changed {
            Ok(true) => {
                let again: Vec<ObjectRef> = missed.iter().map(|i| refs[*i]).collect();
                for (i, g) in missed.into_iter().zip(self.get_many_now(&again)) {
                    out[i] = g;
                }
            }
            Ok(false) => {}
            Err(e) => {
                let why = e.to_string();
                for i in missed {
                    out[i] = Err(StoreError::Io(io::Error::other(why.clone())));
                }
            }
        }
        out
    }

    /// [`Store::get_many`] against the packs as loaded.
    fn get_many_now(&self, refs: &[ObjectRef]) -> Vec<Result<Object>> {
        use rayon::prelude::*;
        let backend = self.backend.as_ref();
        let packs = self.packs.borrow();
        let place: Vec<Option<(usize, PackEntry)>> = refs
            .iter()
            .map(|r| {
                packs
                    .iter()
                    .enumerate()
                    .find_map(|(p, pack)| pack.find(r).map(|e| (p, e)))
            })
            .collect();
        let indexes: &[Arc<PackIndex>] = &packs;
        let found: Vec<Option<Result<Object>>> = refs
            .par_iter()
            .zip(place.par_iter())
            .map(|(r, at)| match at {
                Some((p, e)) => Some(indexes[*p].fetch(backend, r, *e)),
                None => match get_optional(backend, &loose_key(r)) {
                    Ok(Some(bytes)) => Some(parse_loose(&bytes, r)),
                    Ok(None) => None,
                    Err(e) => Some(Err(e)),
                },
            })
            .collect();
        drop(packs);
        found
            .into_iter()
            .zip(refs)
            .map(|(f, r)| {
                f.unwrap_or_else(|| self.pending_get(r).unwrap_or(Err(StoreError::NotFound(*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 encoding = crate::timing::phase("store: encode the change and its texts");
        let bytes = c.encode_streaming(|batch| self.put_texts(&batch))?;
        drop(encoding);
        let _t = crate::timing::phase("store: put the encoding");
        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<()> {
        debug_assert!(ENTRY_INDEXES.iter().any(|(i, _)| *i == index));
        let entry = format!("{}{}", fanout_prefix(index, key), value.to_hex());
        put_new_or_existing(self.backend.as_ref(), &entry, b"")?;
        self.journal_add(entry)
    }

    /// 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
            .entry_keys(&fanout_prefix(index, key))?
            .into_iter()
            .filter_map(|k| k.rsplit('/').next().and_then(ObjectRef::from_hex))
            .collect();
        if !self.reads_legacy() {
            // Older lists folded into entries (spec §13.7).
        } else 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).
    /// Keep a short-lived key's certificate (spec §14.7), with its issuer's
    /// signature, indexed by the key it certifies.
    pub fn put_certificate(&self, c: &crate::cert::SignedCertificate) -> Result<ObjectRef> {
        let r = self.put(ObjectType::Certificate, &c.cert.encode())?;
        self.put_signature(&r, &c.sig)?;
        self.index_add("cert", &c.cert.key_id(), &r)?;
        Ok(r)
    }

    /// A kept certificate with its issuer's signature.
    pub fn get_certificate(&self, r: &ObjectRef) -> Result<crate::cert::SignedCertificate> {
        let cert = crate::cert::Certificate::decode(&self.get_typed(r, ObjectType::Certificate)?)?;
        let sig = self
            .signatures(r)?
            .into_iter()
            .find(|s| s.key == cert.issuer)
            .ok_or(StoreError::NotFound(*r))?;
        Ok(crate::cert::SignedCertificate { cert, sig })
    }

    /// The certificate kept for short-lived key `key`, if any.
    pub fn certificate_of(&self, key: &ObjectRef) -> Result<Option<ObjectRef>> {
        Ok(self
            .index_get("cert", "certs", key)?
            .into_iter()
            .find(|r| self.has(r)))
    }

    /// Record `value` under `key` in index `index` (for repository code).
    pub(crate) fn index_add_pub(
        &self,
        index: &str,
        key: &ObjectRef,
        value: &ObjectRef,
    ) -> Result<()> {
        self.index_add(index, key, value)
    }

    /// The values recorded under `key` in index `index` (which has no
    /// older one-list form: the legacy name only has to differ from it).
    pub(crate) fn index_get_pub(&self, index: &str, key: &ObjectRef) -> Result<Vec<ObjectRef>> {
        self.index_get(index, &format!("{index}-lists"), key)
    }

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

    pub fn index_approval(&self, state: &ObjectRef, attest: &ObjectRef) -> Result<()> {
        self.index_add("approval", state, attest)
    }

    /// Approval attestations recorded for `state` (OxygenFlow approval
    /// gates, oxygenhub#235).
    pub fn approvals_for(&self, state: &ObjectRef) -> Result<Vec<ObjectRef>> {
        self.index_get("approval", "approvals", state)
    }

    /// Every state with an approval, for garbage collection.
    pub fn approval_states(&self) -> Result<Vec<ObjectRef>> {
        self.indexed_states("approval", "approvals")
    }

    /// Every state with a receipt, for garbage collection.
    pub fn receipt_states(&self) -> Result<Vec<ObjectRef>> {
        self.indexed_states("receipt", "receipts")
    }

    /// The states an index (`one/<hh>/<rest>/<attest>` entries, and the
    /// packed `many/<hh>/<rest>` lists) holds attestations for.
    fn indexed_states(&self, one: &str, many: &str) -> Result<Vec<ObjectRef>> {
        let mut out: Vec<ObjectRef> = Vec::new();
        for key in self.entry_keys(&format!("{one}/"))? {
            // receipt/<hh>/<rest>/<attest>
            let mut parts = 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}")));
        }
        let legacy = if self.reads_legacy() {
            self.backend
                .list(&format!("{many}/"))
                .map_err(backend_err)?
        } else {
            Vec::new()
        };
        for e in legacy {
            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 key in self.entry_keys(&fanout_prefix("sig", r))? {
            // Gone since it was listed or journalled: a collection deleted
            // it with its object, or a new signature by the same signer is
            // replacing it.
            let Some(bytes) = get_optional(self.backend.as_ref(), &key)? else {
                continue;
            };
            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.
        let legacy = if self.reads_legacy() {
            get_optional(self.backend.as_ref(), &legacy_key("sigs", r))?
        } else {
            None
        };
        if let Some(bytes) = legacy {
            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()) {
            delete(self.backend.as_ref(), &key)?;
            put_new_or_existing(self.backend.as_ref(), &key, &bytes)?;
        }
        // Journalled even when the key was there: an older writer's
        // entry is not in the journal or any list.
        self.journal_add(key)
    }

    // ---- the journal and entry lists (spec §13.7, #116)

    /// Whether lookups still read the one-list keys of older stores: until
    /// a complete fold has copied them into per-object entries and the
    /// pack listing shows its marker (spec §13.7).
    fn reads_legacy(&self) -> bool {
        !self.legacy_folded.get()
    }

    /// Copy every older one-list key's contents into per-object entries,
    /// keeping an entry already there, then write the marker that tells
    /// readers to stop reading those keys. The keys themselves are left
    /// for older readers. A list that cannot be read is passed over, and
    /// said.
    fn fold_legacy(&self, progress: Progress) -> Result<()> {
        let mut copied = 0usize;
        for (index, legacy) in ENTRY_INDEXES {
            for e in self
                .backend
                .list(&format!("{legacy}/"))
                .map_err(backend_err)?
            {
                // <legacy>/<hh>/<rest>
                let mut parts = e.key.splitn(3, '/');
                let (Some(_), Some(hh), Some(rest)) = (parts.next(), parts.next(), parts.next())
                else {
                    continue;
                };
                let Some(r) = ObjectRef::from_hex(&format!("{hh}{rest}")) else {
                    continue;
                };
                let Some(bytes) = get_optional(self.backend.as_ref(), &e.key)? else {
                    continue;
                };
                if index == "sig" {
                    let Ok(sigs) = Signature::decode_list(&bytes) else {
                        progress(&format!("{}: not a list of signatures, passed over", e.key));
                        continue;
                    };
                    for sig in sigs {
                        let key = format!("{}{}", fanout_prefix("sig", &r), sig.key.to_hex());
                        let one = Signature::encode_list(std::slice::from_ref(&sig));
                        put_new_or_existing(self.backend.as_ref(), &key, &one)?;
                        copied += 1;
                    }
                } else {
                    for v in String::from_utf8_lossy(&bytes)
                        .lines()
                        .filter_map(ObjectRef::from_hex)
                    {
                        let key = format!("{}{}", fanout_prefix(index, &r), v.to_hex());
                        put_new_or_existing(self.backend.as_ref(), &key, b"")?;
                        copied += 1;
                    }
                }
            }
        }
        if copied > 0 {
            progress(&format!("{copied} entries copied out of older lists"));
        }
        put_new_or_existing(
            self.backend.as_ref(),
            LEGACY_FOLDED_KEY,
            format!("{}\n", now_ms()).as_bytes(),
        )?;
        self.legacy_folded.set(true);
        Ok(())
    }

    /// The entry keys under `prefix` (an index's `<index>/` or one
    /// object's `<index>/<hh>/<rest>/`), sorted: on a folded store from
    /// its lists and journal, with no request; otherwise by listing.
    fn entry_keys(&self, prefix: &str) -> Result<Vec<String>> {
        if let Some(keys) = self.with_entries(|set| {
            set.range(prefix.to_string()..)
                .take_while(|k| k.starts_with(prefix))
                .cloned()
                .collect::<Vec<String>>()
        })? {
            return Ok(keys);
        }
        Ok(self
            .backend
            .list(prefix)
            .map_err(backend_err)?
            .into_iter()
            .map(|e| e.key)
            .collect())
    }

    /// `f` over a folded store's entry keys, read on first use; `None`
    /// when the store is not folded (or is this disk's directory), and
    /// lookups list.
    fn with_entries<T>(
        &self,
        f: impl FnOnce(&std::collections::BTreeSet<String>) -> T,
    ) -> Result<Option<T>> {
        if self.local || !self.folded.get() {
            return Ok(None);
        }
        if self.entries.borrow().is_none() {
            let Some(set) = self.read_entries()? else {
                return Ok(None);
            };
            *self.entries.borrow_mut() = Some(set);
        }
        Ok(self.entries.borrow().as_ref().map(f))
    }

    /// Every live entry list's keys and every journal line's, folded or
    /// not, with what this handle has queued. A list deleted since the
    /// packs were listed (retired over an hour ago) has the packs listed
    /// again; `None` when one is still missing then, or the store is no
    /// longer folded: lookups list.
    fn read_entries(&self) -> Result<Option<std::collections::BTreeSet<String>>> {
        'attempt: for attempt in 0..2 {
            if !self.folded.get() {
                return Ok(None);
            }
            let mut set = std::collections::BTreeSet::new();
            let lists = self.entry_lists.borrow().clone();
            for key in &lists {
                match get_optional(self.backend.as_ref(), key)? {
                    Some(bytes) => set.extend(decode_entry_list(key, &bytes)?),
                    None if attempt == 0 => {
                        self.reload_packs()?;
                        continue 'attempt;
                    }
                    None => return Ok(None),
                }
            }
            set.extend(self.read_record(JOURNAL_KEY)?.0.into_keys());
            set.extend(self.journal_queue.borrow().iter().cloned());
            return Ok(Some(set));
        }
        Ok(None)
    }

    /// Add an entry key this handle wrote to the journal: at once, or
    /// when the open batch is flushed. One the handle already knows from
    /// the lists or the journal is not added again. Nothing on this
    /// disk's directory.
    fn journal_add(&self, key: String) -> Result<()> {
        if self.local {
            return Ok(());
        }
        if let Some(set) = self.entries.borrow_mut().as_mut() {
            if !set.insert(key.clone()) {
                return Ok(());
            }
        }
        if self.batch_depth.get() > 0 {
            self.journal_queue.borrow_mut().push(key);
            return Ok(());
        }
        self.journal_append(std::slice::from_ref(&key))
    }

    /// Add what the open batch queued to the journal.
    fn flush_journal(&self) -> Result<()> {
        let queued = std::mem::take(&mut *self.journal_queue.borrow_mut());
        self.journal_append(&queued)
    }

    /// Add `keys` to the journal, unfolded, in one compare-and-swap
    /// update, dropping lines folded over an hour ago; past
    /// [`JOURNAL_FOLD_AT`] unfolded lines, fold them into a partial list.
    fn journal_append(&self, keys: &[String]) -> Result<()> {
        if keys.is_empty() {
            return Ok(());
        }
        let now = now_ms();
        let mut unfolded = 0;
        self.update_record(JOURNAL_KEY, "the journal of entries", |m| {
            expire_folded(m, now);
            for k in keys {
                m.entry(k.clone()).or_insert(0);
            }
            unfolded = m.values().filter(|t| **t == 0).count();
        })?;
        if unfolded > JOURNAL_FOLD_AT {
            self.fold_journal()?;
        }
        Ok(())
    }

    /// Fold the journal's unfolded lines into a partial entry list, and
    /// mark them folded. Lines added meanwhile stay unfolded.
    fn fold_journal(&self) -> Result<()> {
        let (journal, _) = self.read_record(JOURNAL_KEY)?;
        let keys: Vec<String> = journal
            .into_iter()
            .filter(|(_, t)| *t == 0)
            .map(|(k, _)| k)
            .collect();
        if keys.is_empty() {
            return Ok(());
        }
        self.put_entry_list(PART_SUFFIX, keys.iter())?;
        self.mark_folded(&keys)
    }

    /// Mark these journal lines folded now, if they are not yet; drop
    /// those folded over an hour ago.
    fn mark_folded(&self, keys: &[String]) -> Result<()> {
        let now = now_ms();
        self.update_record(JOURNAL_KEY, "the journal of entries", |m| {
            expire_folded(m, now);
            for k in keys {
                if let Some(t) = m.get_mut(k) {
                    if *t == 0 {
                        *t = now;
                    }
                }
            }
        })
    }

    /// Write a new entry list of `keys` (sorted), complete or partial by
    /// `suffix`; its key.
    fn put_entry_list<'a>(
        &self,
        suffix: &str,
        keys: impl Iterator<Item = &'a String>,
    ) -> Result<String> {
        let key = format!("pack-{}{suffix}", fresh_id());
        let bytes = encode_entry_list(keys)?;
        self.backend.put_new(&key, &bytes).map_err(backend_err)?;
        Ok(key)
    }

    /// Fold every signature and index entry into one complete entry list
    /// (spec §13.7), what a collection does last: first the older
    /// one-list keys into per-object entries, once; then the journal's lines and
    /// a listing of every entry prefix, written as a new list, every list
    /// the packs' listing showed retired, and the journal lines read
    /// marked folded. Nothing when the store is this disk's directory, or
    /// when the one live list already holds exactly these entries and the
    /// journal has nothing unfolded. Whether a list was written.
    pub fn fold_entries(&self, progress: Progress) -> Result<bool> {
        if self.local {
            return Ok(false);
        }
        self.reload_packs()?;
        if !self.legacy_folded.get() {
            progress("folding older lists of signatures and index entries");
            self.fold_legacy(progress)?;
        }
        let (journal, _) = self.read_record(JOURNAL_KEY)?;
        progress("folding signatures and index entries");
        let mut all: std::collections::BTreeSet<String> = journal.keys().cloned().collect();
        for (index, _) in ENTRY_INDEXES {
            for e in self
                .backend
                .list(&format!("{index}/"))
                .map_err(backend_err)?
            {
                all.insert(e.key);
            }
        }
        let lists = self.entry_lists.borrow().clone();
        let unfolded: Vec<String> = journal
            .iter()
            .filter(|(_, t)| **t == 0)
            .map(|(k, _)| k.clone())
            .collect();
        if unfolded.is_empty() && lists.len() == 1 && lists[0].ends_with(LIST_SUFFIX) {
            if let Some(bytes) = get_optional(self.backend.as_ref(), &lists[0])? {
                let held: std::collections::BTreeSet<String> =
                    decode_entry_list(&lists[0], &bytes)?.into_iter().collect();
                if held == all {
                    return Ok(false);
                }
            }
        }
        self.put_entry_list(LIST_SUFFIX, all.iter())?;
        self.retire(&lists)?;
        if !unfolded.is_empty() {
            self.mark_folded(&unfolded)?;
        }
        *self.entries.borrow_mut() = None;
        self.reload_packs()?;
        Ok(true)
    }

    /// Finish packs a writer left unfinished: a batch writes its objects to
    /// `.tmp-ID.pack` and only indexes it at the end, so a process that
    /// stopped part way (an import killed after hours) leaves every object
    /// in a pack nothing can read. Each such pack untouched for `min_age`
    /// is read through: every whole entry checked against its reference,
    /// a torn last one cut off, then indexed as a pack of the store.
    /// Returns the packs and objects adopted.
    pub fn adopt_unfinished_packs(
        &self,
        min_age: std::time::Duration,
        progress: &dyn Fn(&str),
    ) -> Result<(usize, usize)> {
        use std::io::{BufReader, Read};
        let mut packs = 0;
        let mut objects = 0;
        let Ok(dir) = fs::read_dir(&self.scratch) else {
            return Ok((0, 0));
        };
        for entry in dir.flatten() {
            let name = entry.file_name().to_string_lossy().to_string();
            let Some(id) = name
                .strip_prefix(".tmp-")
                .and_then(|n| n.strip_suffix(".pack"))
            else {
                continue;
            };
            let path = entry.path();
            let meta = fs::metadata(&path)?;
            let age = meta
                .modified()
                .ok()
                .and_then(|m| m.elapsed().ok())
                .unwrap_or_default();
            if age < min_age {
                progress(&format!(
                    "{name}: written to {}s ago, left alone",
                    age.as_secs()
                ));
                continue;
            }
            let mut f = BufReader::with_capacity(1 << 20, fs::File::open(&path)?);
            let mut magic = [0u8; 8];
            if f.read_exact(&mut magic).is_err() || &magic != PACK_MAGIC {
                progress(&format!("{name}: not a pack, left alone"));
                continue;
            }
            let mut offset = PACK_MAGIC.len() as u64;
            let mut index: Vec<(ObjectRef, PackEntry)> = Vec::new();
            loop {
                let mut head = [0u8; 42];
                if f.read_exact(&mut head).is_err() {
                    break;
                }
                let Some(ty) = ObjectType::ALL.get(head[0] as usize).copied() else {
                    break;
                };
                let flags = head[1];
                let stored_len = u32::from_le_bytes(head[2..6].try_into().unwrap());
                let raw_len = u32::from_le_bytes(head[6..10].try_into().unwrap());
                let r = ObjectRef(head[10..42].try_into().unwrap());
                let mut stored = vec![0u8; stored_len as usize];
                if flags & FLAG_DICT != 0 || f.read_exact(&mut stored).is_err() {
                    break;
                }
                let e = PackEntry {
                    offset: offset + 42,
                    stored_len,
                    raw_len,
                    ty,
                    flags,
                };
                if decode_entry(&r, e, stored, None).is_err() {
                    break;
                }
                index.push((r, e));
                offset += 42 + stored_len as u64;
                if index.len().is_multiple_of(200_000) {
                    progress(&format!(
                        "{name}: {} objects checked, {} MiB",
                        index.len(),
                        offset >> 20
                    ));
                }
            }
            if index.is_empty() {
                progress(&format!("{name}: no whole object, left alone"));
                continue;
            }
            if offset < meta.len() {
                progress(&format!(
                    "{name}: {} torn bytes at the end cut off",
                    meta.len() - offset
                ));
                fs::OpenOptions::new()
                    .write(true)
                    .open(&path)?
                    .set_len(offset)?;
            }
            index.sort_by_key(|(r, _)| *r);
            index.dedup_by_key(|(r, _)| *r);
            objects += index.len();
            packs += 1;
            progress(&format!("{name}: {} objects indexed", index.len()));
            self.finish_pack_files(id, &path, &index)?;
        }
        Ok((packs, objects))
    }

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

/// Whether a read failed in a way a newer list of packs may cure: the
/// object was not found, or its pack's dictionary is gone.
fn is_miss(e: &StoreError) -> bool {
    match e {
        StoreError::NotFound(_) => true,
        StoreError::Corrupt(_, why) => why == NO_DICT,
        _ => false,
    }
}

/// 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);
        }
        let packs = |dir: &Path| {
            let mut v: Vec<String> = fs::read_dir(dir)
                .unwrap()
                .flatten()
                .map(|e| e.file_name().to_string_lossy().into_owned())
                .filter(|n| n.starts_with("pack-"))
                .collect();
            v.sort();
            v
        };
        // With nothing new, a repack leaves that pack as it is: it would
        // only write the same again.
        let before = packs(dir.path());
        let (report, merged) = again.repack_all(REPACK_ZSTD_LEVEL).unwrap();
        assert_eq!((report.objects, merged), (0, 0));
        assert_eq!(packs(dir.path()), before);
        // One new object: a repack replaces the pack and its dictionary,
        // retiring the old ones, which a collection an hour on deletes.
        let extra = again.put(ObjectType::Chunk, b"one more\n").unwrap();
        again.repack_all(REPACK_ZSTD_LEVEL).unwrap();
        let files = packs(dir.path());
        assert_eq!(
            files.len(),
            7,
            "two of pack, index and dictionary, and the retired record: {files:?}"
        );
        assert_eq!(again.pack_count(), 1, "the old pack is not read");
        assert_eq!(&again.get(&refs[7]).unwrap().1, &bodies[7]);
        assert_eq!(again.get(&extra).unwrap().1, b"one more\n");
        again
            .expire_retired(std::time::Duration::ZERO, &mut silent())
            .unwrap();
        let files = packs(dir.path());
        assert_eq!(files.len(), 4, "one pack, index, dictionary: {files:?}");
        assert!(before
            .iter()
            .filter(|f| *f != RETIRED_KEY)
            .all(|f| !files.contains(f)));
        assert_eq!(&again.get(&refs[8]).unwrap().1, &bodies[8]);
    }

    #[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(_))));
        });
    }

    /// A packed object is read from its pack without first asking the
    /// backend for a loose copy (oxygenhub#228: on a bucket each such ask
    /// was a request that found nothing, every time).
    #[test]
    fn packed_objects_cost_no_loose_lookups() {
        use std::sync::atomic::{AtomicUsize, Ordering};
        struct Counting(MemoryStore, AtomicUsize);
        impl crate::strata::Store for Counting {
            fn describe(&self) -> String {
                "counting".into()
            }
            fn has(&self, k: &str) -> crate::strata::Result<bool> {
                self.0.has(k)
            }
            fn get(&self, k: &str) -> crate::strata::Result<Vec<u8>> {
                if k.starts_with("loose/") {
                    self.1.fetch_add(1, Ordering::Relaxed);
                }
                self.0.get(k)
            }
            fn get_range(
                &self,
                k: &str,
                r: std::ops::Range<u64>,
            ) -> crate::strata::Result<Vec<u8>> {
                self.0.get_range(k, r)
            }
            fn put_new(&self, k: &str, b: &[u8]) -> crate::strata::Result<()> {
                self.0.put_new(k, b)
            }
            fn list(&self, p: &str) -> crate::strata::Result<Vec<crate::strata::Entry>> {
                self.0.list(p)
            }
            fn delete(&self, k: &str) -> crate::strata::Result<()> {
                self.0.delete(k)
            }
            fn read_ref(
                &self,
                k: &str,
            ) -> crate::strata::Result<Option<(Vec<u8>, crate::strata::Version)>> {
                self.0.read_ref(k)
            }
            fn write_ref(
                &self,
                k: &str,
                b: &[u8],
                e: Option<&crate::strata::Version>,
            ) -> crate::strata::Result<crate::strata::Version> {
                self.0.write_ref(k, b, e)
            }
        }
        let dir = tempfile::tempdir().unwrap();
        let backend = Arc::new(Counting(MemoryStore::new(), AtomicUsize::new(0)));
        let s = Store::open_on(backend.clone(), dir.path().join("scratch")).unwrap();
        let refs: Vec<ObjectRef> = (0..10)
            .map(|i| {
                s.put(ObjectType::Chunk, format!("p {i}").as_bytes())
                    .unwrap()
            })
            .collect();
        s.pack_loose().unwrap();
        let s = Store::open_on(backend.clone(), dir.path().join("scratch")).unwrap();
        backend.1.store(0, Ordering::Relaxed);
        for r in &refs {
            assert!(s.get(r).is_ok());
        }
        assert!(s.get_many(&refs).iter().all(|g| g.is_ok()));
        assert_eq!(backend.1.load(Ordering::Relaxed), 0, "no loose lookups");
        // An object no pack holds is still looked for loose.
        let missing = hash_object(ObjectType::Chunk, b"never stored");
        assert!(matches!(s.get(&missing), Err(StoreError::NotFound(_))));
        assert!(backend.1.load(Ordering::Relaxed) > 0);
    }
}

/// Readers never wait (oxygenhub#249): a store keeps reading while other
/// handles repack, collect and push into the store under it.
#[cfg(test)]
pub(crate) mod reader_tests {
    use super::*;

    fn chunk(i: usize) -> Vec<u8> {
        format!("object {i} ").repeat(1 + i % 5).into_bytes()
    }

    /// Two handles on one store: a writer, and a reader opened at once.
    fn each_pair(f: impl Fn(&str, Store, Store)) {
        let dir = tempfile::tempdir().unwrap();
        let s = Store::open(dir.path().join("objects")).unwrap();
        let r = Store::open(dir.path().join("objects")).unwrap();
        f("directory", s, r);
        let m: Arc<Backend> = Arc::new(MemoryStore::new());
        let s = Store::open_on(m.clone(), dir.path().join("scratch")).unwrap();
        let r = Store::open_on(m, dir.path().join("scratch-r")).unwrap();
        f("memory", s, r);
    }

    /// Set every retired entry's time back by `by`, as if retired then.
    pub(crate) fn age_retired(s: &Store, by: std::time::Duration) {
        s.update_retired(|m| {
            for t in m.values_mut() {
                *t = t.saturating_sub(by.as_millis() as u64);
            }
        })
        .unwrap();
    }

    fn retired_keys(s: &Store) -> Vec<String> {
        s.read_retired().unwrap().0.into_keys().collect()
    }

    #[test]
    fn a_reader_opened_before_a_repack_reads_every_object_after_it() {
        each_pair(|which, writer, _| {
            let mut refs = Vec::new();
            for i in 0..40 {
                refs.push(writer.put(ObjectType::Chunk, &chunk(i)).unwrap());
            }
            writer.pack_loose().unwrap();
            for i in 40..80 {
                refs.push(writer.put(ObjectType::Chunk, &chunk(i)).unwrap());
            }
            // Opened now: one pack and forty loose objects.
            let reader = Store::open_on(writer.backend().clone(), writer.dir()).unwrap();
            assert_eq!(reader.pack_count(), 1, "{which}");
            let (_, merged) = writer.repack_all(REPACK_ZSTD_LEVEL).unwrap();
            assert_eq!(merged, 1);
            // The old pack and the loose objects are retired, not deleted.
            let retired = retired_keys(&writer);
            assert_eq!(retired.len(), 81, "{which}: {retired:?}");
            assert_eq!(writer.loose_count().unwrap(), 0, "retired is not loose");
            // The reader reads on from what it loaded.
            for (i, r) in refs.iter().enumerate() {
                assert_eq!(reader.get(r).unwrap().1, chunk(i), "{which}: {i}");
            }
            assert_eq!(reader.pack_count(), 1);
            // A store opened now reads the new pack only.
            let fresh = Store::open_on(writer.backend().clone(), writer.dir()).unwrap();
            assert_eq!(fresh.pack_count(), 1);
            assert_ne!(fresh.packs.borrow()[0].key, reader.packs.borrow()[0].key);
            // Once the retired files are deleted, the reader finds every
            // object in the new pack, by get, get_many and has alike.
            let report = writer
                .expire_retired(std::time::Duration::ZERO, &mut silent())
                .unwrap();
            assert_eq!((report.packs, report.loose), (1, 80), "{which}");
            assert!(retired_keys(&writer).is_empty());
            for (i, r) in refs.iter().enumerate() {
                assert_eq!(reader.get(r).unwrap().1, chunk(i), "{which}: {i}");
            }
            let other = Store::open_on(writer.backend().clone(), writer.dir()).unwrap();
            let before = other.packs.borrow()[0].key.clone();
            writer.put(ObjectType::Chunk, b"new").unwrap();
            writer.repack_all(REPACK_ZSTD_LEVEL).unwrap();
            writer
                .expire_retired(std::time::Duration::ZERO, &mut silent())
                .unwrap();
            assert!(other.get_many(&refs).iter().all(|g| g.is_ok()), "{which}");
            assert_ne!(other.packs.borrow()[0].key, before);
            let third = Store::open_on(writer.backend().clone(), writer.dir()).unwrap();
            writer.put(ObjectType::Chunk, b"newer").unwrap();
            writer.repack_all(REPACK_ZSTD_LEVEL).unwrap();
            writer
                .expire_retired(std::time::Duration::ZERO, &mut silent())
                .unwrap();
            assert!(refs.iter().all(|r| third.has(r)), "{which}");
            assert!(!third.has(&hash_object(ObjectType::Chunk, b"never")));
        });
    }

    #[test]
    fn a_reader_opened_before_a_spilled_batch_reads_its_objects() {
        each_pair(|which, writer, reader| {
            assert_eq!(reader.pack_count(), 0);
            let mut refs = Vec::new();
            {
                let batch = writer.begin_batch().unwrap();
                for i in 0..(SPILL_OBJECTS + 50) {
                    refs.push(writer.put(ObjectType::Chunk, &chunk(i)).unwrap());
                }
                batch.finish().unwrap();
            }
            assert_eq!(writer.pack_count(), 1, "{which}: the batch spilled");
            assert_eq!(reader.get(&refs[0]).unwrap().1, chunk(0), "{which}");
            assert_eq!(reader.pack_count(), 1, "the miss loaded the new pack");
            let got = reader.get_many(&refs);
            assert!(got.iter().all(|g| g.is_ok()), "{which}");
            // has, too, from another reader.
            let late = Store::open_on(writer.backend().clone(), writer.dir()).unwrap();
            {
                let batch = writer.begin_batch().unwrap();
                for i in 1000..(1000 + SPILL_OBJECTS + 1) {
                    writer.put(ObjectType::Chunk, &chunk(i)).unwrap();
                }
                batch.finish().unwrap();
            }
            assert!(late.has(&hash_object(ObjectType::Chunk, &chunk(1000))));
        });
    }

    #[test]
    fn retired_entries_expire_after_the_hour_and_are_never_loaded() {
        each_pair(|which, s, _| {
            for i in 0..10 {
                s.put(ObjectType::Chunk, &chunk(i)).unwrap();
            }
            s.pack_loose().unwrap();
            s.put(ObjectType::Chunk, &chunk(10)).unwrap();
            s.repack_all(ZSTD_LEVEL).unwrap();
            // Retired over an hour ago: the first pack and 10 loose objects.
            age_retired(&s, RETIRED_FOR + std::time::Duration::from_secs(60));
            let old = retired_keys(&s);
            assert_eq!(old.len(), 12, "{which}: {old:?}");
            // Retired just now: the second pack.
            s.put(ObjectType::Chunk, &chunk(11)).unwrap();
            s.repack_all(ZSTD_LEVEL).unwrap();
            assert_eq!(retired_keys(&s).len(), 14);
            let listed = s
                .backend()
                .list("pack-")
                .unwrap()
                .into_iter()
                .filter(|e| e.key.ends_with(".idx"))
                .count();
            assert_eq!(listed, 3, "{which}");
            let fresh = Store::open_on(s.backend().clone(), s.dir()).unwrap();
            assert_eq!(
                fresh.pack_count(),
                1,
                "{which}: retired packs are not loaded"
            );
            let report = s.expire_retired(RETIRED_FOR, &mut silent()).unwrap();
            assert_eq!(
                report,
                RetiredReport {
                    packs: 1,
                    loose: 11,
                    kept: 2
                },
                "{which}"
            );
            let young = retired_keys(&s);
            assert_eq!(young.len(), 2, "{which}: {young:?}");
            assert!(young.iter().all(|k| !old.contains(k)));
            for k in old.iter().filter(|k| k.ends_with(".pack")) {
                assert!(!s.backend().has(k).unwrap(), "{which}: {k}");
            }
            for k in &young {
                assert!(s.backend().has(k).unwrap(), "{which}: {k} kept");
            }
            for i in 0..12 {
                let r = hash_object(ObjectType::Chunk, &chunk(i));
                assert_eq!(fresh.get(&r).unwrap().1, chunk(i), "{which}");
            }
        });
    }

    /// A retired loose object whose pack copy was pruned since is not
    /// lost when the retired files go: it is loose again.
    #[test]
    fn a_retired_loose_object_no_pack_holds_is_kept() {
        each_pair(|which, s, _| {
            let a = s.put(ObjectType::Chunk, b"a").unwrap();
            let b = s.put(ObjectType::Chunk, b"b").unwrap();
            s.pack_loose().unwrap();
            assert_eq!(s.loose_count().unwrap(), 0);
            // The pack rewritten without b, as pruning does.
            let first = s.packs.borrow()[0].key.clone();
            s.write_pack(&[a], &[], &[first], ZSTD_LEVEL, &mut silent())
                .unwrap();
            let report = s
                .expire_retired(std::time::Duration::ZERO, &mut silent())
                .unwrap();
            assert_eq!((report.packs, report.loose), (1, 1), "{which}: a's copy");
            assert!(retired_keys(&s).is_empty());
            assert_eq!(s.get(&b).unwrap().1, b"b", "{which}");
            assert_eq!(s.get(&a).unwrap().1, b"a");
            assert_eq!(s.loose_count().unwrap(), 1, "b is loose again");
            // And pruning it, unreachable, deletes it.
            let keep = [a].into_iter().collect();
            let r = s.prune(&keep, std::time::Duration::ZERO).unwrap();
            assert_eq!(r.objects, 1);
            assert!(!s.has(&b));
        });
    }

    /// A backend whose first pack listing names an index that is deleted
    /// before it is read, as a collection running meanwhile would.
    pub(crate) struct Racing {
        inner: MemoryStore,
        lists: std::sync::atomic::AtomicUsize,
        /// Deleted right after the first listing returns.
        pub(crate) doomed: std::sync::Mutex<Option<String>>,
        /// Named in every listing, never stored.
        ghost: bool,
    }

    impl crate::strata::Store for Racing {
        fn describe(&self) -> String {
            "racing".into()
        }
        fn has(&self, k: &str) -> crate::strata::Result<bool> {
            self.inner.has(k)
        }
        fn get(&self, k: &str) -> crate::strata::Result<Vec<u8>> {
            self.inner.get(k)
        }
        fn get_range(&self, k: &str, r: std::ops::Range<u64>) -> crate::strata::Result<Vec<u8>> {
            self.inner.get_range(k, r)
        }
        fn put_new(&self, k: &str, b: &[u8]) -> crate::strata::Result<()> {
            self.inner.put_new(k, b)
        }
        fn list(&self, p: &str) -> crate::strata::Result<Vec<crate::strata::Entry>> {
            let mut out = self.inner.list(p)?;
            if p == "pack-" {
                self.lists
                    .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
                if let Some(k) = self.doomed.lock().unwrap().take() {
                    self.inner.delete(&k)?;
                }
                if self.ghost {
                    for k in ["pack-0000.idx", "pack-0000.pack"] {
                        out.push(crate::strata::Entry {
                            key: k.into(),
                            size: 100,
                            modified_ms: None,
                        });
                    }
                    out.sort_by(|a, b| a.key.cmp(&b.key));
                }
            }
            Ok(out)
        }
        fn delete(&self, k: &str) -> crate::strata::Result<()> {
            self.inner.delete(k)
        }
        fn read_ref(
            &self,
            k: &str,
        ) -> crate::strata::Result<Option<(Vec<u8>, crate::strata::Version)>> {
            self.inner.read_ref(k)
        }
        fn write_ref(
            &self,
            k: &str,
            b: &[u8],
            e: Option<&crate::strata::Version>,
        ) -> crate::strata::Result<crate::strata::Version> {
            self.inner.write_ref(k, b, e)
        }
    }

    pub(crate) fn racing(ghost: bool) -> Arc<Racing> {
        Arc::new(Racing {
            inner: MemoryStore::new(),
            lists: Default::default(),
            doomed: Default::default(),
            ghost,
        })
    }

    #[test]
    fn an_index_deleted_between_listing_and_loading_is_passed_over() {
        let dir = tempfile::tempdir().unwrap();
        let backend = racing(false);
        let s = Store::open_on(backend.clone(), dir.path()).unwrap();
        let a = s.put(ObjectType::Chunk, b"a").unwrap();
        s.pack_loose().unwrap();
        let b = s.put(ObjectType::Chunk, b"b").unwrap();
        s.pack_loose().unwrap();
        let idx = {
            let packs = s.packs.borrow();
            let (i, _) = s.find_packed(&a).unwrap();
            format!("{}.idx", packs[i].key.trim_end_matches(".pack"))
        };
        *backend.doomed.lock().unwrap() = Some(idx);
        let before = backend.lists.load(std::sync::atomic::Ordering::Relaxed);
        let again = Store::open_on(backend.clone(), dir.path()).unwrap();
        assert_eq!(again.pack_count(), 1, "the deleted one is passed over");
        assert_eq!(
            backend.lists.load(std::sync::atomic::Ordering::Relaxed) - before,
            2,
            "and the store listed again"
        );
        assert_eq!(again.get(&b).unwrap().1, b"b");
        // A listing that names an index never there opens too.
        let ghosted = racing(true);
        let s = Store::open_on(ghosted.clone(), dir.path().join("g")).unwrap();
        let c = s.put(ObjectType::Chunk, b"c").unwrap();
        s.pack_loose().unwrap();
        let again = Store::open_on(ghosted, dir.path().join("g")).unwrap();
        assert_eq!(again.pack_count(), 1);
        assert_eq!(again.get(&c).unwrap().1, b"c");
    }
}

#[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)));
        }
    }

    #[test]
    fn a_pack_a_stopped_batch_left_is_adopted_and_its_torn_end_cut() {
        // oxygen#79: a batch never finished (a process killed part way)
        // leaves its objects in an unindexed .tmp pack.
        let dir = tempfile::tempdir().unwrap();
        let mut refs = Vec::new();
        {
            let store = Store::open(dir.path()).unwrap();
            let batch = store.begin_batch().unwrap();
            for i in 0..300u32 {
                let body = format!("object {i} {}", "y".repeat(20_000 + i as usize)).into_bytes();
                refs.push(store.put(ObjectType::Blob, &body).unwrap());
            }
            std::mem::forget(batch);
        }
        let tmp: Vec<_> = fs::read_dir(dir.path())
            .unwrap()
            .flatten()
            .filter(|e| e.file_name().to_string_lossy().starts_with(".tmp-"))
            .map(|e| e.path())
            .collect();
        assert_eq!(tmp.len(), 1, "one unfinished pack");
        // A torn last entry: a header and part of its body.
        let mut f = fs::OpenOptions::new().append(true).open(&tmp[0]).unwrap();
        f.write_all(&[0, 1, 0xff, 0, 0, 0, 9, 9, 0, 0]).unwrap();
        f.write_all(&[7; 40]).unwrap();
        drop(f);
        let store = Store::open(dir.path()).unwrap();
        assert!(store.get(&refs[0]).is_err(), "unreadable before");
        let young = store
            .adopt_unfinished_packs(std::time::Duration::from_secs(3600), &|_| {})
            .unwrap();
        assert_eq!(young, (0, 0), "a pack written to lately is left alone");
        let (packs, objects) = store
            .adopt_unfinished_packs(std::time::Duration::ZERO, &|_| {})
            .unwrap();
        assert_eq!((packs, objects), (1, 300));
        for r in &refs {
            store
                .get(r)
                .unwrap_or_else(|e| panic!("{}: {e}", r.short(12)));
        }
        let again = Store::open(dir.path()).unwrap();
        assert!(again.get(&refs[299]).is_ok(), "indexed for good");
    }
}

#[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);
        }
    }
}

/// Signatures and index entries on a store that is not this disk's
/// directory: the journal and the entry lists (spec §13.7, #116).
#[cfg(test)]
mod entry_tests {
    use super::*;
    use crate::identity::{Domain, SecretKey};
    use crate::strata::Store as _;
    use std::collections::HashMap;
    use std::sync::atomic::{AtomicBool, Ordering};
    use std::sync::Mutex;

    /// A store in memory that counts its requests by kind, as a bucket
    /// store's /health counters do; it can also slip a competing journal
    /// line in just before a write of the journal, as another process
    /// would.
    #[derive(Default)]
    struct Counted {
        inner: MemoryStore,
        counts: Mutex<HashMap<&'static str, usize>>,
        interfere: AtomicBool,
    }

    impl Counted {
        fn count(&self, kind: &'static str) {
            *self.counts.lock().unwrap().entry(kind).or_default() += 1;
        }
        fn n(&self, kind: &str) -> usize {
            self.counts.lock().unwrap().get(kind).copied().unwrap_or(0)
        }
        fn reset(&self) {
            self.counts.lock().unwrap().clear();
        }
        fn total(&self) -> usize {
            self.counts.lock().unwrap().values().sum()
        }
        fn summary(&self) -> String {
            let c = self.counts.lock().unwrap();
            let mut v: Vec<String> = c.iter().map(|(k, n)| format!("{k} {n}")).collect();
            v.sort();
            v.join(", ")
        }
    }

    impl crate::strata::Store for Counted {
        fn describe(&self) -> String {
            "counted".into()
        }
        fn has(&self, k: &str) -> crate::strata::Result<bool> {
            self.count("has");
            self.inner.has(k)
        }
        fn get(&self, k: &str) -> crate::strata::Result<Vec<u8>> {
            self.count(if is_legacy(k) { "get_legacy" } else { "get" });
            self.inner.get(k)
        }
        fn get_range(&self, k: &str, r: std::ops::Range<u64>) -> crate::strata::Result<Vec<u8>> {
            self.count("range");
            self.inner.get_range(k, r)
        }
        fn put_new(&self, k: &str, b: &[u8]) -> crate::strata::Result<()> {
            self.count("put");
            self.inner.put_new(k, b)
        }
        fn list(&self, p: &str) -> crate::strata::Result<Vec<crate::strata::Entry>> {
            self.count(if p == "pack-" {
                "list_packs"
            } else if is_legacy(p) {
                "list_legacy"
            } else {
                "list"
            });
            self.inner.list(p)
        }
        fn delete(&self, k: &str) -> crate::strata::Result<()> {
            self.count("delete");
            self.inner.delete(k)
        }
        fn read_ref(
            &self,
            k: &str,
        ) -> crate::strata::Result<Option<(Vec<u8>, crate::strata::Version)>> {
            self.count(if k == JOURNAL_KEY {
                "read_journal"
            } else {
                "read_ref"
            });
            self.inner.read_ref(k)
        }
        fn write_ref(
            &self,
            k: &str,
            b: &[u8],
            e: Option<&crate::strata::Version>,
        ) -> crate::strata::Result<crate::strata::Version> {
            self.count("write_ref");
            if k == JOURNAL_KEY && self.interfere.swap(false, Ordering::SeqCst) {
                let (mut m, v) = match self.inner.read_ref(k)? {
                    Some((bytes, v)) => (parse_retired(&bytes), Some(v)),
                    None => (Default::default(), None),
                };
                m.insert(
                    format!("seal/{}", "c".repeat(64)) + "/" + &"d".repeat(64),
                    0,
                );
                self.inner.write_ref(k, &encode_retired(&m), v.as_ref())?;
            }
            self.inner.write_ref(k, b, e)
        }
    }

    /// Whether a key or prefix is under an older one-list layout.
    fn is_legacy(k: &str) -> bool {
        ENTRY_INDEXES
            .iter()
            .any(|(_, legacy)| k.starts_with(&format!("{legacy}/")))
    }

    struct Fixture {
        backend: Arc<Counted>,
        dir: tempfile::TempDir,
        handles: std::cell::Cell<usize>,
    }

    impl Fixture {
        fn new() -> Fixture {
            Fixture {
                backend: Arc::new(Counted::default()),
                dir: tempfile::tempdir().unwrap(),
                handles: Default::default(),
            }
        }
        /// Another handle on the store, as another process opens it.
        fn open(&self) -> Store {
            let n = self.handles.get() + 1;
            self.handles.set(n);
            Store::open_on(self.backend.clone(), self.dir.path().join(format!("s{n}"))).unwrap()
        }
    }

    fn signer(n: u8) -> SecretKey {
        SecretKey {
            ed_seed: [n; 32],
            ml_seed: [n + 100; 32],
        }
    }

    fn sign(s: &Store, r: &ObjectRef, n: u8) -> Signature {
        let sk = signer(n);
        let key = s.put_identity(&sk.identity("a", "b", 0)).unwrap();
        let sig = sk.sign(Domain::Change, &key, r);
        s.put_signature(r, &sig).unwrap();
        sig
    }

    fn journal(s: &Store) -> std::collections::BTreeMap<String, u64> {
        s.read_record(JOURNAL_KEY).unwrap().0
    }

    fn fold(s: &Store) -> bool {
        s.fold_entries(&mut silent()).unwrap()
    }

    /// What 0.4.6 writes for a signature: its own key, and nothing else.
    fn old_writer_signs(f: &Fixture, r: &ObjectRef, n: u8) -> Signature {
        let sk = signer(n);
        let sig = sk.sign(Domain::Change, &ObjectRef([n; 32]), r);
        f.backend
            .put_new(
                &format!("{}{}", fanout_prefix("sig", r), sig.key.to_hex()),
                &Signature::encode_list(std::slice::from_ref(&sig)),
            )
            .unwrap();
        sig
    }

    /// After a fold, a signature or seal another handle writes is found
    /// by a store opened later through the journal, with no listing; a
    /// handle that already read the entries sees it once refreshed.
    #[test]
    fn an_entry_written_after_a_fold_is_found_through_the_journal() {
        let f = Fixture::new();
        let w = f.open();
        let r1 = w.put(ObjectType::Chunk, b"one").unwrap();
        let s1 = sign(&w, &r1, 1);
        assert!(fold(&w));
        assert!(!fold(&w), "nothing new: no second list");

        let early = f.open();
        assert_eq!(early.signatures(&r1).unwrap(), vec![s1.clone()]);

        let w2 = f.open();
        let r2 = w2.put(ObjectType::Chunk, b"two").unwrap();
        let s2 = sign(&w2, &r2, 2);
        w2.index_seal(&r2, &ObjectRef([9; 32])).unwrap();
        w2.index_receipt(&r2, &ObjectRef([8; 32])).unwrap();
        // Its own writes, at once.
        assert_eq!(w2.signatures(&r2).unwrap(), vec![s2.clone()]);

        let later = f.open();
        f.backend.reset();
        assert_eq!(later.signatures(&r1).unwrap(), vec![s1]);
        assert_eq!(later.signatures(&r2).unwrap(), vec![s2.clone()]);
        assert_eq!(later.seals_for(&r2).unwrap(), vec![ObjectRef([9; 32])]);
        assert_eq!(later.receipts_for(&r2).unwrap(), vec![ObjectRef([8; 32])]);
        assert_eq!(later.receipt_states().unwrap(), vec![r2]);
        assert!(later.seals_for(&r1).unwrap().is_empty());
        assert_eq!(f.backend.n("list"), 0, "{}", f.backend.summary());
        assert_eq!(f.backend.n("read_journal"), 1, "{}", f.backend.summary());

        // The handle that read the entries before keeps its view until
        // it is refreshed, as it keeps its operation head.
        assert!(early.signatures(&r2).unwrap().is_empty());
        early.refresh().unwrap();
        assert_eq!(early.signatures(&r2).unwrap(), vec![s2]);
    }

    /// A fold marks the journal's lines rather than dropping them, so a
    /// reader whose listing predates it still finds every entry; an hour
    /// on, the old list is deleted and the lines dropped, and readers
    /// find everything in the new list.
    #[test]
    fn a_fold_keeps_the_journal_for_the_hour_readers_may_need_it() {
        let f = Fixture::new();
        let w = f.open();
        let r1 = w.put(ObjectType::Chunk, b"one").unwrap();
        let s1 = sign(&w, &r1, 1);
        assert!(fold(&w));
        let r2 = w.put(ObjectType::Chunk, b"two").unwrap();
        let s2 = sign(&w, &r2, 2);
        assert!(journal(&w).values().any(|t| *t == 0));

        // Opened before the second fold: lists the first list only.
        let stale = f.open();
        let stale_after_expiry = f.open();
        assert!(fold(&w));
        assert!(
            journal(&w).values().all(|t| *t != 0),
            "every line is marked folded"
        );
        assert!(journal(&w).len() >= 2, "and kept");
        assert_eq!(stale.signatures(&r1).unwrap(), vec![s1.clone()]);
        assert_eq!(stale.signatures(&r2).unwrap(), vec![s2.clone()]);

        // An hour later: the first list goes with the next collection,
        // and the next journal update drops the folded lines.
        reader_tests::age_retired(&w, RETIRED_FOR * 2);
        w.update_record(JOURNAL_KEY, "test", |m| {
            for t in m.values_mut() {
                *t = t.saturating_sub(2 * RETIRED_FOR.as_millis() as u64);
            }
        })
        .unwrap();
        w.expire_retired(RETIRED_FOR, &mut silent()).unwrap();
        let lists: Vec<String> = f
            .backend
            .inner
            .list("pack-")
            .unwrap()
            .into_iter()
            .map(|e| e.key)
            .filter(|k| is_entry_list(k))
            .collect();
        assert_eq!(lists.len(), 1, "the replaced list is deleted: {lists:?}");
        let r3 = w.put(ObjectType::Chunk, b"three").unwrap();
        let s3 = sign(&w, &r3, 3);
        let j = journal(&w);
        assert_eq!(j.len(), 1, "only the new line is left: {j:?}");
        assert!(j.keys().all(|k| k.starts_with(&fanout_prefix("sig", &r3))));

        // A handle whose listing named the deleted list lists again.
        assert_eq!(
            stale_after_expiry.signatures(&r1).unwrap(),
            vec![s1.clone()]
        );
        assert_eq!(
            stale_after_expiry.signatures(&r3).unwrap(),
            vec![s3.clone()]
        );
        let fresh = f.open();
        f.backend.reset();
        assert_eq!(fresh.signatures(&r1).unwrap(), vec![s1]);
        assert_eq!(fresh.signatures(&r2).unwrap(), vec![s2]);
        assert_eq!(fresh.signatures(&r3).unwrap(), vec![s3]);
        assert_eq!(f.backend.n("list"), 0);
    }

    /// A store no fold has touched (as 0.4.6 left it: per-object keys and
    /// older one-list keys, no journal, no lists) reads by listing, and
    /// reads the same once folded. An older writer's entry after the
    /// fold is missed until the next fold: the trade-off of §13.7.
    #[test]
    fn an_old_layout_store_reads_before_and_after_its_first_fold() {
        let f = Fixture::new();
        let w = f.open();
        let r1 = w.put(ObjectType::Chunk, b"one").unwrap();
        let r2 = w.put(ObjectType::Chunk, b"two").unwrap();
        let s1 = old_writer_signs(&f, &r1, 1);
        let s2 = SecretKey {
            ed_seed: [2; 32],
            ml_seed: [3; 32],
        }
        .sign(Domain::Change, &ObjectRef([2; 32]), &r2);
        f.backend
            .put_new(
                &legacy_key("sigs", &r2),
                &Signature::encode_list(std::slice::from_ref(&s2)),
            )
            .unwrap();
        f.backend
            .put_new(
                &format!("{}{}", fanout_prefix("seal", &r1), "ab".repeat(32)),
                b"",
            )
            .unwrap();
        f.backend
            .put_new(
                &legacy_key("receipts", &r2),
                format!("{}\n", ObjectRef([7; 32]).to_hex()).as_bytes(),
            )
            .unwrap();
        let check = |s: &Store| {
            assert_eq!(s.signatures(&r1).unwrap(), vec![s1.clone()]);
            assert_eq!(s.signatures(&r2).unwrap(), vec![s2.clone()]);
            assert_eq!(s.seals_for(&r1).unwrap(), vec![ObjectRef([0xab; 32])]);
            assert_eq!(s.receipts_for(&r2).unwrap(), vec![ObjectRef([7; 32])]);
            assert_eq!(s.receipt_states().unwrap(), vec![r2]);
        };
        let reader = f.open();
        assert!(!reader.folded.get());
        check(&reader);
        assert!(f.backend.inner.read_ref(JOURNAL_KEY).unwrap().is_none());

        assert!(fold(&w));
        let reader = f.open();
        assert!(reader.folded.get());
        check(&reader);

        // The older lists' entries are per-object entries now, and the
        // marker is there.
        assert!(f.backend.inner.has(LEGACY_FOLDED_KEY).unwrap());
        assert!(f
            .backend
            .inner
            .has(&format!("{}{}", fanout_prefix("sig", &r2), s2.key.to_hex()))
            .unwrap());
        assert!(f
            .backend
            .inner
            .has(&format!(
                "{}{}",
                fanout_prefix("receipt", &r2),
                ObjectRef([7; 32]).to_hex()
            ))
            .unwrap());

        // An older writer after the fold: missed until the next one.
        let r3 = w.put(ObjectType::Chunk, b"three").unwrap();
        let s3 = old_writer_signs(&f, &r3, 3);
        assert!(f.open().signatures(&r3).unwrap().is_empty());
        assert!(fold(&w));
        assert_eq!(f.open().signatures(&r3).unwrap(), vec![s3]);
    }

    /// Writers racing on the journal never lose a line: a lost
    /// compare-and-swap reads the journal again and retries.
    #[test]
    fn concurrent_journal_appends_lose_nothing() {
        let f = Fixture::new();
        // One race, made to happen.
        let w = f.open();
        f.backend.interfere.store(true, Ordering::SeqCst);
        w.index_seal(&ObjectRef([1; 32]), &ObjectRef([2; 32]))
            .unwrap();
        let j = journal(&w);
        assert_eq!(j.len(), 2, "the competitor's line and ours: {j:?}");
        // Many, from threads, each its own handle.
        let backend: Arc<Backend> = f.backend.clone();
        std::thread::scope(|scope| {
            for t in 0..8u8 {
                let backend = backend.clone();
                let dir = f.dir.path().join(format!("t{t}"));
                scope.spawn(move || {
                    let s = Store::open_on(backend, dir).unwrap();
                    for i in 0..25u8 {
                        s.index_seal(&ObjectRef([t + 10; 32]), &ObjectRef([i; 32]))
                            .unwrap();
                    }
                });
            }
        });
        assert_eq!(journal(&w).len(), 2 + 8 * 25);
        assert!(fold(&w));
        let r = f.open();
        for t in 0..8u8 {
            assert_eq!(r.seals_for(&ObjectRef([t + 10; 32])).unwrap().len(), 25);
        }
    }

    /// A batch adds its entries to the journal in one update when it is
    /// flushed; past the limit the writer folds them into a partial list.
    #[test]
    fn a_long_journal_is_folded_into_a_partial_list() {
        let f = Fixture::new();
        let w = f.open();
        assert!(fold(&w), "folded, with nothing yet");
        let key = ObjectRef([5; 32]);
        f.backend.reset();
        let batch = w.begin_batch().unwrap();
        for i in 0..=JOURNAL_FOLD_AT as u32 {
            let mut v = [0u8; 32];
            v[..4].copy_from_slice(&i.to_le_bytes());
            w.index_seal(&key, &ObjectRef(v)).unwrap();
        }
        assert_eq!(f.backend.n("write_ref"), 0, "nothing until the flush");
        batch.finish().unwrap();
        let parts = f
            .backend
            .inner
            .list("pack-")
            .unwrap()
            .into_iter()
            .filter(|e| e.key.ends_with(PART_SUFFIX))
            .count();
        assert_eq!(parts, 1);
        assert!(journal(&w).values().all(|t| *t != 0), "all folded");
        let r = f.open();
        assert_eq!(r.seals_for(&key).unwrap().len(), JOURNAL_FOLD_AT + 1);
        // The next complete fold takes the partial list in.
        assert!(fold(&w));
        let lists: Vec<String> = f.open().entry_lists.borrow().clone();
        assert_eq!(lists.len(), 1);
        assert!(lists[0].ends_with(LIST_SUFFIX));
        assert_eq!(f.open().seals_for(&key).unwrap().len(), JOURNAL_FOLD_AT + 1);
    }

    /// A collection that deletes an object takes its entries out of the
    /// journal too, so the next fold leaves them out.
    #[test]
    fn pruned_objects_leave_the_journal() {
        let f = Fixture::new();
        let w = f.open();
        let gone = w.put(ObjectType::Chunk, b"gone").unwrap();
        let kept = w.put(ObjectType::Chunk, b"kept").unwrap();
        sign(&w, &gone, 1);
        let s = sign(&w, &kept, 2);
        assert!(fold(&w));
        let keep: std::collections::HashSet<ObjectRef> = w
            .list()
            .unwrap()
            .into_iter()
            .filter(|r| *r != gone)
            .collect();
        w.prune(&keep, std::time::Duration::ZERO).unwrap();
        assert!(journal(&w).keys().all(|k| !k.contains(&gone.to_hex()[2..])));
        fold(&w);
        let r = f.open();
        assert!(r.signatures(&gone).unwrap().is_empty());
        assert_eq!(r.signatures(&kept).unwrap(), vec![s]);
        let all = r.with_entries(|e| e.clone()).unwrap().unwrap();
        assert!(
            all.iter().all(|k| !k.contains(&gone.to_hex()[2..])),
            "{all:?}"
        );
    }

    /// Requests per lookup, before and after the fold (#116): twenty
    /// objects, five of them signed, each looked up for its signatures
    /// and its seals, as a fetch does.
    #[test]
    fn a_folded_store_looks_up_entries_without_listing() {
        let f = Fixture::new();
        let w = f.open();
        let refs: Vec<ObjectRef> = (0..20)
            .map(|i| {
                w.put(ObjectType::Chunk, format!("o {i}").as_bytes())
                    .unwrap()
            })
            .collect();
        for r in &refs[..5] {
            sign(&w, r, 1);
        }
        let lookups = |s: &Store| {
            for r in &refs {
                s.signatures(r).unwrap();
                s.seals_for(r).unwrap();
            }
        };
        let before = f.open();
        f.backend.reset();
        lookups(&before);
        let (before_total, before_list) = (f.backend.total(), f.backend.n("list"));
        eprintln!("40 lookups before the fold: {}", f.backend.summary());

        assert!(fold(&w));
        let after = f.open();
        f.backend.reset();
        lookups(&after);
        eprintln!("40 lookups after the fold: {}", f.backend.summary());
        assert_eq!(before_list, 40);
        assert_eq!(f.backend.n("list"), 0);
        assert_eq!(f.backend.n("get_legacy"), 0, "the older lists are folded");
        assert_eq!(f.backend.n("read_journal"), 1);
        assert!(
            f.backend.total() < before_total / 10,
            "{}",
            f.backend.summary()
        );
    }

    /// Once a fold has copied the older lists into per-object entries and
    /// written its marker, readers stop reading them, folded or not; the
    /// answers are the same.
    #[test]
    fn the_legacy_marker_stops_reads_of_the_older_lists() {
        let f = Fixture::new();
        let w = f.open();
        let r = w.put(ObjectType::Chunk, b"old").unwrap();
        let sig = SecretKey {
            ed_seed: [4; 32],
            ml_seed: [5; 32],
        }
        .sign(Domain::Change, &ObjectRef([4; 32]), &r);
        f.backend
            .put_new(
                &legacy_key("sigs", &r),
                &Signature::encode_list(std::slice::from_ref(&sig)),
            )
            .unwrap();
        for (legacy, v) in [("seals", 9u8), ("certs", 8), ("accepted-lists", 6)] {
            f.backend
                .put_new(
                    &legacy_key(legacy, &r),
                    format!("{}\n", ObjectRef([v; 32]).to_hex()).as_bytes(),
                )
                .unwrap();
        }
        let check = |s: &Store| {
            assert_eq!(s.signatures(&r).unwrap(), vec![sig.clone()]);
            assert_eq!(s.seals_for(&r).unwrap(), vec![ObjectRef([9; 32])]);
            assert_eq!(
                s.index_get("cert", "certs", &r).unwrap(),
                vec![ObjectRef([8; 32])]
            );
            assert_eq!(
                s.index_get_pub("accepted", &r).unwrap(),
                vec![ObjectRef([6; 32])]
            );
        };
        let before = f.open();
        f.backend.reset();
        check(&before);
        assert_eq!(f.backend.n("get_legacy"), 4, "{}", f.backend.summary());

        assert!(fold(&w));
        let folded = f.open();
        f.backend.reset();
        check(&folded);
        assert_eq!(f.backend.n("get_legacy") + f.backend.n("list"), 0);

        // Not folded (its lists gone), but marked: per-object listings,
        // still no older lists.
        for e in f.backend.inner.list("pack-").unwrap() {
            if is_entry_list(&e.key) {
                f.backend.inner.delete(&e.key).unwrap();
            }
        }
        let marked = f.open();
        assert!(!marked.folded.get() && marked.legacy_folded.get());
        f.backend.reset();
        check(&marked);
        assert_eq!(f.backend.n("get_legacy"), 0, "{}", f.backend.summary());
        assert_eq!(f.backend.n("list"), 4);
    }
}