OXYGENHUB

Coverage / crates/oxy-core/src/store.rs 95.2%

2555 of 2685 lines run, 405 of 565 functions · run #354 on doctor-trust · run missed

1
//! Object store (spec §13.1, §13.3, §13.4): loose objects, one per key
2
//! under `loose/<hh>/<rest>`, and packs, `pack-<id>.pack` with a sorted
3
//! `pack-<id>.idx`, holding zstd-compressed objects. Reads check loose
4
//! objects first, then every pack; every read re-hashes.
5
//!
6
//! Everything is kept on a [`crate::strata::Store`] (docs/stores.md): a
7
//! directory, which is what `.oxy/objects` is, or a bucket, which is how a
8
//! repository lives in its owner's region with the hub as one client of
9
//! it. The keys are the same wherever the store is, so a bucket holds
10
//! exactly what a directory would, and a directory store is as fast as
11
//! the files were before there was a trait: a loose object is one file, a
12
//! packed one is one ranged read of the pack. What must be local, the
13
//! pack being streamed out and the derived `metas` cache, lives in a
14
//! scratch directory.
15
//!
16
//! Pack file: `OXYPACK1` then entries of
17
//! `type u8 | flags u8 (bit 0: zstd) | stored_len u32 | raw_len u32 | ref 32 | body`.
18
//! Index file: `OXYIDX01` then `count u32` then sorted entries of
19
//! `ref 32 | offset u64 | stored_len u32 | raw_len u32 | type u8 | flags u8`.
20
//!
21
//! Beside the objects: `nocompress/<hex>` marks a chunk packs keep raw;
22
//! `sig/<hh>/<rest>/<key>` holds one signature over an object, written
23
//! once per signer; `seal/<hh>/<rest>/<attest>` and
24
//! `receipt/<hh>/<rest>/<attest>` index the seals covering a change and
25
//! the receipts on a state. Older stores hold these as one list per
26
//! object (`sigs/`, `seals/`, `receipts/`), which are still read.
27
//! `pack-retired` records the packs and loose objects a repack replaced,
28
//! which stay for readers until a collection an hour later (oxygenhub#249).
29
30
use crate::attest::Attest;
31
use crate::hash::{hash_object, header, ObjectRef, ObjectType};
32
use crate::identity::{Identity, Signature};
33
use crate::objects::{Blob, Change, ObjectError, StateObject};
34
use crate::receipt::Receipt;
35
use crate::strata::{self, Error as BackendError, LocalStore, MemoryStore};
36
use rand::RngCore;
37
use std::cell::RefCell;
38
use std::fs;
39
use std::io::{self, Write};
40
use std::path::{Path, PathBuf};
41
use std::sync::Arc;
42
43
const PACK_MAGIC: &[u8; 8] = b"OXYPACK1";
44
const IDX_MAGIC: &[u8; 8] = b"OXYIDX01";
45
const IDX_ENTRY: usize = 32 + 8 + 4 + 4 + 1 + 1;
46
const FLAG_ZSTD: u8 = 1;
47
/// Compressed with the pack's trained dictionary (`pack-ID.dict`), which a
48
/// full repack makes from a sample of the objects it packs: small files
49
/// share most of their vocabulary, which each one alone cannot use.
50
const FLAG_DICT: u8 = 2;
51
/// The size of a pack's dictionary, and the least a repack must pack for
52
/// one to pay.
53
const DICT_BYTES: usize = 110 << 10;
54
const DICT_MIN_OBJECTS: usize = 1000;
55
/// Objects up to this size are compressed with the dictionary.
56
const DICT_OBJECT_MAX: usize = 64 << 10;
57
const ZSTD_LEVEL: i32 = 3;
58
/// The level of a full repack (`oxy gc`): slower, once, for a smaller store.
59
pub const REPACK_ZSTD_LEVEL: i32 = 9;
60
/// Largest object a pack entry may claim; anything bigger is corruption,
61
/// and refusing it keeps a bad index from driving huge allocations.
62
const MAX_OBJECT_LEN: u32 = 1 << 30;
63
/// What reading an entry of a pack whose dictionary is gone says. A pack
64
/// being deleted can lose its dictionary first, so a read that meets this
65
/// looks at the packs again before it is believed.
66
const NO_DICT: &str = "packed with a dictionary the pack lacks";
67
68
/// The record of what repacks replaced (oxygenhub#249): packs and loose
69
/// objects whose contents a newer pack holds, kept for readers that
70
/// still use them until garbage collection deletes them, at least
71
/// [`RETIRED_FOR`] later. One line per key, `<retired at, wall ms> <key>`.
72
/// It is a reference record (updated by compare-and-swap, so concurrent
73
/// repacks and collections never lose each other's lines), named under
74
/// `pack-` so the pack listing every open makes shows whether it holds
75
/// anything.
76
const RETIRED_KEY: &str = "pack-retired";
77
/// How long a retired pack or loose object stays in the store: a reader
78
/// that loaded its list of packs before a repack reads on from the old
79
/// packs this long, and after that finds the objects in the new pack by
80
/// looking again (see [`Store::get`]).
81
pub const RETIRED_FOR: std::time::Duration = std::time::Duration::from_secs(60 * 60);
82
/// Compare-and-swap rounds for the retired record before giving up.
83
const RETIRED_TRIES: usize = 32;
84
/// After a look at the pack listing found nothing new, how long a missed
85
/// [`Store::has`] answers without looking again: a writer probing for
86
/// thousands of objects it is about to store would otherwise list the
87
/// packs for each.
88
const HAS_RELIST_AFTER: std::time::Duration = std::time::Duration::from_secs(1);
89
90
/// A pack listing as compared: each key with its size and time.
91
type Listed = Vec<(String, u64, Option<u64>)>;
92
93
/// The retired record's lines: each key with the time it was retired.
94725×
fn parse_retired(bytes: &[u8]) -> std::collections::BTreeMap<String, u64> {
95725×
    String::from_utf8_lossy(bytes)
96725×
        .lines()
9735990×
        .filter_map(|l| {
9835990×
            let (t, k) = l.split_once(' ')?;
9935990×
            Some((k.to_string(), t.parse().ok()?))
10035990×
        })
101725×
        .collect()
102725×
}
103
104120×
fn encode_retired(m: &std::collections::BTreeMap<String, u64>) -> Vec<u8> {
105120×
    m.iter()
1065097×
        .map(|(k, t)| format!("{t} {k}\n"))
107120×
        .collect::<String>()
108120×
        .into_bytes()
109120×
}
110
111
/// What deleting retired packs and loose objects did.
112
#[derive(Debug, Clone, Default, PartialEq, Eq)]
113
pub struct RetiredReport {
114
    /// Retired packs deleted, each with its index and dictionary.
115
    pub packs: usize,
116
    /// Retired loose objects deleted.
117
    pub loose: usize,
118
    /// Retired entries younger than [`RETIRED_FOR`], left for the next
119
    /// collection.
120
    pub kept: usize,
121
}
122
123
/// An object read from the store: its type and body.
124
pub type Object = (ObjectType, Vec<u8>);
125
126
/// Where a long operation reports what it is doing (oxygen#66): a phase,
127
/// or a count within one, as a line for a spinner or a log.
128
pub type Progress<'a> = &'a mut dyn FnMut(&str);
129
130
/// The storage under a store: objects by key, see [`crate::strata`].
131
pub type Backend = dyn strata::Store;
132
13389×
fn silent() -> impl FnMut(&str) {
134173×
    |_| {}
13589×
}
136
137
/// A backend error as a store error: what is not found or already there
138
/// is handled where it matters; the rest is an I/O failure.
1394×
fn backend_err(e: BackendError) -> StoreError {
1404×
    match e {
1414×
        BackendError::Io(e) => StoreError::Io(e),
1420×
        other => StoreError::Io(io::Error::other(other.to_string())),
143
    }
1444×
}
145
146
/// `put_new` where an object already there is what we wanted.
1478704×
fn put_new_or_existing(backend: &Backend, key: &str, bytes: &[u8]) -> Result<()> {
1488704×
    match backend.put_new(key, bytes) {
1498704×
        Ok(()) | Err(BackendError::Exists(_)) => Ok(()),
1500×
        Err(e) => Err(backend_err(e)),
151
    }
1528704×
}
153
154
/// `get` as an option: `None` when there is no such key.
15560538×
fn get_optional(backend: &Backend, key: &str) -> Result<Option<Vec<u8>>> {
15660538×
    match backend.get(key) {
15752969×
        Ok(b) => Ok(Some(b)),
1587569×
        Err(BackendError::NotFound(_)) => Ok(None),
1590×
        Err(e) => Err(backend_err(e)),
160
    }
16160538×
}
162
1632670×
fn delete(backend: &Backend, key: &str) -> Result<()> {
1642670×
    backend.delete(key).map_err(backend_err)
1652670×
}
166
167
/// `loose/<hh>/<rest>`.
16897213×
fn loose_key(r: &ObjectRef) -> String {
16997213×
    let h = r.to_hex();
17097213×
    format!("loose/{}/{}", &h[..2], &h[2..])
17197213×
}
172
173
/// The object a loose key names, if it is one.
17426074×
fn loose_ref(key: &str) -> Option<ObjectRef> {
17526074×
    let rest = key.strip_prefix("loose/")?;
17626074×
    let (hh, tail) = rest.split_once('/')?;
17726074×
    if hh.len() != 2 || tail.contains('/') {
1780×
        return None;
17926074×
    }
18026074×
    ObjectRef::from_hex(&format!("{hh}{tail}"))
18126074×
}
182
183
/// `<index>/<hh>/<rest>/` for the entries kept under an object.
1846753×
fn fanout_prefix(index: &str, r: &ObjectRef) -> String {
1856753×
    let h = r.to_hex();
1866753×
    format!("{index}/{}/{}/", &h[..2], &h[2..])
1876753×
}
188
189
/// `<index>/<hh>/<rest>`: the one-file-per-object layout of older stores.
1905873×
fn legacy_key(index: &str, r: &ObjectRef) -> String {
1915873×
    let h = r.to_hex();
1925873×
    format!("{index}/{}/{}", &h[..2], &h[2..])
1935873×
}
194
195153×
fn now_ms() -> u64 {
196153×
    std::time::SystemTime::now()
197153×
        .duration_since(std::time::UNIX_EPOCH)
198153×
        .map(|d| d.as_millis() as u64)
199153×
        .unwrap_or(0)
200153×
}
201
202
#[derive(Clone, Copy, Debug)]
203
struct PackEntry {
204
    offset: u64,
205
    stored_len: u32,
206
    raw_len: u32,
207
    ty: ObjectType,
208
    flags: u8,
209
}
210
211
/// One pack's index, fully in memory and sorted by ObjectRef.
212
struct PackIndex {
213
    /// The pack's key, `pack-<id>.pack`.
214
    key: String,
215
    refs: Vec<ObjectRef>,
216
    entries: Vec<PackEntry>,
217
    /// The pack's dictionary, prepared once, when it has one.
218
    dict: Option<Arc<zstd::dict::DecoderDictionary<'static>>>,
219
}
220
221
impl std::fmt::Debug for PackIndex {
2220×
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
2230×
        f.debug_struct("PackIndex")
2240×
            .field("key", &self.key)
2250×
            .field("objects", &self.refs.len())
2260×
            .field("dict", &self.dict.is_some())
2270×
            .finish()
2280×
    }
229
}
230
231
/// Decode a stored pack entry and check it against its reference.
23217518×
fn decode_entry(
23317518×
    r: &ObjectRef,
23417518×
    e: PackEntry,
23517518×
    stored: Vec<u8>,
23617518×
    dict: Option<&Arc<zstd::dict::DecoderDictionary<'static>>>,
23717518×
) -> Result<Object> {
23817518×
    let body = if e.flags & FLAG_DICT != 0 {
2394502×
        let dict = dict.ok_or_else(|| StoreError::Corrupt(*r, NO_DICT.into()))?;
2404502×
        zstd::bulk::Decompressor::with_prepared_dictionary(dict)
2414502×
            .and_then(|mut d| d.decompress(&stored, e.raw_len as usize))
2424502×
            .map_err(|err| StoreError::Corrupt(*r, format!("zstd: {err}")))?
24313016×
    } else if e.flags & FLAG_ZSTD != 0 {
2447459×
        zstd::bulk::decompress(&stored, e.raw_len as usize)
2457459×
            .map_err(|err| StoreError::Corrupt(*r, format!("zstd: {err}")))?
246
    } else {
2475557×
        stored
248
    };
24917513×
    if body.len() != e.raw_len as usize || hash_object(e.ty, &body) != *r {
250119×
        return Err(StoreError::Corrupt(
251119×
            *r,
252119×
            "packed object does not match its reference".into(),
253119×
        ));
25417394×
    }
25517394×
    Ok((e.ty, body))
25617518×
}
257
258
impl PackIndex {
259
    /// Load a pack's index; `pack_len` is the pack's size, from the listing
260
    /// that found it (a stat of it would be a request of its own). `None`
261
    /// when the index is gone since the listing: the pack was deleted.
262484×
    fn load(
263484×
        backend: &Backend,
264484×
        idx_key: &str,
265484×
        pack_key: &str,
266484×
        pack_len: u64,
267484×
    ) -> Result<Option<PackIndex>> {
268484×
        let data = match backend.get(idx_key) {
269464×
            Ok(d) => d,
27020×
            Err(BackendError::NotFound(_)) => return Ok(None),
2710×
            Err(e) => return Err(backend_err(e)),
272
        };
273464×
        let bad = |m: &str| StoreError::Corrupt(ObjectRef::ZERO, format!("{idx_key}: {m}"));
274464×
        if data.len() < 12 || &data[..8] != IDX_MAGIC {
2752×
            return Err(bad("bad index header"));
276462×
        }
277462×
        let count = u32::from_le_bytes(data[8..12].try_into().unwrap()) as usize;
278462×
        if data.len() != 12 + count * IDX_ENTRY {
27942×
            return Err(bad("index length mismatch"));
280420×
        }
281420×
        let mut refs = Vec::with_capacity(count);
282420×
        let mut entries = Vec::with_capacity(count);
28324231×
        for i in 0..count {
28424231×
            let e = &data[12 + i * IDX_ENTRY..12 + (i + 1) * IDX_ENTRY];
28524231×
            let r = ObjectRef::from_slice(&e[..32]).unwrap();
28624231×
            if let Some(prev) = refs.last() {
28723811×
                if r <= *prev {
2883×
                    return Err(bad("index not sorted"));
28923808×
                }
290420×
            }
29124228×
            refs.push(r);
29224223×
            let entry = PackEntry {
29324228×
                offset: u64::from_le_bytes(e[32..40].try_into().unwrap()),
29424228×
                stored_len: u32::from_le_bytes(e[40..44].try_into().unwrap()),
29524228×
                raw_len: u32::from_le_bytes(e[44..48].try_into().unwrap()),
29624228×
                ty: ObjectType::ALL
29724228×
                    .get(e[48] as usize)
29824228×
                    .copied()
29924228×
                    .ok_or_else(|| bad("unknown object type"))?,
30024223×
                flags: e[49],
301
            };
30224223×
            if entry.stored_len > MAX_OBJECT_LEN || entry.raw_len > MAX_OBJECT_LEN {
3033×
                return Err(StoreError::Corrupt(
3043×
                    r,
3053×
                    format!(
3063×
                        "{idx_key}: the index claims an object of {} bytes ({} stored), more than the {MAX_OBJECT_LEN} an object may be",
3073×
                        entry.raw_len, entry.stored_len
3083×
                    ),
3093×
                ));
31024220×
            }
31124220×
            if entry.offset.saturating_add(entry.stored_len as u64) > pack_len {
31254×
                return Err(bad("index entry points outside the pack"));
31324166×
            }
31424166×
            entries.push(entry);
315
        }
316355×
        let dict_key = format!("{}.dict", pack_key.trim_end_matches(".pack"));
317355×
        let dict = get_optional(backend, &dict_key)?
318355×
            .map(|d| Arc::new(zstd::dict::DecoderDictionary::copy(&d)));
319355×
        Ok(Some(PackIndex {
320355×
            key: pack_key.to_string(),
321355×
            refs,
322355×
            entries,
323355×
            dict,
324355×
        }))
325484×
    }
326
32723085×
    fn find(&self, r: &ObjectRef) -> Option<PackEntry> {
32823085×
        self.refs.binary_search(r).ok().map(|i| self.entries[i])
32923085×
    }
330
331
    /// Read one entry from the pack: one ranged read, then decode.
33217198×
    fn fetch(&self, backend: &Backend, r: &ObjectRef, e: PackEntry) -> Result<Object> {
33317198×
        let stored = backend
33417198×
            .get_range(&self.key, e.offset..e.offset + e.stored_len as u64)
33517198×
            .map_err(|err| match err {
336162×
                BackendError::NotFound(_) => StoreError::NotFound(*r),
3370×
                other => backend_err(other),
338162×
            })?;
33917036×
        decode_entry(r, e, stored, self.dict.as_ref())
34017198×
    }
341
}
342
34315754×
fn type_index(ty: ObjectType) -> u8 {
34422180×
    ObjectType::ALL.iter().position(|t| *t == ty).unwrap() as u8
34515754×
}
346
347
/// What `pack_loose` did.
348
#[derive(Debug, Clone, Default, PartialEq, Eq)]
349
pub struct PackReport {
350
    pub objects: usize,
351
    pub raw_bytes: u64,
352
    pub packed_bytes: u64,
353
}
354
355
/// Packs tolerated before `merge_packs` folds them into one (spec §13.4).
356
pub const MAX_PACKS: usize = 16;
357
358
/// What `prune` did.
359
#[derive(Debug, Clone, Default, PartialEq, Eq)]
360
pub struct PruneReport {
361
    pub objects: usize,
362
    pub bytes: u64,
363
    /// Unreachable objects left alone because they are younger than the grace period.
364
    pub deferred: usize,
365
}
366
367
#[derive(Debug, thiserror::Error)]
368
pub enum StoreError {
369
    #[error("io: {0}")]
370
    Io(#[from] io::Error),
371
    #[error("corrupt object {0}: {1}")]
372
    Corrupt(ObjectRef, String),
373
    #[error("object {0}")]
374
    Object(#[from] ObjectError),
375
    #[error("object {0} is a {1}, expected {2}")]
376
    WrongType(ObjectRef, &'static str, &'static str),
377
    #[error("object {0} not found")]
378
    NotFound(ObjectRef),
379
    #[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)")]
380
    TooLarge {
381
        kind: &'static str,
382
        size: usize,
383
        limit: u32,
384
    },
385
}
386
387
pub type Result<T> = std::result::Result<T, StoreError>;
388
389
pub struct Store {
390
    backend: Arc<Backend>,
391
    /// Temporary files (a pack being streamed) and derived caches (`metas`).
392
    /// For a store in a directory it is that directory.
393
    scratch: PathBuf,
394
    /// The backend is that directory, so the old file paths still hold.
395
    local: bool,
396
    packs: RefCell<Vec<Arc<PackIndex>>>,
397
    /// The pack listing `packs` was loaded from (key, size, time), to tell
398
    /// whether a look at the store would find anything new.
399
    packs_listed: RefCell<Listed>,
400
    /// When a look at the pack listing for a missed [`Store::has`] last
401
    /// found nothing new.
402
    has_relisted: std::cell::Cell<Option<std::time::Instant>>,
403
    /// An open write batch (spec §13.4): objects go here instead of to loose
404
    /// files, and are written out with one `fsync` when the batch finishes.
405
    pending: RefCell<Option<Pending>>,
406
    batch_depth: std::cell::Cell<u32>,
407
    /// Change metadata already read, from the `metas` cache file.
408
    metas: RefCell<Option<crate::fasthash::FastMap<ObjectRef, crate::state::ChangeMeta>>>,
409
}
410
411
impl std::fmt::Debug for Store {
4120×
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
4130×
        f.debug_struct("Store")
4140×
            .field("backend", &self.backend.describe())
4150×
            .field("scratch", &self.scratch)
4160×
            .field("packs", &self.packs.borrow().len())
4170×
            .finish()
4180×
    }
419
}
420
421
/// Objects written since `begin_batch`. Small batches stay in memory and
422
/// become loose objects; once `SPILL_OBJECTS` or `SPILL_BYTES` is exceeded
423
/// the batch streams into a pack file instead.
424
#[derive(Debug)]
425
struct Pending {
426
    buffered: Vec<(ObjectType, Vec<u8>)>,
427
    buffered_bytes: usize,
428
    lookup: std::collections::HashMap<ObjectRef, PendingEntry>,
429
    pack: Option<PendingPack>,
430
    report: PackReport,
431
}
432
433
#[derive(Debug, Clone, Copy)]
434
enum PendingEntry {
435
    Buffered(usize),
436
    Packed(PackEntry),
437
}
438
439
#[derive(Debug)]
440
struct PendingPack {
441
    id: String,
442
    tmp_path: PathBuf,
443
    writer: io::BufWriter<fs::File>,
444
    offset: u64,
445
    index: Vec<(ObjectRef, PackEntry)>,
446
}
447
448
const SPILL_OBJECTS: usize = 256;
449
const SPILL_BYTES: usize = 8 << 20;
450
451
/// Finishes the enclosing batch when dropped; call `finish` to see errors.
452
pub struct Batch<'a> {
453
    store: &'a Store,
454
    done: bool,
455
}
456
457
impl Batch<'_> {
4582959×
    pub fn finish(mut self) -> Result<()> {
4592959×
        self.done = true;
4602959×
        self.store.finish_batch()
4612959×
    }
462
}
463
464
impl Drop for Batch<'_> {
4652961×
    fn drop(&mut self) {
4662961×
        if !self.done {
4672×
            let _ = self.store.finish_batch();
4682959×
        }
4692961×
    }
470
}
471
472
/// Append one stored object to a pending pack and index it.
4733226×
fn write_pack_entry(
4743226×
    pack: &mut PendingPack,
4753226×
    r: &ObjectRef,
4763226×
    ty: ObjectType,
4773226×
    raw_len: usize,
4783226×
    flags: u8,
4793226×
    stored: &[u8],
4803226×
    report: &mut PackReport,
4813226×
) -> Result<PackEntry> {
4823226×
    let header_len = 1 + 1 + 4 + 4 + 32;
4833226×
    pack.writer.write_all(&[type_index(ty), flags])?;
4843226×
    pack.writer
4853226×
        .write_all(&(stored.len() as u32).to_le_bytes())?;
4863226×
    pack.writer.write_all(&(raw_len as u32).to_le_bytes())?;
4873226×
    pack.writer.write_all(&r.0)?;
4883226×
    pack.writer.write_all(stored)?;
4893226×
    let entry = PackEntry {
4903226×
        offset: pack.offset + header_len,
4913226×
        stored_len: stored.len() as u32,
4923226×
        raw_len: raw_len as u32,
4933226×
        ty,
4943226×
        flags,
4953226×
    };
4963226×
    pack.offset += header_len + stored.len() as u64;
4973226×
    pack.index.push((*r, entry));
4983226×
    report.objects += 1;
4993226×
    report.raw_bytes += raw_len as u64;
5003226×
    report.packed_bytes += stored.len() as u64;
5013226×
    Ok(entry)
5023226×
}
503
504
/// A random name for a temporary file or a pack.
505108×
fn fresh_id() -> String {
506108×
    let mut nonce = [0u8; 8];
507108×
    rand::rngs::OsRng.fill_bytes(&mut nonce);
508108×
    hex::encode(nonce)
509108×
}
510
511
/// Under `OXY_STORE_BACKEND=memory` every store opens on a store in
512
/// memory shared by directory, so the whole test suite can run without a
513
/// directory store, as it would over a bucket. Objects live for the
514
/// process.
5152896×
fn memory_backend_for(dir: &Path) -> Option<Arc<Backend>> {
516
    use std::collections::HashMap;
517
    use std::sync::{Mutex, OnceLock};
5182896×
    if std::env::var_os("OXY_STORE_BACKEND").as_deref() != Some(std::ffi::OsStr::new("memory")) {
5192896×
        return None;
5200×
    }
521
    static STORES: OnceLock<Mutex<HashMap<PathBuf, Arc<MemoryStore>>>> = OnceLock::new();
5220×
    let key = dir.canonicalize().unwrap_or_else(|_| dir.to_path_buf());
5230×
    let mut map = STORES
5240×
        .get_or_init(Default::default)
5250×
        .lock()
5260×
        .unwrap_or_else(|p| p.into_inner());
5270×
    let store = map
5280×
        .entry(key)
5290×
        .or_insert_with(|| Arc::new(MemoryStore::new()))
5300×
        .clone();
5310×
    Some(store)
5322896×
}
533
534
#[cfg(test)]
535
thread_local! {
536
    /// A backend every store opened on this thread uses instead of its
537
    /// directory: tests of a repository kept on a store that is not this
538
    /// disk's directory, with several handles on one store.
539
    pub(crate) static TEST_BACKEND: RefCell<Option<Arc<Backend>>> = const { RefCell::new(None) };
540
}
541
542
impl Store {
543
    /// Open the store in `dir`, which is also its scratch directory.
5442905×
    pub fn open(dir: impl Into<PathBuf>) -> io::Result<Store> {
5452905×
        let dir = dir.into();
5462905×
        fs::create_dir_all(dir.join("loose"))?;
547
        #[cfg(test)]
5481332×
        if let Some(backend) = TEST_BACKEND.with(|t| t.borrow().clone()) {
5499×
            return Store::open_on(backend, dir);
5501323×
        }
5512896×
        if let Some(backend) = memory_backend_for(&dir) {
5520×
            return Store::open_on(backend, dir);
5532896×
        }
5542896×
        let backend: Arc<Backend> =
5552896×
            Arc::new(LocalStore::open(&dir).map_err(|e| io::Error::other(e.to_string()))?);
5562896×
        let mut store = Store::open_on(backend, dir)?;
5572787×
        store.local = true;
5582787×
        Ok(store)
5592905×
    }
560
561
    /// Open the store kept on `backend`, with `scratch` for what must be
562
    /// on this disk: a pack while it is being written, and caches.
5632945×
    pub fn open_on(backend: Arc<Backend>, scratch: impl Into<PathBuf>) -> io::Result<Store> {
5642945×
        let scratch = scratch.into();
5652945×
        fs::create_dir_all(&scratch)?;
5662945×
        let store = Store {
5672945×
            backend,
5682945×
            scratch,
5692945×
            local: false,
5702945×
            packs: RefCell::new(Vec::new()),
5712945×
            packs_listed: RefCell::new(Vec::new()),
5722945×
            has_relisted: std::cell::Cell::new(None),
5732945×
            pending: RefCell::new(None),
5742945×
            batch_depth: std::cell::Cell::new(0),
5752945×
            metas: RefCell::new(None),
5762945×
        };
5772945×
        store
5782945×
            .reload_packs()
5792945×
            .map_err(|e| io::Error::other(e.to_string()))?;
5802836×
        Ok(store)
5812945×
    }
582
583
    /// The storage under this store.
58458×
    pub fn backend(&self) -> &Arc<Backend> {
58558×
        &self.backend
58658×
    }
587
588
    /// Whether the store is the directory `dir()` names, where every key
589
    /// is a file.
59014024×
    pub fn is_local(&self) -> bool {
59114024×
        self.local
59214024×
    }
593
594
    /// Load the list of packs again, from a new listing: what other
595
    /// writers packed since becomes readable, and packs since retired or
596
    /// deleted are let go. Packs already loaded are kept, not read again.
597467×
    pub fn refresh(&self) -> Result<()> {
598467×
        self.reload_packs()
599467×
    }
600
601
    /// The pack listing, with how `packs_listed` records it.
6025922×
    fn list_packs(&self) -> Result<(Vec<strata::Entry>, Listed)> {
6035922×
        let listed = self.backend.list("pack-").map_err(backend_err)?;
6045922×
        let sig = listed
6055922×
            .iter()
6065922×
            .map(|e| (e.key.clone(), e.size, e.modified_ms))
6075922×
            .collect();
6085922×
        Ok((listed, sig))
6095922×
    }
610
611
    /// Load every pack the listing names that is not retired. An index
612
    /// that is gone by the time it is read (its pack deleted after the
613
    /// listing) is passed over, and the store listed once more, since
614
    /// what replaced it may be new since the first listing.
6153647×
    fn reload_packs(&self) -> Result<()> {
6163647×
        let (listed, sig) = self.list_packs()?;
6173647×
        self.load_listed(listed, sig)
6183647×
    }
619
6203654×
    fn load_listed(&self, mut listed: Vec<strata::Entry>, mut sig: Listed) -> Result<()> {
6213654×
        let had: std::collections::HashMap<String, Arc<PackIndex>> = self
6223654×
            .packs
6233654×
            .borrow()
6243654×
            .iter()
6253654×
            .map(|p| (p.key.clone(), p.clone()))
6263654×
            .collect();
6273654×
        let mut loaded: std::collections::HashMap<String, Arc<PackIndex>> = Default::default();
6283664×
        for attempt in 0..2 {
6293664×
            let retired = self.retired_in(&listed)?;
6303664×
            let mut idx_keys: Vec<&str> = listed
6313664×
                .iter()
6323664×
                .map(|e| e.key.as_str())
6333664×
                .filter(|k| k.ends_with(".idx"))
6343664×
                .collect();
6353664×
            idx_keys.sort();
6363664×
            let mut packs = Vec::new();
6373664×
            let mut vanished = false;
6383664×
            for idx in idx_keys {
6391475×
                let pack = format!("{}.pack", idx.trim_end_matches(".idx"));
6401475×
                if retired.contains_key(&pack) {
641173×
                    continue;
6421302×
                }
64311414×
                let Some(p) = listed.iter().find(|e| e.key == pack) else {
6440×
                    continue;
645
                };
6461302×
                if let Some(kept) = had.get(&pack).or_else(|| loaded.get(&pack)) {
647818×
                    packs.push(kept.clone());
648818×
                    continue;
649484×
                }
650484×
                match PackIndex::load(self.backend.as_ref(), idx, &pack, p.size)? {
651355×
                    Some(index) => {
652355×
                        let index = Arc::new(index);
653355×
                        loaded.insert(pack, index.clone());
654355×
                        packs.push(index);
655355×
                    }
65620×
                    None => vanished = true,
657
                }
658
            }
6593555×
            if vanished && attempt == 0 {
66010×
                (listed, sig) = self.list_packs()?;
66110×
                continue;
6623545×
            }
6633545×
            *self.packs.borrow_mut() = packs;
6643545×
            *self.packs_listed.borrow_mut() = sig;
6653545×
            return Ok(());
666
        }
6670×
        unreachable!("the second round always returns")
6683654×
    }
669
670
    /// Load the packs again when the pack listing differs from the one
671
    /// they were loaded from; whether it did. What a miss calls before it
672
    /// is believed (oxygenhub#249): another writer may have packed the
673
    /// object, or a repack moved it, since this store loaded its packs.
674561×
    fn reload_if_changed(&self) -> Result<bool> {
675561×
        let (listed, sig) = self.list_packs()?;
676561×
        if *self.packs_listed.borrow() == sig {
677554×
            return Ok(false);
6787×
        }
6797×
        self.load_listed(listed, sig)?;
6807×
        Ok(true)
681561×
    }
682
683
    /// The retired record, read when `listed` (a pack listing) shows it
684
    /// holds anything.
6855368×
    fn retired_in(
6865368×
        &self,
6875368×
        listed: &[strata::Entry],
6885368×
    ) -> Result<std::collections::BTreeMap<String, u64>> {
6895368×
        match listed.iter().find(|e| e.key == RETIRED_KEY) {
690661×
            Some(e) if e.size > 0 => Ok(self.read_retired()?.0),
6914742×
            _ => Ok(Default::default()),
692
        }
6935368×
    }
694
695
    /// The retired record and its version (`None`: never written).
696758×
    fn read_retired(
697758×
        &self,
698758×
    ) -> Result<(
699758×
        std::collections::BTreeMap<String, u64>,
700758×
        Option<strata::Version>,
701758×
    )> {
702758×
        match self.backend.read_ref(RETIRED_KEY).map_err(backend_err)? {
703725×
            Some((bytes, v)) => Ok((parse_retired(&bytes), Some(v))),
70433×
            None => Ok((Default::default(), None)),
705
        }
706758×
    }
707
708
    /// Change the retired record by compare-and-swap, reading it again
709
    /// whenever another writer changed it first. `edit` must give the
710
    /// same result however often it runs.
711120×
    fn update_retired(
712120×
        &self,
713120×
        mut edit: impl FnMut(&mut std::collections::BTreeMap<String, u64>),
714120×
    ) -> Result<()> {
715120×
        for _ in 0..RETIRED_TRIES {
716120×
            let (mut m, version) = self.read_retired()?;
717120×
            let before = m.clone();
718120×
            edit(&mut m);
719120×
            if m == before && version.is_some() {
7200×
                return Ok(());
721120×
            }
722120×
            match self
723120×
                .backend
724120×
                .write_ref(RETIRED_KEY, &encode_retired(&m), version.as_ref())
725
            {
726120×
                Ok(_) => return Ok(()),
7270×
                Err(BackendError::Conflict(_)) => continue,
7280×
                Err(e) => return Err(backend_err(e)),
729
            }
730
        }
7310×
        Err(StoreError::Io(io::Error::other(
7320×
            "other writers kept changing the record of retired packs; run the command again",
7330×
        )))
734120×
    }
735
736
    /// Record `keys` (packs, by their `.pack` key, and loose objects) as
737
    /// retired now: kept for the readers still using them, and deleted by
738
    /// a collection at least [`RETIRED_FOR`] later. A key already retired
739
    /// keeps its first time.
74099×
    fn retire(&self, keys: &[String]) -> Result<()> {
74199×
        if keys.is_empty() {
7420×
            return Ok(());
74399×
        }
74499×
        let now = now_ms();
74599×
        self.update_retired(|m| {
7462508×
            for k in keys {
7472508×
                m.entry(k.clone()).or_insert(now);
7482508×
            }
74999×
        })
75099×
    }
751
752
    /// Delete what was retired at least `older_than` ago (a collection
753
    /// does, with [`RETIRED_FOR`]): each pack with its index and
754
    /// dictionary, and each loose object a pack now holds. A retired loose
755
    /// object no pack holds any more (the pack that had it was pruned, and
756
    /// the object stored again since) is kept, as an ordinary loose object.
757757×
    pub fn expire_retired(
758757×
        &self,
759757×
        older_than: std::time::Duration,
760757×
        progress: Progress,
761757×
    ) -> Result<RetiredReport> {
762757×
        let (listed, _) = self.list_packs()?;
763757×
        let retired = self.retired_in(&listed)?;
764757×
        let mut report = RetiredReport::default();
765757×
        if retired.is_empty() {
766733×
            return Ok(report);
76724×
        }
76824×
        let now = now_ms();
76924×
        let older_than = older_than.as_millis() as u64;
77024×
        let due: Vec<(String, u64)> = retired
77124×
            .iter()
7721969×
            .filter(|(_, t)| now.saturating_sub(**t) >= older_than)
7731722×
            .map(|(k, t)| (k.clone(), *t))
77424×
            .collect();
77524×
        report.kept = retired.len() - due.len();
77624×
        if due.is_empty() {
77712×
            return Ok(report);
77812×
        }
77912×
        progress(&format!("deleting {} retired files", due.len()));
780
        // Which objects the live packs hold.
78112×
        self.reload_packs()?;
7821722×
        for (key, _) in &due {
7831722×
            if key.ends_with(".pack") {
78412×
                self.remove_pack(key)?;
78512×
                report.packs += 1;
7861710×
            } else if let Some(r) = loose_ref(key) {
7871710×
                if self.find_packed(&r).is_some() {
7881708×
                    delete(self.backend.as_ref(), key)?;
7891708×
                    report.loose += 1;
7902×
                }
7910×
            }
792
        }
79312×
        self.update_retired(|m| {
7941722×
            for (k, t) in &due {
7951722×
                if m.get(k) == Some(t) {
7961722×
                    m.remove(k);
7971722×
                }
798
            }
79912×
        })?;
80012×
        self.reload_packs()?;
80112×
        Ok(report)
802757×
    }
803
804
    /// The loose objects retired and not yet deleted, by key.
805947×
    fn retired_loose(&self) -> Result<std::collections::HashSet<String>> {
806947×
        let (listed, _) = self.list_packs()?;
807947×
        Ok(self
808947×
            .retired_in(&listed)?
809947×
            .into_keys()
8105365×
            .filter(|k| k.starts_with("loose/"))
811947×
            .collect())
812947×
    }
813
81435×
    pub fn pack_count(&self) -> usize {
81535×
        self.packs.borrow().len()
81635×
    }
817
818102584×
    fn find_packed(&self, r: &ObjectRef) -> Option<(usize, PackEntry)> {
819102584×
        self.packs
820102584×
            .borrow()
821102584×
            .iter()
822102584×
            .enumerate()
823102584×
            .find_map(|(i, p)| p.find(r).map(|e| (i, e)))
824102584×
    }
825
826
    /// Loose objects, with their keys and ages; not those retired, whose
827
    /// objects a pack holds.
828917×
    fn loose_objects(&self) -> Result<Vec<(ObjectRef, strata::Entry)>> {
829917×
        let retired = self.retired_loose()?;
830917×
        let mut out = self.loose_objects_all()?;
83123839×
        out.retain(|(_, e)| !retired.contains(&e.key));
832917×
        Ok(out)
833917×
    }
834
835
    /// Every loose object, retired or not.
836947×
    fn loose_objects_all(&self) -> Result<Vec<(ObjectRef, strata::Entry)>> {
837947×
        let mut out: Vec<(ObjectRef, strata::Entry)> = self
838947×
            .backend
839947×
            .list("loose/")
840947×
            .map_err(backend_err)?
841947×
            .into_iter()
84224364×
            .filter_map(|e| loose_ref(&e.key).map(|r| (r, e)))
843947×
            .collect();
844947×
        out.sort_by_key(|(r, _)| *r);
845947×
        Ok(out)
846947×
    }
847
848741×
    pub fn loose_count(&self) -> io::Result<usize> {
849741×
        self.loose_objects()
850741×
            .map(|l| l.len())
851741×
            .map_err(|e| io::Error::other(e.to_string()))
852741×
    }
853
854
    /// Move every loose object into a new pack (spec §13.4). Loose objects
855
    /// are deleted only after the pack and its index are durable.
85654×
    pub fn pack_loose(&self) -> Result<PackReport> {
85754×
        let loose = self.loose_objects()?;
85854×
        self.write_pack(
85954×
            &loose.iter().map(|(r, _)| *r).collect::<Vec<_>>(),
860541×
            &loose.iter().map(|(_, e)| e.key.clone()).collect::<Vec<_>>(),
86154×
            &[],
862
            ZSTD_LEVEL,
86354×
            &mut silent(),
864
        )
86554×
    }
866
867
    /// Fold every pack and loose object into one pack compressed at
868
    /// `level` (spec §13.4): what an explicit `oxy gc` does, as `git gc`
869
    /// repacks. Returns the report and how many packs were folded in.
87013×
    pub fn repack_all(&self, level: i32) -> Result<(PackReport, usize)> {
87113×
        self.repack_all_with(level, &mut silent())
87213×
    }
873
874
    /// [`Store::repack_all`], reporting progress.
87536×
    pub fn repack_all_with(&self, level: i32, progress: Progress) -> Result<(PackReport, usize)> {
87636×
        let loose = self.loose_objects()?;
87736×
        let (mut refs, keys, trained): (Vec<ObjectRef>, Vec<String>, bool) = {
87836×
            let packs = self.packs.borrow();
879
            (
88036×
                packs.iter().flat_map(|p| p.refs.iter().copied()).collect(),
88136×
                packs.iter().map(|p| p.key.clone()).collect(),
88236×
                packs.first().is_some_and(|p| p.dict.is_some()),
883
            )
884
        };
885
        // Nothing loose and one pack already made by a full repack (only
886
        // such a pack has a trained dictionary): rewriting it would give
887
        // the same pack. A hub's nightly gc rewrote andymccall/oxygen's
888
        // 48 000 objects (37 s under its write lock) every time.
88936×
        if loose.is_empty() && keys.len() <= 1 && (level == ZSTD_LEVEL || trained) {
8901×
            return Ok((PackReport::default(), 0));
89135×
        }
89235×
        refs.extend(loose.iter().map(|(r, _)| *r));
89335×
        let loose_keys: Vec<String> = loose.into_iter().map(|(_, e)| e.key).collect();
89435×
        let n = keys.len();
89535×
        let report = self.write_pack(&refs, &loose_keys, &keys, level, progress)?;
89635×
        Ok((report, n))
89736×
    }
898
899
    /// Fold every pack into one when there are more than `MAX_PACKS`
900
    /// (spec §13.4). Returns how many packs were merged.
9014×
    pub fn merge_packs(&self) -> Result<usize> {
9022×
        let (refs, keys): (Vec<ObjectRef>, Vec<String>) = {
9034×
            let packs = self.packs.borrow();
9044×
            if packs.len() <= MAX_PACKS {
9052×
                return Ok(0);
9062×
            }
907
            (
90836×
                packs.iter().flat_map(|p| p.refs.iter().copied()).collect(),
90936×
                packs.iter().map(|p| p.key.clone()).collect(),
910
            )
911
        };
9122×
        let n = keys.len();
9132×
        self.write_pack(&refs, &[], &keys, ZSTD_LEVEL, &mut silent())?;
9142×
        Ok(n)
9154×
    }
916
917
    /// Delete every object not in `keep` (spec §21). Loose objects and packs
918
    /// younger than `grace` are left alone, since a concurrent writer may not
919
    /// have recorded its references yet. Packs holding unreachable objects
920
    /// are rewritten without them. Signatures, seal and receipt entries of
921
    /// deleted objects go too.
9227×
    pub fn prune(
9237×
        &self,
9247×
        keep: &std::collections::HashSet<ObjectRef>,
9257×
        grace: std::time::Duration,
9267×
    ) -> Result<PruneReport> {
9277×
        self.prune_with(keep, grace, &mut silent())
9287×
    }
929
930
    /// [`Store::prune`], reporting progress.
93130×
    pub fn prune_with(
93230×
        &self,
93330×
        keep: &std::collections::HashSet<ObjectRef>,
93430×
        grace: std::time::Duration,
93530×
        progress: Progress,
93630×
    ) -> Result<PruneReport> {
93730×
        progress("pruning loose objects");
93830×
        let now = now_ms();
93930×
        let grace_ms = grace.as_millis() as u64;
940
        // An object whose age the store does not know is old enough.
94130×
        let old_enough =
94284×
            |modified: Option<u64>| modified.is_none_or(|t| now.saturating_sub(t) >= grace_ms);
94330×
        let mut report = PruneReport::default();
944
        // Loose objects. A retired one (its object packed) whose object is
945
        // unreachable goes with it, uncounted: the pack's copy is counted.
94630×
        let retired = self.retired_loose()?;
94730×
        let mut unretired: Vec<String> = Vec::new();
948525×
        for (r, entry) in self.loose_objects_all()? {
949525×
            if keep.contains(&r) {
950450×
                continue;
95175×
            }
95275×
            if retired.contains(&entry.key) {
95345×
                if old_enough(entry.modified_ms) {
95441×
                    delete(self.backend.as_ref(), &entry.key)?;
95541×
                    unretired.push(entry.key);
9564×
                }
95745×
                continue;
95830×
            }
95930×
            if !old_enough(entry.modified_ms) {
9604×
                report.deferred += 1;
9614×
                continue;
96226×
            }
96326×
            report.bytes += entry.size;
96426×
            report.objects += 1;
96526×
            delete(self.backend.as_ref(), &entry.key)?;
96626×
            self.remove_sidecars(&r)?;
967
        }
96830×
        if !unretired.is_empty() {
9696×
            self.update_retired(|m| {
97041×
                for k in &unretired {
97141×
                    m.remove(k);
97241×
                }
9736×
            })?;
97424×
        }
975
        // Packs: rewrite those that hold unreachable objects.
97630×
        let mut keep_refs: Vec<ObjectRef> = Vec::new();
97730×
        let mut old_keys: Vec<String> = Vec::new();
97830×
        let mut rewrite = false;
97930×
        progress("pruning packs");
98030×
        for p in self.packs.borrow().iter() {
981213×
            let unreachable = p.refs.iter().filter(|r| !keep.contains(r)).count();
98215×
            if unreachable == 0 {
9836×
                continue;
9849×
            }
9859×
            let modified = self
9869×
                .backend
9879×
                .stat(&p.key)
9889×
                .map_err(backend_err)?
9899×
                .and_then(|e| e.modified_ms);
9909×
            if !old_enough(modified) {
9913×
                report.deferred += unreachable;
9923×
                continue;
9936×
            }
9946×
            rewrite = true;
9956×
            old_keys.push(p.key.clone());
996102×
            for (r, e) in p.refs.iter().zip(p.entries.iter()) {
997102×
                if keep.contains(r) {
99861×
                    keep_refs.push(*r);
99961×
                } else {
100041×
                    report.objects += 1;
100141×
                    report.bytes += e.stored_len as u64;
100241×
                }
1003
            }
1004
        }
100530×
        if rewrite {
10066×
            let dropped: Vec<ObjectRef> = {
10076×
                let packs = self.packs.borrow();
10086×
                packs
10096×
                    .iter()
10106×
                    .filter(|p| old_keys.contains(&p.key))
10116×
                    .flat_map(|p| p.refs.iter().copied())
1012102×
                    .filter(|r| !keep.contains(r))
10136×
                    .collect()
1014
            };
10156×
            self.write_pack(&keep_refs, &[], &old_keys, ZSTD_LEVEL, progress)?;
101641×
            for r in dropped {
101741×
                self.remove_sidecars(&r)?;
1018
            }
101924×
        }
102030×
        Ok(report)
102130×
    }
1022
1023
    /// Everything kept beside an object: its signatures and the seal and
1024
    /// receipt entries under it, in both layouts.
102567×
    fn remove_sidecars(&self, r: &ObjectRef) -> Result<()> {
1026201×
        for index in ["sigs", "seals", "receipts"] {
1027201×
            delete(self.backend.as_ref(), &legacy_key(index, r))?;
1028
        }
1029201×
        for index in ["sig", "seal", "receipt"] {
1030201×
            for e in self
1031201×
                .backend
1032201×
                .list(&fanout_prefix(index, r))
1033201×
                .map_err(backend_err)?
1034
            {
103512×
                delete(self.backend.as_ref(), &e.key)?;
1036
            }
1037
        }
103867×
        Ok(())
103967×
    }
1040
1041
    /// Put a complete, synced temporary pack file and its index into the
1042
    /// store; `index` must be sorted by ObjectRef. The pack goes first: an
1043
    /// index never names a pack that is not there.
1044108×
    fn finish_pack_files(
1045108×
        &self,
1046108×
        id: &str,
1047108×
        tmp_pack: &Path,
1048108×
        index: &[(ObjectRef, PackEntry)],
1049108×
    ) -> Result<()> {
1050108×
        let mut idx = Vec::with_capacity(12 + index.len() * IDX_ENTRY);
1051108×
        idx.extend_from_slice(IDX_MAGIC);
1052108×
        idx.extend_from_slice(&(index.len() as u32).to_le_bytes());
10537877×
        for (r, e) in index {
10547877×
            idx.extend_from_slice(&r.0);
10557877×
            idx.extend_from_slice(&e.offset.to_le_bytes());
10567877×
            idx.extend_from_slice(&e.stored_len.to_le_bytes());
10577877×
            idx.extend_from_slice(&e.raw_len.to_le_bytes());
10587877×
            idx.extend_from_slice(&[type_index(e.ty), e.flags]);
10597877×
        }
1060108×
        let put = self
1061108×
            .backend
1062108×
            .put_new_file(&format!("pack-{id}.pack"), tmp_pack)
1063108×
            .map_err(backend_err);
1064
        // A store on another disk copied it; the one in this directory
1065
        // moved it and this finds nothing.
1066108×
        let _ = fs::remove_file(tmp_pack);
1067108×
        put?;
1068108×
        self.backend
1069108×
            .put_new(&format!("pack-{id}.idx"), &idx)
1070108×
            .map_err(backend_err)?;
1071108×
        self.reload_packs()
1072108×
    }
1073
1074
    /// Remove a pack with its index and dictionary: the index first, so
1075
    /// a store opened meanwhile does not load a pack that is going.
107612×
    fn remove_pack(&self, pack_key: &str) -> Result<()> {
107712×
        let stem = pack_key.trim_end_matches(".pack");
107812×
        delete(self.backend.as_ref(), &format!("{stem}.idx"))?;
107912×
        delete(self.backend.as_ref(), pack_key)?;
108012×
        delete(self.backend.as_ref(), &format!("{stem}.dict"))
108112×
    }
1082
1083
    /// Write `refs` into a new pack, then retire `loose_keys` and
1084
    /// `old_packs`: they stay in the store, unused by stores opened from
1085
    /// now on, until a collection [`RETIRED_FOR`] later deletes them, so a
1086
    /// reader that loaded its packs before goes on reading (oxygenhub#249).
108799×
    fn write_pack(
108899×
        &self,
108999×
        refs: &[ObjectRef],
109099×
        loose_keys: &[String],
109199×
        old_packs: &[String],
109299×
        level: i32,
109399×
        progress: Progress,
109499×
    ) -> Result<PackReport> {
109599×
        if refs.is_empty() {
10960×
            self.retire(old_packs)?;
10970×
            self.reload_packs()?;
10980×
            return Ok(PackReport::default());
109999×
        }
110099×
        let mut refs: Vec<ObjectRef> = refs.to_vec();
110199×
        refs.sort();
110299×
        refs.dedup();
110399×
        let id = fresh_id();
110499×
        let tmp_pack = self.scratch.join(format!(".tmp-{id}.pack"));
110599×
        let mut report = PackReport::default();
110699×
        let mut index: Vec<(ObjectRef, PackEntry)> = Vec::with_capacity(refs.len());
1107
        // A full repack of enough objects trains a dictionary for them.
110899×
        let dict = if level == REPACK_ZSTD_LEVEL && refs.len() >= DICT_MIN_OBJECTS {
11092×
            let _t = crate::timing::phase("pack: train a dictionary");
11102×
            progress(&format!("training a dictionary on {} objects", refs.len()));
11112×
            self.train_dictionary(&refs)
1112
        } else {
111397×
            None
1114
        };
111599×
        let encoder = dict
111699×
            .as_ref()
111799×
            .map(|d| zstd::dict::EncoderDictionary::copy(d, level));
111899×
        let nocompress = self.nocompress_set()?;
1119
        {
112099×
            let mut f = io::BufWriter::new(fs::File::create(&tmp_pack)?);
112199×
            f.write_all(PACK_MAGIC)?;
112299×
            let mut offset = PACK_MAGIC.len() as u64;
1123
            // Read a batch (the store is not shared between threads),
1124
            // compress it on every core, write it in order.
112599×
            let mut batches = refs.chunks(1024);
112699×
            let mut compressed_batch = Vec::new();
112799×
            let total = refs.len();
112899×
            let mut reported = usize::MAX;
1129
            loop {
11304750×
                if compressed_batch.is_empty() {
1131200×
                    let Some(batch) = batches.next() else { break };
1132
                    // Every tenth, and the first batch.
1133101×
                    let tenth = report.objects * 10 / total.max(1);
1134101×
                    if tenth != reported {
1135101×
                        reported = tenth;
1136101×
                        progress(&format!("packing {} of {total} objects", report.objects));
1137101×
                    }
1138101×
                    let mut read: Vec<(ObjectRef, ObjectType, Vec<u8>, bool)> =
1139101×
                        Vec::with_capacity(batch.len());
11404651×
                    for r in batch {
11414651×
                        let (ty, body) = self.get(r)?;
11424651×
                        read.push((*r, ty, body, nocompress.contains(r)));
1143
                    }
1144
                    use rayon::prelude::*;
1145101×
                    compressed_batch = read
1146101×
                        .into_par_iter()
1147101×
                        .map_init(
11481251×
                            || {
11491251×
                                encoder.as_ref().and_then(|e| {
1150192×
                                    zstd::bulk::Compressor::with_prepared_dictionary(e).ok()
1151192×
                                })
11521251×
                            },
11534651×
                            |with_dict, (r, ty, body, raw)| {
11544651×
                                let packed = if raw {
11552×
                                    None
1156
                                } else {
1157
                                    // Small objects with the dictionary;
1158
                                    // a large one has context enough.
11594649×
                                    let dicted = with_dict
11604649×
                                        .as_mut()
11614649×
                                        .filter(|_| body.len() <= DICT_OBJECT_MAX)
11624649×
                                        .and_then(|c| c.compress(&body).ok())
11634649×
                                        .map(|c| (FLAG_ZSTD | FLAG_DICT, c));
11644649×
                                    dicted
11654649×
                                        .or_else(|| {
11661648×
                                            zstd::bulk::compress(&body, level)
11671648×
                                                .ok()
11681648×
                                                .map(|c| (FLAG_ZSTD, c))
11691648×
                                        })
11704649×
                                        .filter(|(_, c)| c.len() < body.len())
1171
                                };
11724651×
                                (r, ty, body, packed)
11734651×
                            },
1174
                        )
1175101×
                        .collect::<Vec<_>>();
1176101×
                    compressed_batch.reverse();
11774550×
                }
11784651×
                let Some((r, ty, body, packed)) = compressed_batch.pop() else {
11790×
                    continue;
1180
                };
11814651×
                let r = &r;
11824651×
                let (flags, stored) = match packed {
11833801×
                    Some(c) => c,
1184850×
                    None => (0, body.clone()),
1185
                };
11864651×
                let header_len = 1 + 1 + 4 + 4 + 32;
11874651×
                f.write_all(&[type_index(ty), flags])?;
11884651×
                f.write_all(&(stored.len() as u32).to_le_bytes())?;
11894651×
                f.write_all(&(body.len() as u32).to_le_bytes())?;
11904651×
                f.write_all(&r.0)?;
11914651×
                f.write_all(&stored)?;
11924651×
                index.push((
11934651×
                    *r,
11944651×
                    PackEntry {
11954651×
                        offset: offset + header_len,
11964651×
                        stored_len: stored.len() as u32,
11974651×
                        raw_len: body.len() as u32,
11984651×
                        ty,
11994651×
                        flags,
12004651×
                    },
12014651×
                ));
12024651×
                offset += header_len + stored.len() as u64;
12034651×
                report.objects += 1;
12044651×
                report.raw_bytes += body.len() as u64;
12054651×
                report.packed_bytes += stored.len() as u64;
1206
            }
120799×
            f.flush()?;
120899×
            f.get_ref().sync_all()?;
1209
        }
1210
        // The dictionary before the pack that needs it becomes visible.
121199×
        if let Some(d) = &dict {
12122×
            self.backend
12132×
                .put_new(&format!("pack-{id}.dict"), d)
12142×
                .map_err(backend_err)?;
121597×
        }
121699×
        progress(&format!("writing the index of {} objects", index.len()));
121799×
        self.finish_pack_files(&id, &tmp_pack, &index)?;
121899×
        let mut retired = old_packs.to_vec();
121999×
        retired.extend_from_slice(loose_keys);
122099×
        self.retire(&retired)?;
122199×
        self.reload_packs()?;
122299×
        Ok(report)
122399×
    }
1224
1225
    /// A dictionary for these objects, trained on a sample of them (at
1226
    /// most 10 000, 32 MB); `None` when training fails.
12272×
    fn train_dictionary(&self, refs: &[ObjectRef]) -> Option<Vec<u8>> {
12282×
        let step = (refs.len() / 10_000).max(1);
12292×
        let mut samples: Vec<Vec<u8>> = Vec::new();
12302×
        let mut total = 0usize;
12313001×
        for r in refs.iter().step_by(step) {
12323001×
            if total > 32 << 20 {
12330×
                break;
12343001×
            }
12353001×
            let Ok((ty, body)) = self.get(r) else {
12360×
                continue;
1237
            };
12380×
            if matches!(
12393001×
                ty,
1240
                ObjectType::Chunk | ObjectType::Blob | ObjectType::Change
12413001×
            ) && !body.is_empty()
12423001×
                && body.len() < 1 << 20
12433001×
            {
12443001×
                total += body.len();
12453001×
                samples.push(body);
12463001×
            }
1247
        }
12482×
        zstd::dict::from_samples(&samples, DICT_BYTES).ok()
12492×
    }
1250
1251
    /// The scratch directory: for a store in a directory, that directory.
125219×
    pub fn dir(&self) -> &Path {
125319×
        &self.scratch
125419×
    }
1255
1256
    /// Where a loose object is kept, for a store in a directory (tests
1257
    /// tamper with and remove objects there).
1258
    #[cfg(test)]
12594×
    pub(crate) fn path_for(&self, r: &ObjectRef) -> PathBuf {
12604×
        self.scratch.join(loose_key(r))
12614×
    }
1262
1263
    /// Remove a loose object, whatever the store is on (tests make an
1264
    /// object go missing).
1265
    #[cfg(test)]
12663×
    pub(crate) fn remove_loose_for_test(&self, r: &ObjectRef) -> Result<()> {
12673×
        delete(self.backend.as_ref(), &loose_key(r))
12683×
    }
1269
1270
    /// Whether the store holds `r`. A miss looks at the pack listing
1271
    /// again before it is believed, as [`Store::get`]'s does; a writer
1272
    /// asking about many objects it is about to store looks at most once
1273
    /// in [`HAS_RELIST_AFTER`] while nothing new turns up.
127412864×
    pub fn has(&self, r: &ObjectRef) -> bool {
127512864×
        if self.has_now(r) {
127612174×
            return true;
1277690×
        }
1278690×
        let recent = self
1279690×
            .has_relisted
1280690×
            .get()
1281690×
            .is_some_and(|t| t.elapsed() < HAS_RELIST_AFTER);
1282690×
        if recent {
1283180×
            return false;
1284510×
        }
1285510×
        match self.reload_if_changed() {
1286
            Ok(true) => {
12874×
                self.has_relisted.set(None);
12884×
                self.has_now(r)
1289
            }
1290
            _ => {
1291506×
                self.has_relisted.set(Some(std::time::Instant::now()));
1292506×
                false
1293
            }
1294
        }
129512864×
    }
1296
1297
    /// [`Store::has`] against the packs as loaded.
129812868×
    fn has_now(&self, r: &ObjectRef) -> bool {
129912868×
        self.find_packed(r).is_some()
130012321×
            || self
130112321×
                .pending
130212321×
                .borrow()
130312321×
                .as_ref()
130412321×
                .is_some_and(|p| p.lookup.contains_key(r))
130511984×
            || self.backend.has(&loose_key(r)).unwrap_or(false)
130612868×
    }
1307
1308
    /// Start a write batch: until the matching `finish`, `put` collects
1309
    /// objects and writes them out together, with one `fsync` for a large
1310
    /// batch (a pack) instead of one per object. Batches nest; only the
1311
    /// outermost does the work. Objects are readable while pending.
13122962×
    pub fn begin_batch(&self) -> Result<Batch<'_>> {
13132962×
        let depth = self.batch_depth.get();
13142962×
        if depth == 0 {
13152395×
            *self.pending.borrow_mut() = Some(Pending {
13162395×
                buffered: vec![],
13172395×
                buffered_bytes: 0,
13182395×
                lookup: Default::default(),
13192395×
                pack: None,
13202395×
                report: PackReport::default(),
13212395×
            });
13222395×
        }
13232962×
        self.batch_depth.set(depth + 1);
13242962×
        Ok(Batch {
13252962×
            store: self,
13262962×
            done: false,
13272962×
        })
13282962×
    }
1329
13302961×
    fn finish_batch(&self) -> Result<()> {
13312961×
        let depth = self.batch_depth.get();
13322961×
        if depth == 0 {
13330×
            return Ok(());
13342961×
        }
13352961×
        self.batch_depth.set(depth - 1);
13362961×
        if depth > 1 {
1337567×
            return Ok(());
13382394×
        }
13392394×
        let Some(pending) = self.pending.borrow_mut().take() else {
13402×
            return Ok(());
1341
        };
13422392×
        self.write_pending(pending)
13432961×
    }
1344
1345
    /// Make every object written so far durable, even inside an open
1346
    /// batch, which stays open. Call before a reference that names them
1347
    /// is moved (spec §13.6): the operation head, a bookmark pointer.
13483736×
    pub fn flush(&self) -> Result<()> {
13493736×
        let Some(pending) = self.pending.borrow_mut().take() else {
13502608×
            return Ok(());
1351
        };
13521128×
        self.write_pending(pending)?;
13531126×
        *self.pending.borrow_mut() = Some(Pending {
13541126×
            buffered: vec![],
13551126×
            buffered_bytes: 0,
13561126×
            lookup: Default::default(),
13571126×
            pack: None,
13581126×
            report: PackReport::default(),
13591126×
        });
13601126×
        Ok(())
13613736×
    }
1362
13633520×
    fn write_pending(&self, pending: Pending) -> Result<()> {
13643520×
        match pending.pack {
1365
            None => {
13663512×
                let objects: Vec<(ObjectType, &[u8])> = pending
13673512×
                    .buffered
13683512×
                    .iter()
13695521×
                    .map(|(ty, body)| (*ty, body.as_slice()))
13703512×
                    .collect();
13713512×
                self.put_loose_many(&objects)
1372
            }
13738×
            Some(mut pack) => {
13748×
                pack.writer.flush()?;
13758×
                pack.writer.get_ref().sync_all()?;
13768×
                drop(pack.writer);
13778×
                pack.index.sort_by_key(|(r, _)| *r);
13788×
                self.finish_pack_files(&pack.id, &pack.tmp_path, &pack.index)?;
13798×
                Ok(())
1380
            }
1381
        }
13823520×
    }
1383
1384
    /// Whether a batch is open and holds `r`.
1385908×
    fn pending_get(&self, r: &ObjectRef) -> Option<Result<Object>> {
1386908×
        let mut guard = self.pending.borrow_mut();
1387908×
        let pending = guard.as_mut()?;
1388857×
        let entry = *pending.lookup.get(r)?;
1389857×
        Some(match entry {
1390675×
            PendingEntry::Buffered(i) => {
1391675×
                let (ty, body) = &pending.buffered[i];
1392675×
                Ok((*ty, body.clone()))
1393
            }
1394182×
            PendingEntry::Packed(e) => {
1395182×
                let Some(pack) = pending.pack.as_mut() else {
13960×
                    return Some(Err(StoreError::NotFound(*r)));
1397
                };
1398
                // Read from the pack file being written, on this disk.
1399182×
                (|| {
1400182×
                    pack.writer.flush()?;
1401
                    use std::io::{Read, Seek, SeekFrom};
1402182×
                    let mut f = fs::File::open(&pack.tmp_path)?;
1403182×
                    f.seek(SeekFrom::Start(e.offset))?;
1404182×
                    let mut stored = vec![0u8; e.stored_len as usize];
1405182×
                    f.read_exact(&mut stored)?;
1406182×
                    decode_entry(r, e, stored, None)
1407
                })()
1408
            }
1409
        })
1410908×
    }
1411
1412
    /// Append one object to the pending pack file.
14133026×
    fn pending_pack_write(
14143026×
        &self,
14153026×
        pack: &mut PendingPack,
14163026×
        r: &ObjectRef,
14173026×
        ty: ObjectType,
14183026×
        body: &[u8],
14193026×
        report: &mut PackReport,
14203026×
    ) -> Result<PackEntry> {
14213026×
        let (flags, stored) = if self.is_nocompress(r) {
1422120×
            (0, body.to_vec())
1423
        } else {
14242906×
            let compressed = zstd::bulk::compress(body, ZSTD_LEVEL)
14252906×
                .map_err(|e| StoreError::Corrupt(*r, format!("zstd: {e}")))?;
14262906×
            if compressed.len() < body.len() {
14272292×
                (FLAG_ZSTD, compressed)
1428
            } else {
1429614×
                (0, body.to_vec())
1430
            }
1431
        };
14323026×
        write_pack_entry(pack, r, ty, body.len(), flags, &stored, report)
14333026×
    }
1434
14358666×
    fn put_pending(&self, r: ObjectRef, ty: ObjectType, body: &[u8]) -> Result<()> {
14368666×
        let mut guard = self.pending.borrow_mut();
14378666×
        let pending = guard.as_mut().expect("batch open");
14388666×
        if pending.lookup.contains_key(&r) {
1439119×
            return Ok(());
14408547×
        }
14418547×
        if pending.pack.is_none()
14427483×
            && pending.buffered.len() < SPILL_OBJECTS
14437477×
            && pending.buffered_bytes + body.len() <= SPILL_BYTES
1444
        {
14457475×
            pending.buffered.push((ty, body.to_vec()));
14467475×
            pending.buffered_bytes += body.len();
14477475×
            pending
14487475×
                .lookup
14497475×
                .insert(r, PendingEntry::Buffered(pending.buffered.len() - 1));
14507475×
            return Ok(());
14511072×
        }
14521072×
        self.ensure_pack(pending)?;
14531072×
        let mut report = std::mem::take(&mut pending.report);
14541072×
        let pack = pending.pack.as_mut().unwrap();
14551072×
        let e = self.pending_pack_write(pack, &r, ty, body, &mut report)?;
14561072×
        pending.report = report;
14571072×
        pending.lookup.insert(r, PendingEntry::Packed(e));
14581072×
        Ok(())
14598666×
    }
1460
1461
    /// Open the pending batch's pack, moving the objects buffered so far
1462
    /// into it (a spill).
14631073×
    fn ensure_pack(&self, pending: &mut Pending) -> Result<()> {
14641073×
        if pending.pack.is_none() {
1465
            // Spill: open the pack and move the buffered objects into it.
14669×
            let id = fresh_id();
14679×
            let tmp_path = self.scratch.join(format!(".tmp-{id}.pack"));
14689×
            let mut writer = io::BufWriter::new(fs::File::create(&tmp_path)?);
14699×
            writer.write_all(PACK_MAGIC)?;
14709×
            let mut pack = PendingPack {
14719×
                id,
14729×
                tmp_path,
14739×
                writer,
14749×
                offset: PACK_MAGIC.len() as u64,
14759×
                index: vec![],
14769×
            };
14779×
            let buffered = std::mem::take(&mut pending.buffered);
14789×
            pending.buffered_bytes = 0;
14799×
            let mut report = std::mem::take(&mut pending.report);
14801954×
            for (bty, bbody) in &buffered {
14811954×
                let br = hash_object(*bty, bbody);
14821954×
                let e = self.pending_pack_write(&mut pack, &br, *bty, bbody, &mut report)?;
14831954×
                pending.lookup.insert(br, PendingEntry::Packed(e));
1484
            }
14859×
            pending.report = report;
14869×
            pending.pack = Some(pack);
14871064×
        }
14881073×
        Ok(())
14891073×
    }
1490
1491
    /// Store many objects at once. Inside a large write batch they are
1492
    /// compressed on every core and appended in order; otherwise each is
1493
    /// stored as [`Store::put`] does.
1494836×
    pub fn put_many(&self, objects: &[(ObjectType, &[u8])]) -> Result<Vec<ObjectRef>> {
1495
        use rayon::prelude::*;
1496903×
        let total: usize = objects.iter().map(|(_, b)| b.len()).sum();
1497836×
        if self.pending.borrow().is_none() || total < SPILL_BYTES {
1498835×
            return objects.iter().map(|(ty, b)| self.put(*ty, b)).collect();
14991×
        }
1500202×
        for (ty, b) in objects {
1501202×
            if b.len() > MAX_OBJECT_LEN as usize {
15020×
                return Err(StoreError::TooLarge {
15030×
                    kind: ty.name(),
15040×
                    size: b.len(),
15050×
                    limit: MAX_OBJECT_LEN,
15060×
                });
1507202×
            }
1508
        }
15091×
        let refs: Vec<ObjectRef> = objects
15101×
            .par_iter()
1511202×
            .map(|(ty, b)| hash_object(*ty, b))
15121×
            .collect();
1513
        // What is new: not stored already, not earlier in this call.
15141×
        let mut seen: std::collections::HashSet<ObjectRef> = std::collections::HashSet::new();
15151×
        let mut todo: Vec<usize> = Vec::new();
1516202×
        for (i, r) in refs.iter().enumerate() {
1517202×
            let known = self
1518202×
                .pending
1519202×
                .borrow()
1520202×
                .as_ref()
1521202×
                .is_some_and(|p| p.lookup.contains_key(r));
1522202×
            if seen.insert(*r)
1523201×
                && !known
1524201×
                && self.find_packed(r).is_none()
1525201×
                && !self.backend.has(&loose_key(r)).unwrap_or(false)
1526200×
            {
1527200×
                todo.push(i);
1528200×
            }
1529
        }
15301×
        let raw = self.nocompress_set()?;
15311×
        let packed: Vec<(u8, Vec<u8>)> = todo
15321×
            .par_iter()
1533200×
            .map(|i| {
1534200×
                let body = objects[*i].1;
1535200×
                if raw.contains(&refs[*i]) {
15360×
                    return (0, body.to_vec());
1537200×
                }
1538200×
                match zstd::bulk::compress(body, ZSTD_LEVEL) {
1539200×
                    Ok(c) if c.len() < body.len() => (FLAG_ZSTD, c),
15400×
                    _ => (0, body.to_vec()),
1541
                }
1542200×
            })
15431×
            .collect();
15441×
        let mut guard = self.pending.borrow_mut();
15451×
        let pending = guard.as_mut().expect("batch open");
15461×
        self.ensure_pack(pending)?;
15471×
        let mut report = std::mem::take(&mut pending.report);
15481×
        let pack = pending.pack.as_mut().unwrap();
1549200×
        for (i, (flags, stored)) in todo.iter().zip(packed) {
1550200×
            let (ty, body) = objects[*i];
1551200×
            let e = write_pack_entry(pack, &refs[*i], ty, body.len(), flags, &stored, &mut report)?;
1552200×
            pending.lookup.insert(refs[*i], PendingEntry::Packed(e));
1553
        }
15541×
        pending.report = report;
15551×
        Ok(refs)
1556836×
    }
1557
1558
    /// Store `body` as an object of type `ty`; idempotent.
155918174×
    pub fn put(&self, ty: ObjectType, body: &[u8]) -> Result<ObjectRef> {
1560
        // Refuse what the store could not read back (spec §13.2), before
1561
        // anything is written: a pack must never hold an entry its index
1562
        // would reject.
156318174×
        if body.len() > MAX_OBJECT_LEN as usize {
15641×
            return Err(StoreError::TooLarge {
15651×
                kind: ty.name(),
15661×
                size: body.len(),
15671×
                limit: MAX_OBJECT_LEN,
15681×
            });
156918173×
        }
157018173×
        let r = hash_object(ty, body);
157118173×
        if self.find_packed(&r).is_some() || self.backend.has(&loose_key(&r)).unwrap_or(false) {
15721988×
            return Ok(r);
157316185×
        }
157416185×
        if self.pending.borrow().is_some() {
15758666×
            self.put_pending(r, ty, body)?;
15768666×
            return Ok(r);
15777519×
        }
15787519×
        self.put_loose(ty, body)?;
15797514×
        Ok(r)
158018174×
    }
1581
1582
    /// Write one loose object, durable when this returns.
15837751×
    fn put_loose(&self, ty: ObjectType, body: &[u8]) -> Result<ObjectRef> {
15847751×
        let r = hash_object(ty, body);
15857751×
        let mut bytes = header(ty, body.len());
15867751×
        bytes.extend_from_slice(body);
15877751×
        crate::durable::step("write loose object")?;
15887746×
        put_new_or_existing(self.backend.as_ref(), &loose_key(&r), &bytes)?;
15897746×
        Ok(r)
15907751×
    }
1591
1592
    /// Write several loose objects with as few waits on the disk as the
1593
    /// guarantees allow (a directory store writes them all, then syncs
1594
    /// each directory once). Durable when this returns, as `put_loose`
1595
    /// one at a time would be (spec §13.6).
15963512×
    fn put_loose_many(&self, objects: &[(ObjectType, &[u8])]) -> Result<()> {
15973512×
        if objects.len() <= 1 {
15982015×
            for (ty, body) in objects {
1599232×
                self.put_loose(*ty, body)?;
1600
            }
16012015×
            return Ok(());
16021497×
        }
16031497×
        let mut seen = std::collections::HashSet::new();
16041497×
        let items: Vec<(String, Vec<u8>)> = objects
16051497×
            .iter()
16065289×
            .filter_map(|(ty, body)| {
16075289×
                let r = hash_object(*ty, body);
16085289×
                seen.insert(r).then(|| {
16095289×
                    let mut bytes = header(*ty, body.len());
16105289×
                    bytes.extend_from_slice(body);
16115289×
                    (loose_key(&r), bytes)
16125289×
                })
16135289×
            })
16141497×
            .collect();
16151497×
        self.backend.put_many_new(&items).map_err(backend_err)
16163512×
    }
1617
1618
    /// Read and verify an object, packed, loose, or pending in a batch.
1619
    /// The packs' indexes are in memory, so they are asked first: asking
1620
    /// for a loose copy of a packed object costs a request that finds
1621
    /// nothing on a store in a bucket, every time (oxygenhub#228).
1622
    ///
1623
    /// A miss is not believed at once (oxygenhub#249): when the pack it
1624
    /// was in is gone (deleted after a repack retired it) the packs are
1625
    /// loaded again, and when the object is nowhere the pack listing is
1626
    /// looked at, and the packs loaded again if it changed (another
1627
    /// writer packed it, or a repack moved it); then it is read once more.
162868068×
    pub fn get(&self, r: &ObjectRef) -> Result<Object> {
162968068×
        match self.get_now(r) {
1630173×
            Err(e) if is_miss(&e) => {
1631
                // Read from a pack this store loaded, which is gone.
163246×
                let packed_miss =
163346×
                    !matches!(e, StoreError::NotFound(_)) || self.find_packed(r).is_some();
163446×
                let changed = if packed_miss {
16352×
                    self.reload_packs()?;
16362×
                    true
1637
                } else {
163844×
                    self.reload_if_changed()?
1639
                };
164046×
                if changed {
16415×
                    self.get_now(r)
1642
                } else {
164341×
                    Err(e)
1644
                }
1645
            }
164668022×
            other => other,
1647
        }
164868068×
    }
1649
1650
    /// [`Store::get`] against the packs as loaded.
165168073×
    fn get_now(&self, r: &ObjectRef) -> Result<Object> {
165268073×
        if let Some((i, e)) = self.find_packed(r) {
165316132×
            let pack = self.packs.borrow()[i].clone();
165416132×
            return pack.fetch(self.backend.as_ref(), r, e);
165551941×
        }
165651941×
        match get_optional(self.backend.as_ref(), &loose_key(r))? {
165751040×
            Some(bytes) => parse_loose(&bytes, r),
1658901×
            None => self.pending_get(r).unwrap_or(Err(StoreError::NotFound(*r))),
1659
        }
166068073×
    }
1661
1662
    /// Many objects at once, read and verified in parallel: packed ones
1663
    /// from their packs, the others loose (as [`Store::get`], packs first),
1664
    /// then what is pending in a batch. Results are in the order of `refs`.
1665
    ///
1666
    /// Misses are looked for again as [`Store::get`]'s are, together.
16671026×
    pub fn get_many(&self, refs: &[ObjectRef]) -> Vec<Result<Object>> {
16681026×
        let mut out = self.get_many_now(refs);
16691026×
        let missed: Vec<usize> = out
16701026×
            .iter()
16711026×
            .enumerate()
16722787×
            .filter(|(_, g)| g.as_ref().is_err_and(is_miss))
16731026×
            .map(|(i, _)| i)
16741026×
            .collect();
16751026×
        if missed.is_empty() {
16761017×
            return out;
16779×
        }
16789×
        let packed_miss = missed.iter().any(|i| {
16799×
            !matches!(out[*i], Err(StoreError::NotFound(_)))
16809×
                || self.find_packed(&refs[*i]).is_some()
16819×
        });
16829×
        let changed = if packed_miss {
16832×
            self.reload_packs().map(|_| true)
1684
        } else {
16857×
            self.reload_if_changed()
1686
        };
16879×
        match changed {
1688
            Ok(true) => {
1689160×
                let again: Vec<ObjectRef> = missed.iter().map(|i| refs[*i]).collect();
1690160×
                for (i, g) in missed.into_iter().zip(self.get_many_now(&again)) {
1691160×
                    out[i] = g;
1692160×
                }
1693
            }
16947×
            Ok(false) => {}
16950×
            Err(e) => {
16960×
                let why = e.to_string();
16970×
                for i in missed {
16980×
                    out[i] = Err(StoreError::Io(io::Error::other(why.clone())));
16990×
                }
1700
            }
1701
        }
17029×
        out
17031026×
    }
1704
1705
    /// [`Store::get_many`] against the packs as loaded.
17061028×
    fn get_many_now(&self, refs: &[ObjectRef]) -> Vec<Result<Object>> {
1707
        use rayon::prelude::*;
17081028×
        let backend = self.backend.as_ref();
17091028×
        let packs = self.packs.borrow();
17101028×
        let place: Vec<Option<(usize, PackEntry)>> = refs
17111028×
            .iter()
17122947×
            .map(|r| {
17132947×
                packs
17142947×
                    .iter()
17152947×
                    .enumerate()
17162947×
                    .find_map(|(p, pack)| pack.find(r).map(|e| (p, e)))
17172947×
            })
17181028×
            .collect();
17191028×
        let indexes: &[Arc<PackIndex>] = &packs;
17201028×
        let found: Vec<Option<Result<Object>>> = refs
17211028×
            .par_iter()
17221028×
            .zip(place.par_iter())
17232947×
            .map(|(r, at)| match at {
17241065×
                Some((p, e)) => Some(indexes[*p].fetch(backend, r, *e)),
17251882×
                None => match get_optional(backend, &loose_key(r)) {
17261875×
                    Ok(Some(bytes)) => Some(parse_loose(&bytes, r)),
17277×
                    Ok(None) => None,
17280×
                    Err(e) => Some(Err(e)),
1729
                },
17302947×
            })
17311028×
            .collect();
17321028×
        drop(packs);
17331028×
        found
17341028×
            .into_iter()
17351028×
            .zip(refs)
17362947×
            .map(|(f, r)| {
17372947×
                f.unwrap_or_else(|| self.pending_get(r).unwrap_or(Err(StoreError::NotFound(*r))))
17382947×
            })
17391028×
            .collect()
17401028×
    }
1741
174247591×
    pub fn get_typed(&self, r: &ObjectRef, want: ObjectType) -> Result<Vec<u8>> {
174347591×
        let (ty, body) = self.get(r)?;
174447552×
        if ty != want {
17451×
            return Err(StoreError::WrongType(*r, ty.name(), want.name()));
174647551×
        }
174747551×
        Ok(body)
174847591×
    }
1749
1750
    /// Store a change and the text blobs (and their chunks) its large
1751
    /// inserts are stored as (spec §6.1).
17523161×
    pub fn put_change(&self, c: &Change) -> Result<ObjectRef> {
17533161×
        let encoding = crate::timing::phase("store: encode the change and its texts");
17543161×
        let bytes = c.encode_streaming(|batch| self.put_texts(&batch))?;
17553161×
        drop(encoding);
17563161×
        let _t = crate::timing::phase("store: put the encoding");
17573161×
        self.put(ObjectType::Change, &bytes)
17583161×
    }
1759
1760
    /// Store text blobs and their chunks: every chunk at once (compressed
1761
    /// on every core in a large batch), then the blobs.
176262×
    pub fn put_texts(&self, texts: &[crate::objects::StoredText]) -> Result<()> {
176362×
        let mut chunks: Vec<(ObjectType, &[u8])> = Vec::new();
176463×
        for t in texts {
176563×
            if !self.has(&t.blob.hash()) {
176652×
                chunks.extend(
176752×
                    crate::chunk::chunks(&t.content)
176852×
                        .into_iter()
176993×
                        .map(|c| (ObjectType::Chunk, c)),
1770
                );
177111×
            }
1772
        }
177362×
        self.put_many(&chunks)?;
177463×
        for t in texts {
177563×
            self.put(ObjectType::Blob, &t.blob.encode())?;
1776
        }
177762×
        Ok(())
177862×
    }
1779
17805201×
    pub fn get_change(&self, r: &ObjectRef) -> Result<Change> {
17815201×
        Ok(self.get_change_parts(r)?.0)
17825201×
    }
1783
1784
    /// A change and the objects its stored inserts are made of: each text
1785
    /// blob and its chunks, which travel and are kept with it.
17865363×
    pub fn get_change_parts(&self, r: &ObjectRef) -> Result<(Change, Vec<ObjectRef>)> {
17875363×
        let bytes = self.get_typed(r, ObjectType::Change)?;
17885351×
        let mut parts = Vec::new();
17895351×
        let mut failed = None;
17905351×
        let decoded = Change::decode_with(&bytes, &mut |text, count, form| match self
179159×
            .read_text(text)
1792
        {
179358×
            Ok((blob, content)) => {
179458×
                parts.push(*text);
179558×
                parts.extend(blob.chunks);
179658×
                crate::objects::text_lines(&content, count, form)
1797
            }
17981×
            Err(e) => {
17991×
                let why = e.to_string();
18001×
                failed = Some(e);
18011×
                Err(ObjectError::Schema(format!(
18021×
                    "text blob {}: {why}",
18031×
                    text.short(12)
18041×
                )))
1805
            }
180659×
        });
18075351×
        match (decoded, failed) {
18085350×
            (Ok(c), _) => Ok((c, parts)),
18091×
            (Err(_), Some(StoreError::NotFound(t))) => Err(StoreError::Corrupt(
18101×
                *r,
18111×
                format!("its text blob {} is missing", t.short(12)),
18121×
            )),
18130×
            (Err(e), _) => Err(StoreError::Corrupt(*r, e.to_string())),
1814
        }
18155363×
    }
1816
1817
    /// A change with every insert left without its lines (empty
1818
    /// placeholders), and where each one's lines are, in order: for
1819
    /// history-only states, which read text only for the graphs they build
1820
    /// (see [`Change::decode_skeleton`]).
1821555×
    pub fn get_change_lazy(&self, r: &ObjectRef) -> Result<(Change, Vec<crate::objects::TextRef>)> {
1822555×
        let bytes = self.get_typed(r, ObjectType::Change)?;
1823553×
        Change::decode_skeleton(&bytes, *r).map_err(|e| StoreError::Corrupt(*r, e.to_string()))
1824555×
    }
1825
1826
    /// A change's metadata (id, dependencies, author, clock, message, kind)
1827
    /// without its contents: read from its skeleton, so neither its text
1828
    /// blobs nor its inline lines are read (spec §11).
1829
    ///
1830
    /// Served from the `metas` cache when this store read it before (a
1831
    /// change object can be as large as the tree it adds), as long as the
1832
    /// store still has the change.
18334705×
    pub fn get_change_meta(&self, r: &ObjectRef) -> Result<crate::state::ChangeMeta> {
18344705×
        if self.metas.borrow().is_none() {
1835977×
            *self.metas.borrow_mut() = Some(self.read_metas_cache());
18363728×
        }
18374705×
        let cached = self.metas.borrow().as_ref().and_then(|m| m.get(r).cloned());
18384705×
        if let Some(meta) = cached {
18393372×
            if self.has(r) {
1840
                #[cfg(test)]
18411834×
                assert_eq!(
1842
                    meta,
18431834×
                    Change::decode_meta(&self.get_typed(r, ObjectType::Change)?).unwrap(),
1844
                    "the metadata cache holds a change's metadata"
1845
                );
18463372×
                return Ok(meta);
18470×
            }
18481333×
        }
18491333×
        let bytes = self.get_typed(r, ObjectType::Change)?;
18501333×
        let meta =
18511333×
            Change::decode_meta(&bytes).map_err(|e| StoreError::Corrupt(*r, e.to_string()))?;
18521333×
        self.remember_meta(r, &meta);
18531333×
        Ok(meta)
18544705×
    }
1855
1856
    /// The metadata cache: records of a change hash, its metadata in the
1857
    /// flat layout, and the first 8 bytes of BLAKE3 over both, each
1858
    /// appended in one write. Derived only: a record that does not check
1859
    /// ends the reading (a torn write), and the cache can be deleted. It
1860
    /// lives in the scratch directory, on this disk.
1861977×
    fn read_metas_cache(&self) -> crate::fasthash::FastMap<ObjectRef, crate::state::ChangeMeta> {
1862977×
        let mut out = crate::fasthash::FastMap::default();
1863977×
        let Ok(bytes) = fs::read(self.scratch.join("metas")) else {
1864376×
            return out;
1865
        };
1866601×
        let mut r = crate::flat::Reader::new(&bytes);
1867601×
        let mut good = 0;
18682068×
        while !r.at_end() {
18691468×
            let mut record = || -> std::result::Result<_, ObjectError> {
18701468×
                let h = r.object()?;
18711467×
                let body = r.bytes()?;
18721467×
                let check = r.fixed::<8>()?;
18731467×
                Ok((h, body, check))
18741468×
            };
18751468×
            let Ok((h, body, check)) = record() else {
18761×
                break;
1877
            };
18781467×
            let mut sum = blake3::Hasher::new();
18791467×
            sum.update(&h.0).update(body);
18801467×
            if sum.finalize().as_bytes()[..8] != check {
18810×
                break;
18821467×
            }
18831467×
            match crate::state::ChangeMeta::read_flat(&mut crate::flat::Reader::new(body)) {
18841467×
                Ok(m) => out.insert(h, m),
18850×
                Err(_) => break,
1886
            };
18871467×
            good += 32 + 4 + body.len() + 8;
1888
        }
1889
        // Cut a bad tail, so records appended later are read.
1890601×
        if good < bytes.len() {
18911×
            let _ = fs::OpenOptions::new()
18921×
                .write(true)
18931×
                .open(self.scratch.join("metas"))
18941×
                .and_then(|f| f.set_len(good as u64));
1895600×
        }
1896601×
        out
1897977×
    }
1898
1899
    /// Add a change's metadata to the cache; quietly nothing when it
1900
    /// cannot be written (it is only a cache).
19011333×
    fn remember_meta(&self, r: &ObjectRef, meta: &crate::state::ChangeMeta) {
19021333×
        let mut body = crate::flat::Writer::default();
19031333×
        meta.write_flat(&mut body);
19041333×
        let mut w = crate::flat::Writer::default();
19051333×
        w.object(r);
19061333×
        w.bytes(&body.0);
19071333×
        let mut sum = blake3::Hasher::new();
19081333×
        sum.update(&r.0).update(&body.0);
19091333×
        w.fixed(&sum.finalize().as_bytes()[..8]);
19101333×
        let _ = fs::OpenOptions::new()
19111333×
            .create(true)
19121333×
            .append(true)
19131333×
            .open(self.scratch.join("metas"))
19141333×
            .and_then(|mut f| f.write_all(&w.0));
19151333×
        if let Some(m) = self.metas.borrow_mut().as_mut() {
19161333×
            m.insert(*r, meta.clone());
19171333×
        }
19181333×
    }
1919
192059×
    fn read_text(&self, r: &ObjectRef) -> Result<(Blob, Vec<u8>)> {
192159×
        let blob = Blob::decode(&self.get_typed(r, ObjectType::Blob)?)?;
192258×
        let mut content = Vec::with_capacity(blob.length as usize);
192398×
        for c in &blob.chunks {
192498×
            content.extend(self.get_typed(c, ObjectType::Chunk)?);
1925
        }
192658×
        Ok((blob, content))
192759×
    }
1928
1929199×
    pub fn put_chunk(&self, bytes: &[u8]) -> Result<ObjectRef> {
1930199×
        self.put(ObjectType::Chunk, bytes)
1931199×
    }
1932
1933
    /// Store a chunk that packs will keep uncompressed (`nocompress`, spec §13.3).
1934123×
    pub fn put_chunk_nocompress(&self, bytes: &[u8]) -> Result<ObjectRef> {
1935123×
        let r = hash_object(ObjectType::Chunk, bytes);
1936123×
        put_new_or_existing(
1937123×
            self.backend.as_ref(),
1938123×
            &format!("nocompress/{}", r.to_hex()),
1939123×
            b"",
19400×
        )?;
1941123×
        self.put(ObjectType::Chunk, bytes)
1942123×
    }
1943
19443026×
    fn is_nocompress(&self, r: &ObjectRef) -> bool {
19453026×
        self.backend
19463026×
            .has(&format!("nocompress/{}", r.to_hex()))
19473026×
            .unwrap_or(false)
19483026×
    }
1949
1950
    /// Every chunk marked to be kept raw.
1951100×
    fn nocompress_set(&self) -> Result<std::collections::HashSet<ObjectRef>> {
1952100×
        Ok(self
1953100×
            .backend
1954100×
            .list("nocompress/")
1955100×
            .map_err(backend_err)?
1956100×
            .into_iter()
1957100×
            .filter_map(|e| ObjectRef::from_hex(e.key.trim_start_matches("nocompress/")))
1958100×
            .collect())
1959100×
    }
1960
19613155×
    pub fn put_state(&self, st: &StateObject) -> Result<ObjectRef> {
19623155×
        self.put(ObjectType::State, &st.encode())
19633155×
    }
1964
19655638×
    pub fn get_state(&self, r: &ObjectRef) -> Result<StateObject> {
19665638×
        Ok(StateObject::decode(&self.get_typed(r, ObjectType::State)?)?)
19675638×
    }
1968
1969435×
    pub fn put_attest(&self, a: &Attest) -> Result<ObjectRef> {
1970435×
        self.put(ObjectType::Attest, &a.encode())
1971435×
    }
1972
1973685×
    pub fn get_attest(&self, r: &ObjectRef) -> Result<Attest> {
1974685×
        Ok(Attest::decode(&self.get_typed(r, ObjectType::Attest)?)?)
1975685×
    }
1976
19773×
    pub fn put_receipt(&self, r: &Receipt) -> Result<ObjectRef> {
19783×
        self.put(ObjectType::Receipt, &r.encode())
19793×
    }
1980
198112×
    pub fn get_receipt(&self, r: &ObjectRef) -> Result<Receipt> {
198212×
        Ok(Receipt::decode(&self.get_typed(r, ObjectType::Receipt)?)?)
198312×
    }
1984
1985
    /// Record `value` under `key` in `index`: one write-once entry, so two
1986
    /// writers of a shared store never lose each other's.
1987192×
    fn index_add(&self, index: &str, key: &ObjectRef, value: &ObjectRef) -> Result<()> {
1988192×
        let entry = format!("{}{}", fanout_prefix(index, key), value.to_hex());
1989192×
        put_new_or_existing(self.backend.as_ref(), &entry, b"")
1990192×
    }
1991
1992
    /// The values recorded under `key` in `index`, with those an older
1993
    /// store kept as one list in `legacy`, sorted.
19942417×
    fn index_get(&self, index: &str, legacy: &str, key: &ObjectRef) -> Result<Vec<ObjectRef>> {
19952417×
        let mut values: Vec<ObjectRef> = self
19962417×
            .backend
19972417×
            .list(&fanout_prefix(index, key))
19982417×
            .map_err(backend_err)?
19992417×
            .into_iter()
20002417×
            .filter_map(|e| e.key.rsplit('/').next().and_then(ObjectRef::from_hex))
20012417×
            .collect();
20022417×
        if let Some(bytes) = get_optional(self.backend.as_ref(), &legacy_key(legacy, key))? {
20033×
            values.extend(
20043×
                String::from_utf8_lossy(&bytes)
20053×
                    .lines()
20063×
                    .filter_map(ObjectRef::from_hex),
20073×
            );
20082414×
        }
20092417×
        values.sort();
20102417×
        values.dedup();
20112417×
        Ok(values)
20122417×
    }
2013
2014
    /// Record that seal attestation `attest` covers `change`.
2015157×
    pub fn index_seal(&self, change: &ObjectRef, attest: &ObjectRef) -> Result<()> {
2016157×
        self.index_add("seal", change, attest)
2017157×
    }
2018
2019
    /// Seal attestations known to cover `change`.
20201521×
    pub fn seals_for(&self, change: &ObjectRef) -> Result<Vec<ObjectRef>> {
20211521×
        self.index_get("seal", "seals", change)
20221521×
    }
2023
2024
    /// Record receipt attestation `attest` for `state` (spec §14.6).
2025
    /// Keep a short-lived key's certificate (spec §14.7), with its issuer's
2026
    /// signature, indexed by the key it certifies.
202728×
    pub fn put_certificate(&self, c: &crate::cert::SignedCertificate) -> Result<ObjectRef> {
202828×
        let r = self.put(ObjectType::Certificate, &c.cert.encode())?;
202928×
        self.put_signature(&r, &c.sig)?;
203028×
        self.index_add("cert", &c.cert.key_id(), &r)?;
203128×
        Ok(r)
203228×
    }
2033
2034
    /// A kept certificate with its issuer's signature.
203577×
    pub fn get_certificate(&self, r: &ObjectRef) -> Result<crate::cert::SignedCertificate> {
203677×
        let cert = crate::cert::Certificate::decode(&self.get_typed(r, ObjectType::Certificate)?)?;
203777×
        let sig = self
203877×
            .signatures(r)?
203977×
            .into_iter()
204077×
            .find(|s| s.key == cert.issuer)
204177×
            .ok_or(StoreError::NotFound(*r))?;
204277×
        Ok(crate::cert::SignedCertificate { cert, sig })
204377×
    }
2044
2045
    /// The certificate kept for short-lived key `key`, if any.
2046702×
    pub fn certificate_of(&self, key: &ObjectRef) -> Result<Option<ObjectRef>> {
2047702×
        Ok(self
2048702×
            .index_get("cert", "certs", key)?
2049702×
            .into_iter()
2050702×
            .find(|r| self.has(r)))
2051702×
    }
2052
2053
    /// Record `value` under `key` in index `index` (for repository code).
20541×
    pub(crate) fn index_add_pub(
20551×
        &self,
20561×
        index: &str,
20571×
        key: &ObjectRef,
20581×
        value: &ObjectRef,
20591×
    ) -> Result<()> {
20601×
        self.index_add(index, key, value)
20611×
    }
2062
2063
    /// The values recorded under `key` in index `index` (which has no
2064
    /// older one-list form: the legacy name only has to differ from it).
2065177×
    pub(crate) fn index_get_pub(&self, index: &str, key: &ObjectRef) -> Result<Vec<ObjectRef>> {
2066177×
        self.index_get(index, &format!("{index}-lists"), key)
2067177×
    }
2068
20695×
    pub fn index_receipt(&self, state: &ObjectRef, attest: &ObjectRef) -> Result<()> {
20705×
        self.index_add("receipt", state, attest)
20715×
    }
2072
2073
    /// Receipt attestations recorded for `state`.
207414×
    pub fn receipts_for(&self, state: &ObjectRef) -> Result<Vec<ObjectRef>> {
207514×
        self.index_get("receipt", "receipts", state)
207614×
    }
2077
20781×
    pub fn index_approval(&self, state: &ObjectRef, attest: &ObjectRef) -> Result<()> {
20791×
        self.index_add("approval", state, attest)
20801×
    }
2081
2082
    /// Approval attestations recorded for `state` (OxygenFlow approval
2083
    /// gates, oxygenhub#235).
20843×
    pub fn approvals_for(&self, state: &ObjectRef) -> Result<Vec<ObjectRef>> {
20853×
        self.index_get("approval", "approvals", state)
20863×
    }
2087
2088
    /// Every state with an approval, for garbage collection.
208923×
    pub fn approval_states(&self) -> Result<Vec<ObjectRef>> {
209023×
        self.indexed_states("approval", "approvals")
209123×
    }
2092
2093
    /// Every state with a receipt, for garbage collection.
209429×
    pub fn receipt_states(&self) -> Result<Vec<ObjectRef>> {
209529×
        self.indexed_states("receipt", "receipts")
209629×
    }
2097
2098
    /// The states an index (`one/<hh>/<rest>/<attest>` entries, and the
2099
    /// packed `many/<hh>/<rest>` lists) holds attestations for.
210052×
    fn indexed_states(&self, one: &str, many: &str) -> Result<Vec<ObjectRef>> {
210152×
        let mut out: Vec<ObjectRef> = Vec::new();
210252×
        for e in self.backend.list(&format!("{one}/")).map_err(backend_err)? {
2103
            // receipt/<hh>/<rest>/<attest>
21045×
            let mut parts = e.key.splitn(4, '/');
21055×
            let (Some(_), Some(hh), Some(rest), Some(_)) =
21065×
                (parts.next(), parts.next(), parts.next(), parts.next())
2107
            else {
21080×
                continue;
2109
            };
21105×
            out.extend(ObjectRef::from_hex(&format!("{hh}{rest}")));
2111
        }
211252×
        for e in self
211352×
            .backend
211452×
            .list(&format!("{many}/"))
211552×
            .map_err(backend_err)?
2116
        {
21171×
            let mut parts = e.key.splitn(3, '/');
21181×
            let (Some(_), Some(hh), Some(rest)) = (parts.next(), parts.next(), parts.next()) else {
21190×
                continue;
2120
            };
21211×
            out.extend(ObjectRef::from_hex(&format!("{hh}{rest}")));
2122
        }
212352×
        out.sort();
212452×
        out.dedup();
212552×
        Ok(out)
212652×
    }
2127
2128335×
    pub fn put_identity(&self, id: &Identity) -> Result<ObjectRef> {
2129335×
        self.put(ObjectType::Identity, &id.encode())
2130335×
    }
2131
2132870×
    pub fn get_identity(&self, r: &ObjectRef) -> Result<Identity> {
2133870×
        Ok(Identity::decode(&self.get_typed(r, ObjectType::Identity)?)?)
2134870×
    }
2135
2136
    /// Signatures stored alongside object `r` (spec §5.4), one per signing
2137
    /// key, sorted by key. Empty if none.
21383255×
    pub fn signatures(&self, r: &ObjectRef) -> Result<Vec<Signature>> {
21393255×
        let mut sigs: Vec<Signature> = Vec::new();
21403255×
        for e in self
21413255×
            .backend
21423255×
            .list(&fanout_prefix("sig", r))
21433255×
            .map_err(backend_err)?
2144
        {
21451962×
            let bytes = self.backend.get(&e.key).map_err(backend_err)?;
21461962×
            sigs.extend(Signature::decode_list(&bytes)?);
2147
        }
2148
        // An older store's one list per object; a signer kept there and
2149
        // here counts once, as written here.
21503255×
        if let Some(bytes) = get_optional(self.backend.as_ref(), &legacy_key("sigs", r))? {
21512×
            for sig in Signature::decode_list(&bytes)? {
21522×
                if !sigs.iter().any(|s| s.key == sig.key) {
21531×
                    sigs.push(sig);
21541×
                }
2155
            }
21563253×
        }
21573255×
        sigs.sort_by_key(|s| s.key);
21583255×
        sigs.dedup_by_key(|s| s.key);
21593255×
        Ok(sigs)
21603255×
    }
2161
2162
    /// Attach a signature to object `r`, replacing any earlier one by the
2163
    /// same key: one entry per signer, so the latest by a key is the one
2164
    /// kept, and two signers never touch each other's.
2165688×
    pub fn put_signature(&self, r: &ObjectRef, sig: &Signature) -> Result<()> {
2166688×
        let key = format!("{}{}", fanout_prefix("sig", r), sig.key.to_hex());
2167688×
        let bytes = Signature::encode_list(std::slice::from_ref(sig));
2168688×
        if get_optional(self.backend.as_ref(), &key)?.as_deref() == Some(bytes.as_slice()) {
216945×
            return Ok(());
2170643×
        }
2171643×
        delete(self.backend.as_ref(), &key)?;
2172643×
        put_new_or_existing(self.backend.as_ref(), &key, &bytes)
2173688×
    }
2174
2175
    /// Finish packs a writer left unfinished: a batch writes its objects to
2176
    /// `.tmp-ID.pack` and only indexes it at the end, so a process that
2177
    /// stopped part way (an import killed after hours) leaves every object
2178
    /// in a pack nothing can read. Each such pack untouched for `min_age`
2179
    /// is read through: every whole entry checked against its reference,
2180
    /// a torn last one cut off, then indexed as a pack of the store.
2181
    /// Returns the packs and objects adopted.
21822×
    pub fn adopt_unfinished_packs(
21832×
        &self,
21842×
        min_age: std::time::Duration,
21852×
        progress: &dyn Fn(&str),
21862×
    ) -> Result<(usize, usize)> {
2187
        use std::io::{BufReader, Read};
21882×
        let mut packs = 0;
21892×
        let mut objects = 0;
21902×
        let Ok(dir) = fs::read_dir(&self.scratch) else {
21910×
            return Ok((0, 0));
2192
        };
21934×
        for entry in dir.flatten() {
21944×
            let name = entry.file_name().to_string_lossy().to_string();
21954×
            let Some(id) = name
21964×
                .strip_prefix(".tmp-")
21974×
                .and_then(|n| n.strip_suffix(".pack"))
2198
            else {
21992×
                continue;
2200
            };
22012×
            let path = entry.path();
22022×
            let meta = fs::metadata(&path)?;
22032×
            let age = meta
22042×
                .modified()
22052×
                .ok()
22062×
                .and_then(|m| m.elapsed().ok())
22072×
                .unwrap_or_default();
22082×
            if age < min_age {
22091×
                progress(&format!(
22101×
                    "{name}: written to {}s ago, left alone",
22111×
                    age.as_secs()
22121×
                ));
22131×
                continue;
22141×
            }
22151×
            let mut f = BufReader::with_capacity(1 << 20, fs::File::open(&path)?);
22161×
            let mut magic = [0u8; 8];
22171×
            if f.read_exact(&mut magic).is_err() || &magic != PACK_MAGIC {
22180×
                progress(&format!("{name}: not a pack, left alone"));
22190×
                continue;
22201×
            }
22211×
            let mut offset = PACK_MAGIC.len() as u64;
22221×
            let mut index: Vec<(ObjectRef, PackEntry)> = Vec::new();
2223
            loop {
2224301×
                let mut head = [0u8; 42];
2225301×
                if f.read_exact(&mut head).is_err() {
22260×
                    break;
2227301×
                }
2228301×
                let Some(ty) = ObjectType::ALL.get(head[0] as usize).copied() else {
22290×
                    break;
2230
                };
2231301×
                let flags = head[1];
2232301×
                let stored_len = u32::from_le_bytes(head[2..6].try_into().unwrap());
2233301×
                let raw_len = u32::from_le_bytes(head[6..10].try_into().unwrap());
2234301×
                let r = ObjectRef(head[10..42].try_into().unwrap());
2235301×
                let mut stored = vec![0u8; stored_len as usize];
2236301×
                if flags & FLAG_DICT != 0 || f.read_exact(&mut stored).is_err() {
22371×
                    break;
2238300×
                }
2239300×
                let e = PackEntry {
2240300×
                    offset: offset + 42,
2241300×
                    stored_len,
2242300×
                    raw_len,
2243300×
                    ty,
2244300×
                    flags,
2245300×
                };
2246300×
                if decode_entry(&r, e, stored, None).is_err() {
22470×
                    break;
2248300×
                }
2249300×
                index.push((r, e));
2250300×
                offset += 42 + stored_len as u64;
2251300×
                if index.len().is_multiple_of(200_000) {
22520×
                    progress(&format!(
22530×
                        "{name}: {} objects checked, {} MiB",
22540×
                        index.len(),
22550×
                        offset >> 20
22560×
                    ));
2257300×
                }
2258
            }
22591×
            if index.is_empty() {
22600×
                progress(&format!("{name}: no whole object, left alone"));
22610×
                continue;
22621×
            }
22631×
            if offset < meta.len() {
22641×
                progress(&format!(
22651×
                    "{name}: {} torn bytes at the end cut off",
22661×
                    meta.len() - offset
22671×
                ));
22681×
                fs::OpenOptions::new()
22691×
                    .write(true)
22701×
                    .open(&path)?
22711×
                    .set_len(offset)?;
22720×
            }
22731×
            index.sort_by_key(|(r, _)| *r);
22741×
            index.dedup_by_key(|(r, _)| *r);
22751×
            objects += index.len();
22761×
            packs += 1;
22771×
            progress(&format!("{name}: {} objects indexed", index.len()));
22781×
            self.finish_pack_files(id, &path, &index)?;
2279
        }
22802×
        Ok((packs, objects))
22812×
    }
2282
2283
    /// Every object reference in the store, loose and packed (slow; for gc and tests).
228486×
    pub fn list(&self) -> io::Result<Vec<ObjectRef>> {
228586×
        let mut out: Vec<ObjectRef> = self
228686×
            .loose_objects()
228786×
            .map_err(|e| io::Error::other(e.to_string()))?
228886×
            .into_iter()
228986×
            .map(|(r, _)| r)
229086×
            .collect();
229186×
        for p in self.packs.borrow().iter() {
229210×
            out.extend(p.refs.iter().copied());
229310×
        }
229486×
        out.sort();
229586×
        out.dedup();
229686×
        Ok(out)
229786×
    }
2298
}
2299
2300
/// Whether a read failed in a way a newer list of packs may cure: the
2301
/// object was not found, or its pack's dictionary is gone.
2302341×
fn is_miss(e: &StoreError) -> bool {
2303341×
    match e {
2304213×
        StoreError::NotFound(_) => true,
2305128×
        StoreError::Corrupt(_, why) => why == NO_DICT,
23060×
        _ => false,
2307
    }
2308341×
}
2309
2310
/// A loose object's bytes, verified against `r`.
231152915×
fn parse_loose(data: &[u8], r: &ObjectRef) -> Result<Object> {
231252915×
    let nul = data
231352915×
        .iter()
2314753836×
        .position(|&b| b == 0)
231552915×
        .ok_or_else(|| StoreError::Corrupt(*r, "no header".into()))?;
231652915×
    let head = std::str::from_utf8(&data[..nul])
231752915×
        .map_err(|_| StoreError::Corrupt(*r, "header not UTF-8".into()))?;
231852915×
    let mut parts = head.split(' ');
231952915×
    let (Some("oxy1"), Some(ty), Some(len), None) =
232052915×
        (parts.next(), parts.next(), parts.next(), parts.next())
2321
    else {
23220×
        return Err(StoreError::Corrupt(*r, format!("bad header {head:?}")));
2323
    };
232452915×
    let ty = ObjectType::from_name(ty)
232552915×
        .ok_or_else(|| StoreError::Corrupt(*r, format!("unknown type {ty}")))?;
232652915×
    let len: usize = len
232752915×
        .parse()
232852915×
        .map_err(|_| StoreError::Corrupt(*r, "bad length".into()))?;
232952915×
    let body = &data[nul + 1..];
233052915×
    if body.len() != len {
23310×
        return Err(StoreError::Corrupt(
23320×
            *r,
23330×
            format!("length {} != header {len}", body.len()),
23340×
        ));
233552915×
    }
233652915×
    if hash_object(ty, body) != *r {
23374×
        return Err(StoreError::Corrupt(*r, "hash mismatch".into()));
233852911×
    }
233952911×
    Ok((ty, body.to_vec()))
234052915×
}
2341
2342
/// Whether the objects are files that a test may tamper with: not under
2343
/// `OXY_STORE_BACKEND=memory`.
2344
#[cfg(test)]
23456×
fn files_on_disk() -> bool {
23466×
    std::env::var_os("OXY_STORE_BACKEND").is_none()
23476×
}
2348
2349
#[cfg(test)]
2350
mod tests {
2351
    use super::*;
2352
2353
    /// The store under test twice: in a directory, and on a store in
2354
    /// memory with a scratch directory, as over a bucket.
23556×
    fn each_store(dir: &Path, f: impl Fn(&str, Store)) {
23566×
        f("directory", Store::open(dir.join("objects")).unwrap());
23576×
        f(
23586×
            "memory",
23596×
            Store::open_on(Arc::new(MemoryStore::new()), dir.join("scratch")).unwrap(),
23606×
        );
23616×
    }
2362
2363
    #[test]
23641×
    fn put_get_verify() {
23651×
        if !super::files_on_disk() {
23660×
            return;
23671×
        }
23681×
        let dir = tempfile::tempdir().unwrap();
23691×
        let s = Store::open(dir.path().join("objects")).unwrap();
23701×
        let r = s.put(ObjectType::Chunk, b"hello").unwrap();
23711×
        assert!(s.has(&r));
23721×
        assert_eq!(s.get(&r).unwrap(), (ObjectType::Chunk, b"hello".to_vec()));
23731×
        assert_eq!(s.put(ObjectType::Chunk, b"hello").unwrap(), r);
23741×
        assert_eq!(s.list().unwrap(), vec![r]);
23751×
        assert!(matches!(
23761×
            s.get_typed(&r, ObjectType::Blob),
2377
            Err(StoreError::WrongType(..))
2378
        ));
23791×
        assert!(matches!(
23801×
            s.get(&ObjectRef([9; 32])),
2381
            Err(StoreError::NotFound(_))
2382
        ));
2383
2384
        // Tamper: flip a byte, read must fail.
23851×
        let path = s.path_for(&r);
23861×
        let mut data = fs::read(&path).unwrap();
23871×
        let last = data.len() - 1;
23881×
        data[last] ^= 1;
23891×
        fs::write(&path, data).unwrap();
23901×
        assert!(matches!(s.get(&r), Err(StoreError::Corrupt(..))));
23911×
    }
2392
2393
    #[test]
23941×
    fn packing_keeps_every_object_readable() {
23951×
        if !super::files_on_disk() {
23960×
            return;
23971×
        }
23981×
        let dir = tempfile::tempdir().unwrap();
23991×
        let s = Store::open(dir.path()).unwrap();
24001×
        let mut refs = Vec::new();
2401300×
        for i in 0..300u32 {
2402300×
            let body = format!("object {i} ").repeat(1 + (i % 7) as usize);
2403300×
            refs.push((s.put(ObjectType::Chunk, body.as_bytes()).unwrap(), body));
2404300×
        }
24051×
        let c = crate::objects::tests::sample_change();
24061×
        let cr = s.put_change(&c).unwrap();
24071×
        assert_eq!(s.loose_count().unwrap(), 301);
24081×
        let report = s.pack_loose().unwrap();
24091×
        assert_eq!(report.objects, 301);
24101×
        assert!(report.packed_bytes < report.raw_bytes, "{report:?}");
24111×
        assert_eq!(s.loose_count().unwrap(), 0);
24121×
        assert_eq!(s.pack_count(), 1);
2413300×
        for (r, body) in &refs {
2414300×
            assert!(s.has(r));
2415300×
            assert_eq!(
2416300×
                s.get(r).unwrap(),
2417300×
                (ObjectType::Chunk, body.as_bytes().to_vec())
2418
            );
2419
        }
24201×
        assert_eq!(s.get_change(&cr).unwrap(), c);
24211×
        assert_eq!(s.list().unwrap().len(), 301);
2422
        // Re-putting a packed object is a no-op; new objects go loose and a
2423
        // second pack coexists with the first.
24241×
        assert_eq!(
24251×
            s.put(ObjectType::Chunk, refs[0].1.as_bytes()).unwrap(),
24261×
            refs[0].0
2427
        );
24281×
        assert_eq!(s.loose_count().unwrap(), 0);
24291×
        let extra = s.put(ObjectType::Chunk, b"after packing").unwrap();
24301×
        s.pack_loose().unwrap();
24311×
        assert_eq!(s.pack_count(), 2);
24321×
        assert_eq!(s.get(&extra).unwrap().1, b"after packing");
2433
        // A reopened store sees the packs.
24341×
        let s2 = Store::open(dir.path()).unwrap();
24351×
        assert_eq!(s2.pack_count(), 2);
24361×
        assert_eq!(s2.get(&refs[5].0).unwrap().1, refs[5].1.as_bytes());
24371×
        assert!(matches!(
24381×
            s2.get(&ObjectRef([9; 32])),
2439
            Err(StoreError::NotFound(_))
2440
        ));
2441
        // Corrupting an object's body inside the pack is detected on read.
24421×
        let target = refs[5].0;
24431×
        let (pack_path, offset) = {
24441×
            let packs = s2.packs.borrow();
24451×
            let (i, e) = s2.find_packed(&target).unwrap();
24461×
            (dir.path().join(&packs[i].key), e.offset as usize)
24471×
        };
24481×
        let mut data = fs::read(&pack_path).unwrap();
24491×
        data[offset] ^= 0xff;
24501×
        fs::write(&pack_path, data).unwrap();
24511×
        let s3 = Store::open(dir.path()).unwrap();
24521×
        assert!(matches!(s3.get(&target), Err(StoreError::Corrupt(..))));
24531×
        assert_eq!(
24541×
            s3.get(&refs[6].0).unwrap().1,
24551×
            refs[6].1.as_bytes(),
2456
            "other objects are unaffected"
2457
        );
24581×
    }
2459
2460
    #[test]
24611×
    fn a_full_repack_trains_a_dictionary_and_reads_back() {
24621×
        if !super::files_on_disk() {
24630×
            return;
24641×
        }
24651×
        let dir = tempfile::tempdir().unwrap();
24661×
        let s = Store::open(dir.path()).unwrap();
2467
        // Many small, alike files: what a dictionary is for.
24681×
        let bodies: Vec<Vec<u8>> = (0..1500)
24691500×
            .map(|i| {
24701500×
                format!(
2471
                    "// SPDX-License-Identifier: GPL-2.0\n#include <linux/module.h>\n\
2472
                     static int probe_{i}(struct device *dev)\n{{\n\treturn {i};\n}}\n"
2473
                )
24741500×
                .into_bytes()
24751500×
            })
24761×
            .collect();
24771×
        let refs: Vec<ObjectRef> = bodies
24781×
            .iter()
24791500×
            .map(|b| s.put(ObjectType::Chunk, b).unwrap())
24801×
            .collect();
24811×
        s.repack_all(REPACK_ZSTD_LEVEL).unwrap();
24821×
        let dicts = fs::read_dir(dir.path())
24831×
            .unwrap()
24841×
            .flatten()
24855×
            .filter(|e| e.path().extension().is_some_and(|x| x == "dict"))
24861×
            .count();
24871×
        assert_eq!(dicts, 1, "the pack has a dictionary");
24881×
        let with_dict = refs
24891×
            .iter()
24901500×
            .filter(|r| s.find_packed(r).unwrap().1.flags & FLAG_DICT != 0)
24911×
            .count();
24921×
        assert!(with_dict > 1000, "most use it: {with_dict}");
24931×
        let again = Store::open(dir.path()).unwrap();
24941500×
        for (r, b) in refs.iter().zip(&bodies) {
24951500×
            assert_eq!(&again.get(r).unwrap().1, b);
2496
        }
24974×
        let packs = |dir: &Path| {
24984×
            let mut v: Vec<String> = fs::read_dir(dir)
24994×
                .unwrap()
25004×
                .flatten()
250122×
                .map(|e| e.file_name().to_string_lossy().into_owned())
250222×
                .filter(|n| n.starts_with("pack-"))
25034×
                .collect();
25044×
            v.sort();
25054×
            v
25064×
        };
2507
        // With nothing new, a repack leaves that pack as it is: it would
2508
        // only write the same again.
25091×
        let before = packs(dir.path());
25101×
        let (report, merged) = again.repack_all(REPACK_ZSTD_LEVEL).unwrap();
25111×
        assert_eq!((report.objects, merged), (0, 0));
25121×
        assert_eq!(packs(dir.path()), before);
2513
        // One new object: a repack replaces the pack and its dictionary,
2514
        // retiring the old ones, which a collection an hour on deletes.
25151×
        let extra = again.put(ObjectType::Chunk, b"one more\n").unwrap();
25161×
        again.repack_all(REPACK_ZSTD_LEVEL).unwrap();
25171×
        let files = packs(dir.path());
25181×
        assert_eq!(
25191×
            files.len(),
2520
            7,
2521
            "two of pack, index and dictionary, and the retired record: {files:?}"
2522
        );
25231×
        assert_eq!(again.pack_count(), 1, "the old pack is not read");
25241×
        assert_eq!(&again.get(&refs[7]).unwrap().1, &bodies[7]);
25251×
        assert_eq!(again.get(&extra).unwrap().1, b"one more\n");
25261×
        again
25271×
            .expire_retired(std::time::Duration::ZERO, &mut silent())
25281×
            .unwrap();
25291×
        let files = packs(dir.path());
25301×
        assert_eq!(files.len(), 4, "one pack, index, dictionary: {files:?}");
25311×
        assert!(before
25321×
            .iter()
25334×
            .filter(|f| *f != RETIRED_KEY)
25343×
            .all(|f| !files.contains(f)));
25351×
        assert_eq!(&again.get(&refs[8]).unwrap().1, &bodies[8]);
25361×
    }
2537
2538
    #[test]
25391×
    fn nocompress_chunks_pack_raw() {
25401×
        let dir = tempfile::tempdir().unwrap();
25411×
        let s = Store::open(dir.path()).unwrap();
25421×
        let body = vec![b'a'; 10_000];
25431×
        let raw = s.put_chunk_nocompress(&body).unwrap();
25441×
        let compressible = s.put(ObjectType::Chunk, &vec![b'b'; 10_000]).unwrap();
25451×
        s.pack_loose().unwrap();
25461×
        let packs = s.packs.borrow();
25471×
        let (i, e) = s.find_packed(&raw).unwrap();
25481×
        assert_eq!((e.flags, e.stored_len as usize), (0, 10_000));
25491×
        let (_, c) = s.find_packed(&compressible).unwrap();
25501×
        assert_eq!(c.flags, FLAG_ZSTD);
25511×
        assert!(c.stored_len < 10_000);
25521×
        assert_eq!(packs[i].fetch(s.backend.as_ref(), &raw, e).unwrap().1, body);
25531×
    }
2554
2555
    #[test]
25561×
    fn many_packs_merge_into_one() {
25571×
        let dir = tempfile::tempdir().unwrap();
25582×
        each_store(dir.path(), |which, s| {
25592×
            let mut refs = Vec::new();
256036×
            for i in 0..(MAX_PACKS + 2) {
256136×
                refs.push(
256236×
                    s.put(ObjectType::Chunk, format!("pack {i}").as_bytes())
256336×
                        .unwrap(),
256436×
                );
256536×
                s.pack_loose().unwrap();
256636×
            }
25672×
            assert_eq!(s.pack_count(), MAX_PACKS + 2, "{which}");
25682×
            assert_eq!(s.merge_packs().unwrap(), MAX_PACKS + 2);
25692×
            assert_eq!(s.pack_count(), 1);
25702×
            assert_eq!(s.merge_packs().unwrap(), 0);
257136×
            for (i, r) in refs.iter().enumerate() {
257236×
                assert_eq!(s.get(r).unwrap().1, format!("pack {i}").as_bytes());
2573
            }
25742×
            let again = Store::open_on(s.backend().clone(), s.dir()).unwrap();
25752×
            assert_eq!(again.pack_count(), 1, "{which}: reopened");
25762×
        });
25771×
    }
2578
2579
    #[test]
25801×
    fn prune_removes_unreachable_loose_and_packed_objects() {
2581
        use std::collections::HashSet;
25821×
        let dir = tempfile::tempdir().unwrap();
25832×
        each_store(dir.path(), |which, s| {
25842×
            let keep_a = s.put(ObjectType::Chunk, b"keep a").unwrap();
25852×
            let drop_a = s.put(ObjectType::Chunk, b"drop a").unwrap();
25862×
            s.pack_loose().unwrap();
25872×
            let keep_b = s.put(ObjectType::Chunk, b"keep b").unwrap();
25882×
            let drop_b = s.put(ObjectType::Chunk, b"drop b").unwrap();
25892×
            let sk = crate::identity::SecretKey {
25902×
                ed_seed: [1; 32],
25912×
                ml_seed: [2; 32],
25922×
            };
25932×
            let id = sk.identity("a", "b", 0);
25942×
            let key = s.put_identity(&id).unwrap();
25952×
            s.put_signature(
25962×
                &drop_b,
25972×
                &sk.sign(crate::identity::Domain::Change, &key, &drop_b),
2598
            )
25992×
            .unwrap();
26002×
            s.index_seal(&drop_a, &ObjectRef([3; 32])).unwrap();
26012×
            let keep: HashSet<ObjectRef> = [keep_a, keep_b, key].into_iter().collect();
2602
            // Everything is too young under a grace period.
26032×
            let r = s
26042×
                .prune(&keep, std::time::Duration::from_secs(3600))
26052×
                .unwrap();
26062×
            assert_eq!((r.objects, r.deferred), (0, 2), "{which}");
26072×
            let r = s.prune(&keep, std::time::Duration::ZERO).unwrap();
26082×
            assert_eq!(r.objects, 2, "{which}: {r:?}");
26092×
            assert!(s.has(&keep_a) && s.has(&keep_b) && s.has(&key));
26102×
            assert!(!s.has(&drop_a) && !s.has(&drop_b));
26112×
            assert!(s.signatures(&drop_b).unwrap().is_empty());
26122×
            assert!(s.seals_for(&drop_a).unwrap().is_empty());
26132×
            assert_eq!(s.pack_count(), 1);
26142×
            assert_eq!(s.get(&keep_a).unwrap().1, b"keep a");
26152×
            assert_eq!(s.list().unwrap().len(), 3);
26162×
        });
26171×
    }
2618
2619
    #[test]
26201×
    fn signatures_are_stored_beside_objects() {
2621
        use crate::identity::{Domain, SecretKey};
26221×
        let dir = tempfile::tempdir().unwrap();
26232×
        each_store(dir.path(), |which, s| {
26242×
            let r = s.put(ObjectType::Chunk, b"x").unwrap();
26252×
            assert!(s.signatures(&r).unwrap().is_empty());
26262×
            let sk = SecretKey {
26272×
                ed_seed: [1; 32],
26282×
                ml_seed: [2; 32],
26292×
            };
26302×
            let id = sk.identity("a", "b", 0);
26312×
            let key = s.put_identity(&id).unwrap();
26322×
            assert_eq!(s.get_identity(&key).unwrap(), id);
26332×
            let sig = sk.sign(Domain::Change, &key, &r);
26342×
            s.put_signature(&r, &sig).unwrap();
26352×
            s.put_signature(&r, &sig).unwrap();
26362×
            assert_eq!(s.signatures(&r).unwrap(), vec![sig.clone()], "{which}");
2637
            // A second signer is kept beside the first, sorted by key.
26382×
            let sk2 = SecretKey {
26392×
                ed_seed: [3; 32],
26402×
                ml_seed: [4; 32],
26412×
            };
26422×
            let key2 = s.put_identity(&sk2.identity("c", "d", 0)).unwrap();
26432×
            let sig2 = sk2.sign(Domain::Change, &key2, &r);
26442×
            s.put_signature(&r, &sig2).unwrap();
26452×
            let mut both = vec![sig.clone(), sig2.clone()];
26462×
            both.sort_by_key(|s| s.key);
26472×
            assert_eq!(s.signatures(&r).unwrap(), both);
2648
            // The object itself is untouched by its signatures.
26492×
            assert_eq!(s.get(&r).unwrap().1, b"x");
2650
            // Seal and receipt entries, and the states with receipts.
26512×
            s.index_seal(&r, &ObjectRef([5; 32])).unwrap();
26522×
            s.index_seal(&r, &ObjectRef([4; 32])).unwrap();
26532×
            s.index_seal(&r, &ObjectRef([5; 32])).unwrap();
26542×
            assert_eq!(
26552×
                s.seals_for(&r).unwrap(),
26562×
                vec![ObjectRef([4; 32]), ObjectRef([5; 32])]
2657
            );
26582×
            let st = ObjectRef([6; 32]);
26592×
            s.index_receipt(&st, &ObjectRef([7; 32])).unwrap();
26602×
            assert_eq!(s.receipts_for(&st).unwrap(), vec![ObjectRef([7; 32])]);
26612×
            assert_eq!(s.receipt_states().unwrap(), vec![st]);
26622×
        });
26631×
    }
2664
2665
    /// A store written before signatures, seals and receipts were one
2666
    /// entry each (`sigs/`, `seals/`, `receipts/` lists) still reads, and
2667
    /// what is written now sits beside what was there.
2668
    #[test]
26691×
    fn an_older_store_s_sidecar_lists_are_still_read() {
26701×
        if !super::files_on_disk() {
26710×
            return;
26721×
        }
2673
        use crate::identity::{Domain, SecretKey};
26741×
        let dir = tempfile::tempdir().unwrap();
26751×
        let s = Store::open(dir.path()).unwrap();
26761×
        let r = s.put(ObjectType::Chunk, b"old").unwrap();
26771×
        let h = r.to_hex();
26781×
        let sk = SecretKey {
26791×
            ed_seed: [1; 32],
26801×
            ml_seed: [2; 32],
26811×
        };
26821×
        let key = s.put_identity(&sk.identity("a", "b", 0)).unwrap();
26831×
        let sig = sk.sign(Domain::Change, &key, &r);
26841×
        let legacy = dir.path().join("sigs").join(&h[..2]);
26851×
        fs::create_dir_all(&legacy).unwrap();
26861×
        fs::write(
26871×
            legacy.join(&h[2..]),
26881×
            Signature::encode_list(std::slice::from_ref(&sig)),
2689
        )
26901×
        .unwrap();
26911×
        let seals = dir.path().join("seals").join(&h[..2]);
26921×
        fs::create_dir_all(&seals).unwrap();
26931×
        fs::write(
26941×
            seals.join(&h[2..]),
26951×
            format!("{}\n", ObjectRef([9; 32]).to_hex()),
2696
        )
26971×
        .unwrap();
26981×
        let receipts = dir.path().join("receipts").join(&h[..2]);
26991×
        fs::create_dir_all(&receipts).unwrap();
27001×
        fs::write(
27011×
            receipts.join(&h[2..]),
27021×
            format!("{}\n", ObjectRef([8; 32]).to_hex()),
2703
        )
27041×
        .unwrap();
2705
27061×
        assert_eq!(s.signatures(&r).unwrap(), vec![sig.clone()]);
27071×
        assert_eq!(s.seals_for(&r).unwrap(), vec![ObjectRef([9; 32])]);
27081×
        assert_eq!(s.receipts_for(&r).unwrap(), vec![ObjectRef([8; 32])]);
27091×
        assert_eq!(s.receipt_states().unwrap(), vec![r]);
2710
        // New entries join the old ones; the same signer is not doubled.
27111×
        s.put_signature(&r, &sig).unwrap();
27121×
        s.index_seal(&r, &ObjectRef([1; 32])).unwrap();
27131×
        assert_eq!(s.signatures(&r).unwrap().len(), 1);
27141×
        assert_eq!(
27151×
            s.seals_for(&r).unwrap(),
27161×
            vec![ObjectRef([1; 32]), ObjectRef([9; 32])]
2717
        );
2718
        // Pruning the object takes both layouts with it.
27191×
        let r2 = s.put(ObjectType::Chunk, b"kept").unwrap();
27201×
        let keep = [r2, key].into_iter().collect();
27211×
        s.prune(&keep, std::time::Duration::ZERO).unwrap();
27221×
        assert!(s.signatures(&r).unwrap().is_empty());
27231×
        assert!(s.seals_for(&r).unwrap().is_empty());
27241×
        assert!(s.receipt_states().unwrap().is_empty());
27251×
    }
2726
2727
    #[test]
27281×
    fn change_round_trip_through_store() {
27291×
        let dir = tempfile::tempdir().unwrap();
27302×
        each_store(dir.path(), |which, s| {
27312×
            let c = crate::objects::tests::sample_change();
27322×
            let r = s.put_change(&c).unwrap();
27332×
            assert_eq!(r, c.hash(), "{which}");
27342×
            assert_eq!(s.get_change(&r).unwrap(), c);
27352×
            assert_eq!(s.get_change_meta(&r).unwrap().message, c.message);
27362×
        });
27371×
    }
2738
27398057203×
    fn xorshift(state: &mut u64) -> u64 {
27408057203×
        *state ^= *state << 13;
27418057203×
        *state ^= *state >> 7;
27428057203×
        *state ^= *state << 17;
27438057203×
        *state
27448057203×
    }
2745
2746
    /// Random corruption of pack and index files never panics: reads fail
2747
    /// with an error or the store refuses to open, and untouched objects
2748
    /// stay readable when the index survives.
2749
    #[test]
27501×
    fn corrupted_packs_and_indexes_fail_cleanly() {
27511×
        if !super::files_on_disk() {
27520×
            return;
27531×
        }
27541×
        let dir = tempfile::tempdir().unwrap();
27551×
        let objects = dir.path().join("objects");
27561×
        let s = Store::open(&objects).unwrap();
27571×
        let mut refs = Vec::new();
27581×
        let mut seed = 0x9e3779b97f4a7c15u64;
275940×
        for i in 0..40u32 {
276040×
            let len = (xorshift(&mut seed) % 5000) as usize;
276140×
            let mut body = vec![0u8; len];
2762104686×
            for b in body.iter_mut() {
2763104686×
                *b = if i % 2 == 0 {
276456338×
                    (xorshift(&mut seed) & 0xff) as u8
2765
                } else {
276648348×
                    b'a' + (i % 26) as u8
2767
                };
2768
            }
276940×
            refs.push(s.put(ObjectType::Chunk, &body).unwrap());
2770
        }
27711×
        s.pack_loose().unwrap();
27721×
        let pack = fs::read_dir(&objects)
27731×
            .unwrap()
27743×
            .filter_map(|e| e.ok().map(|e| e.path()))
27753×
            .find(|p| p.extension().is_some_and(|x| x == "pack"))
27761×
            .unwrap();
27771×
        let idx = pack.with_extension("idx");
27781×
        let pristine_pack = fs::read(&pack).unwrap();
27791×
        let pristine_idx = fs::read(&idx).unwrap();
2780
2781300×
        for round in 0..300u32 {
2782300×
            let (path, pristine) = if round % 2 == 0 {
2783150×
                (&pack, &pristine_pack)
2784
            } else {
2785150×
                (&idx, &pristine_idx)
2786
            };
2787300×
            let mut data = pristine.clone();
2788300×
            match xorshift(&mut seed) % 4 {
278981×
                0 => {
279081×
                    let at = (xorshift(&mut seed) as usize) % data.len();
279181×
                    data[at] ^= 1 << (xorshift(&mut seed) % 8);
279281×
                }
279373×
                1 => {
279473×
                    let at = (xorshift(&mut seed) as usize) % data.len();
279573×
                    data[at] = (xorshift(&mut seed) & 0xff) as u8;
279673×
                }
279775×
                2 => {
279875×
                    let keep = (xorshift(&mut seed) as usize) % data.len();
279975×
                    data.truncate(keep);
280075×
                }
2801
                _ => {
280271×
                    let at = (xorshift(&mut seed) as usize) % data.len();
280371×
                    let n = ((xorshift(&mut seed) % 8) + 1) as usize;
2804324×
                    for k in 0..n.min(data.len() - at) {
2805324×
                        data[at + k] = 0xff;
2806324×
                    }
2807
                }
2808
            }
2809300×
            fs::write(path, &data).unwrap();
2810
            // Opening may refuse a broken index; a store that opens must
2811
            // either return each object intact or report it corrupt/missing.
2812300×
            if let Ok(s2) = Store::open(&objects) {
28137640×
                for r in &refs {
28147640×
                    match s2.get(r) {
28157517×
                        Ok((ty, body)) => assert_eq!(
28167517×
                            hash_object(ty, &body),
2817
                            *r,
2818
                            "a returned object always matches its reference"
2819
                        ),
2820
                        Err(StoreError::Corrupt(..))
2821
                        | Err(StoreError::NotFound(_))
2822123×
                        | Err(StoreError::Io(_)) => {}
28230×
                        Err(e) => panic!("unexpected error {e}"),
2824
                    }
2825
                }
2826109×
            }
2827300×
            fs::write(path, pristine).unwrap();
2828
        }
2829
        // Restored files read back in full.
28301×
        let s3 = Store::open(&objects).unwrap();
283140×
        for r in &refs {
283240×
            let (ty, body) = s3.get(r).unwrap();
283340×
            assert_eq!(hash_object(ty, &body), *r);
2834
        }
28351×
    }
2836
2837
    /// A small batch becomes loose objects; a large one becomes a single
2838
    /// pack. Objects are readable while pending, and batches nest.
2839
    #[test]
28401×
    fn batches_write_loose_or_one_pack() {
28411×
        let dir = tempfile::tempdir().unwrap();
28422×
        each_store(dir.path(), |which, s| {
28432×
            let outer = s.begin_batch().unwrap();
28442×
            let a = s.put(ObjectType::Chunk, b"small a").unwrap();
2845
            {
28462×
                let inner = s.begin_batch().unwrap();
28472×
                let b = s.put(ObjectType::Chunk, b"small b").unwrap();
28482×
                assert!(s.has(&b));
28492×
                assert_eq!(s.get(&b).unwrap().1, b"small b");
28502×
                inner.finish().unwrap();
28512×
                assert_eq!(
28522×
                    s.loose_count().unwrap(),
2853
                    0,
2854
                    "{which}: the inner batch defers to the outer one"
2855
                );
2856
            }
28572×
            assert_eq!(s.get(&a).unwrap().1, b"small a");
28582×
            assert_eq!(
28592×
                s.put(ObjectType::Chunk, b"small a").unwrap(),
2860
                a,
2861
                "idempotent while pending"
2862
            );
28632×
            outer.finish().unwrap();
28642×
            assert_eq!(s.loose_count().unwrap(), 2);
28652×
            assert_eq!(s.pack_count(), 0);
28662×
            assert_eq!(s.get(&a).unwrap().1, b"small a");
2867
2868
            // Large: spills into one pack, readable during and after.
28692×
            let big = s.begin_batch().unwrap();
28702×
            let mut refs = Vec::new();
28712×
            let mut seed = 7u64;
2872600×
            for i in 0..300u32 {
2873600×
                let mut body = vec![b'z'; 40_000];
2874600×
                if i % 3 == 0 {
28758000000×
                    for b in body.iter_mut() {
28768000000×
                        *b = (xorshift(&mut seed) & 0xff) as u8;
28778000000×
                    }
2878400×
                }
2879600×
                body[..4].copy_from_slice(&i.to_le_bytes());
2880600×
                let r = if i % 5 == 0 {
2881120×
                    s.put_chunk_nocompress(&body).unwrap()
2882
                } else {
2883480×
                    s.put(ObjectType::Chunk, &body).unwrap()
2884
                };
2885600×
                assert_eq!(
2886600×
                    s.get(&r).unwrap().1,
2887
                    body,
2888
                    "readable while pending (object {i})"
2889
                );
2890600×
                refs.push((r, body));
2891
            }
28922×
            assert!(s.has(&refs[0].0));
28932×
            big.finish().unwrap();
28942×
            assert_eq!(s.pack_count(), 1);
28952×
            assert_eq!(
28962×
                s.loose_count().unwrap(),
2897
                2,
2898
                "nothing from the big batch is loose"
2899
            );
29002×
            assert!(fs::read_dir(s.dir())
29012×
                .unwrap()
29022×
                .flatten()
29034×
                .all(|e| !e.file_name().to_string_lossy().contains("tmp-")));
29042×
            let s2 = Store::open_on(s.backend().clone(), s.dir()).unwrap();
2905600×
            for (r, body) in &refs {
2906600×
                assert_eq!(s2.get(r).unwrap().1, *body);
2907
            }
2908
            // A dropped batch still writes its objects.
29092×
            {
29102×
                let _b = s2.begin_batch().unwrap();
29112×
                s2.put(ObjectType::Chunk, b"dropped").unwrap();
29122×
            }
29132×
            assert_eq!(
29142×
                s2.get(&hash_object(ObjectType::Chunk, b"dropped"))
29152×
                    .unwrap()
2916
                    .1,
2917
                b"dropped"
2918
            );
29192×
        });
29201×
    }
2921
2922
    #[test]
29231×
    fn get_many_reads_loose_and_packed_objects_in_order() {
29241×
        let dir = tempfile::tempdir().unwrap();
29252×
        each_store(dir.path(), |which, s| {
29262×
            let packed: Vec<ObjectRef> = (0..20)
292740×
                .map(|i| {
292840×
                    s.put(ObjectType::Chunk, format!("packed {i}").as_bytes())
292940×
                        .unwrap()
293040×
                })
29312×
                .collect();
29322×
            s.pack_loose().unwrap();
29332×
            let loose: Vec<ObjectRef> = (0..20)
293440×
                .map(|i| {
293540×
                    s.put(ObjectType::Chunk, format!("loose {i}").as_bytes())
293640×
                        .unwrap()
293740×
                })
29382×
                .collect();
29392×
            let missing = hash_object(ObjectType::Chunk, b"never stored");
29402×
            let mut refs: Vec<ObjectRef> = Vec::new();
294140×
            for i in 0..20 {
294240×
                refs.push(loose[i]);
294340×
                refs.push(packed[i]);
294440×
            }
29452×
            refs.push(missing);
29462×
            let got = s.get_many(&refs);
29472×
            assert_eq!(got.len(), refs.len());
294880×
            for (i, r) in refs.iter().enumerate().take(40) {
294980×
                let want = if i % 2 == 0 {
295040×
                    format!("loose {}", i / 2)
2951
                } else {
295240×
                    format!("packed {}", i / 2)
2953
                };
295480×
                let (ty, body) = got[i].as_ref().unwrap();
295580×
                assert_eq!(
295680×
                    (*ty, body.as_slice()),
295780×
                    (ObjectType::Chunk, want.as_bytes()),
2958
                    "{which}: {r:?}"
2959
                );
2960
            }
29612×
            assert!(matches!(got[40], Err(StoreError::NotFound(_))));
29622×
        });
29631×
    }
2964
2965
    /// A packed object is read from its pack without first asking the
2966
    /// backend for a loose copy (oxygenhub#228: on a bucket each such ask
2967
    /// was a request that found nothing, every time).
2968
    #[test]
29691×
    fn packed_objects_cost_no_loose_lookups() {
2970
        use std::sync::atomic::{AtomicUsize, Ordering};
2971
        struct Counting(MemoryStore, AtomicUsize);
2972
        impl crate::strata::Store for Counting {
29730×
            fn describe(&self) -> String {
29740×
                "counting".into()
29750×
            }
297610×
            fn has(&self, k: &str) -> crate::strata::Result<bool> {
297710×
                self.0.has(k)
297810×
            }
297915×
            fn get(&self, k: &str) -> crate::strata::Result<Vec<u8>> {
298015×
                if k.starts_with("loose/") {
298111×
                    self.1.fetch_add(1, Ordering::Relaxed);
298211×
                }
298315×
                self.0.get(k)
298415×
            }
298520×
            fn get_range(
298620×
                &self,
298720×
                k: &str,
298820×
                r: std::ops::Range<u64>,
298920×
            ) -> crate::strata::Result<Vec<u8>> {
299020×
                self.0.get_range(k, r)
299120×
            }
299212×
            fn put_new(&self, k: &str, b: &[u8]) -> crate::strata::Result<()> {
299312×
                self.0.put_new(k, b)
299412×
            }
29958×
            fn list(&self, p: &str) -> crate::strata::Result<Vec<crate::strata::Entry>> {
29968×
                self.0.list(p)
29978×
            }
29980×
            fn delete(&self, k: &str) -> crate::strata::Result<()> {
29990×
                self.0.delete(k)
30000×
            }
30013×
            fn read_ref(
30023×
                &self,
30033×
                k: &str,
30043×
            ) -> crate::strata::Result<Option<(Vec<u8>, crate::strata::Version)>> {
30053×
                self.0.read_ref(k)
30063×
            }
30071×
            fn write_ref(
30081×
                &self,
30091×
                k: &str,
30101×
                b: &[u8],
30111×
                e: Option<&crate::strata::Version>,
30121×
            ) -> crate::strata::Result<crate::strata::Version> {
30131×
                self.0.write_ref(k, b, e)
30141×
            }
3015
        }
30161×
        let dir = tempfile::tempdir().unwrap();
30171×
        let backend = Arc::new(Counting(MemoryStore::new(), AtomicUsize::new(0)));
30181×
        let s = Store::open_on(backend.clone(), dir.path().join("scratch")).unwrap();
30191×
        let refs: Vec<ObjectRef> = (0..10)
302010×
            .map(|i| {
302110×
                s.put(ObjectType::Chunk, format!("p {i}").as_bytes())
302210×
                    .unwrap()
302310×
            })
30241×
            .collect();
30251×
        s.pack_loose().unwrap();
30261×
        let s = Store::open_on(backend.clone(), dir.path().join("scratch")).unwrap();
30271×
        backend.1.store(0, Ordering::Relaxed);
302810×
        for r in &refs {
302910×
            assert!(s.get(r).is_ok());
3030
        }
303110×
        assert!(s.get_many(&refs).iter().all(|g| g.is_ok()));
30321×
        assert_eq!(backend.1.load(Ordering::Relaxed), 0, "no loose lookups");
3033
        // An object no pack holds is still looked for loose.
30341×
        let missing = hash_object(ObjectType::Chunk, b"never stored");
30351×
        assert!(matches!(s.get(&missing), Err(StoreError::NotFound(_))));
30361×
        assert!(backend.1.load(Ordering::Relaxed) > 0);
30371×
    }
3038
}
3039
3040
/// Readers never wait (oxygenhub#249): a store keeps reading while other
3041
/// handles repack, collect and push into the store under it.
3042
#[cfg(test)]
3043
pub(crate) mod reader_tests {
3044
    use super::*;
3045
30461682×
    fn chunk(i: usize) -> Vec<u8> {
30471682×
        format!("object {i} ").repeat(1 + i % 5).into_bytes()
30481682×
    }
3049
3050
    /// Two handles on one store: a writer, and a reader opened at once.
30514×
    fn each_pair(f: impl Fn(&str, Store, Store)) {
30524×
        let dir = tempfile::tempdir().unwrap();
30534×
        let s = Store::open(dir.path().join("objects")).unwrap();
30544×
        let r = Store::open(dir.path().join("objects")).unwrap();
30554×
        f("directory", s, r);
30564×
        let m: Arc<Backend> = Arc::new(MemoryStore::new());
30574×
        let s = Store::open_on(m.clone(), dir.path().join("scratch")).unwrap();
30584×
        let r = Store::open_on(m, dir.path().join("scratch-r")).unwrap();
30594×
        f("memory", s, r);
30604×
    }
3061
3062
    /// Set every retired entry's time back by `by`, as if retired then.
30633×
    pub(crate) fn age_retired(s: &Store, by: std::time::Duration) {
30643×
        s.update_retired(|m| {
306544×
            for t in m.values_mut() {
306644×
                *t = t.saturating_sub(by.as_millis() as u64);
306744×
            }
30683×
        })
30693×
        .unwrap();
30703×
    }
3071
307212×
    fn retired_keys(s: &Store) -> Vec<String> {
307312×
        s.read_retired().unwrap().0.into_keys().collect()
307412×
    }
3075
3076
    #[test]
30771×
    fn a_reader_opened_before_a_repack_reads_every_object_after_it() {
30782×
        each_pair(|which, writer, _| {
30792×
            let mut refs = Vec::new();
308080×
            for i in 0..40 {
308180×
                refs.push(writer.put(ObjectType::Chunk, &chunk(i)).unwrap());
308280×
            }
30832×
            writer.pack_loose().unwrap();
308480×
            for i in 40..80 {
308580×
                refs.push(writer.put(ObjectType::Chunk, &chunk(i)).unwrap());
308680×
            }
3087
            // Opened now: one pack and forty loose objects.
30882×
            let reader = Store::open_on(writer.backend().clone(), writer.dir()).unwrap();
30892×
            assert_eq!(reader.pack_count(), 1, "{which}");
30902×
            let (_, merged) = writer.repack_all(REPACK_ZSTD_LEVEL).unwrap();
30912×
            assert_eq!(merged, 1);
3092
            // The old pack and the loose objects are retired, not deleted.
30932×
            let retired = retired_keys(&writer);
30942×
            assert_eq!(retired.len(), 81, "{which}: {retired:?}");
30952×
            assert_eq!(writer.loose_count().unwrap(), 0, "retired is not loose");
3096
            // The reader reads on from what it loaded.
3097160×
            for (i, r) in refs.iter().enumerate() {
3098160×
                assert_eq!(reader.get(r).unwrap().1, chunk(i), "{which}: {i}");
3099
            }
31002×
            assert_eq!(reader.pack_count(), 1);
3101
            // A store opened now reads the new pack only.
31022×
            let fresh = Store::open_on(writer.backend().clone(), writer.dir()).unwrap();
31032×
            assert_eq!(fresh.pack_count(), 1);
31042×
            assert_ne!(fresh.packs.borrow()[0].key, reader.packs.borrow()[0].key);
3105
            // Once the retired files are deleted, the reader finds every
3106
            // object in the new pack, by get, get_many and has alike.
31072×
            let report = writer
31082×
                .expire_retired(std::time::Duration::ZERO, &mut silent())
31092×
                .unwrap();
31102×
            assert_eq!((report.packs, report.loose), (1, 80), "{which}");
31112×
            assert!(retired_keys(&writer).is_empty());
3112160×
            for (i, r) in refs.iter().enumerate() {
3113160×
                assert_eq!(reader.get(r).unwrap().1, chunk(i), "{which}: {i}");
3114
            }
31152×
            let other = Store::open_on(writer.backend().clone(), writer.dir()).unwrap();
31162×
            let before = other.packs.borrow()[0].key.clone();
31172×
            writer.put(ObjectType::Chunk, b"new").unwrap();
31182×
            writer.repack_all(REPACK_ZSTD_LEVEL).unwrap();
31192×
            writer
31202×
                .expire_retired(std::time::Duration::ZERO, &mut silent())
31212×
                .unwrap();
3122160×
            assert!(other.get_many(&refs).iter().all(|g| g.is_ok()), "{which}");
31232×
            assert_ne!(other.packs.borrow()[0].key, before);
31242×
            let third = Store::open_on(writer.backend().clone(), writer.dir()).unwrap();
31252×
            writer.put(ObjectType::Chunk, b"newer").unwrap();
31262×
            writer.repack_all(REPACK_ZSTD_LEVEL).unwrap();
31272×
            writer
31282×
                .expire_retired(std::time::Duration::ZERO, &mut silent())
31292×
                .unwrap();
3130160×
            assert!(refs.iter().all(|r| third.has(r)), "{which}");
31312×
            assert!(!third.has(&hash_object(ObjectType::Chunk, b"never")));
31322×
        });
31331×
    }
3134
3135
    #[test]
31361×
    fn a_reader_opened_before_a_spilled_batch_reads_its_objects() {
31372×
        each_pair(|which, writer, reader| {
31382×
            assert_eq!(reader.pack_count(), 0);
31392×
            let mut refs = Vec::new();
3140
            {
31412×
                let batch = writer.begin_batch().unwrap();
3142612×
                for i in 0..(SPILL_OBJECTS + 50) {
3143612×
                    refs.push(writer.put(ObjectType::Chunk, &chunk(i)).unwrap());
3144612×
                }
31452×
                batch.finish().unwrap();
3146
            }
31472×
            assert_eq!(writer.pack_count(), 1, "{which}: the batch spilled");
31482×
            assert_eq!(reader.get(&refs[0]).unwrap().1, chunk(0), "{which}");
31492×
            assert_eq!(reader.pack_count(), 1, "the miss loaded the new pack");
31502×
            let got = reader.get_many(&refs);
3151612×
            assert!(got.iter().all(|g| g.is_ok()), "{which}");
3152
            // has, too, from another reader.
31532×
            let late = Store::open_on(writer.backend().clone(), writer.dir()).unwrap();
3154
            {
31552×
                let batch = writer.begin_batch().unwrap();
3156514×
                for i in 1000..(1000 + SPILL_OBJECTS + 1) {
3157514×
                    writer.put(ObjectType::Chunk, &chunk(i)).unwrap();
3158514×
                }
31592×
                batch.finish().unwrap();
3160
            }
31612×
            assert!(late.has(&hash_object(ObjectType::Chunk, &chunk(1000))));
31622×
        });
31631×
    }
3164
3165
    #[test]
31661×
    fn retired_entries_expire_after_the_hour_and_are_never_loaded() {
31672×
        each_pair(|which, s, _| {
316820×
            for i in 0..10 {
316920×
                s.put(ObjectType::Chunk, &chunk(i)).unwrap();
317020×
            }
31712×
            s.pack_loose().unwrap();
31722×
            s.put(ObjectType::Chunk, &chunk(10)).unwrap();
31732×
            s.repack_all(ZSTD_LEVEL).unwrap();
3174
            // Retired over an hour ago: the first pack and 10 loose objects.
31752×
            age_retired(&s, RETIRED_FOR + std::time::Duration::from_secs(60));
31762×
            let old = retired_keys(&s);
31772×
            assert_eq!(old.len(), 12, "{which}: {old:?}");
3178
            // Retired just now: the second pack.
31792×
            s.put(ObjectType::Chunk, &chunk(11)).unwrap();
31802×
            s.repack_all(ZSTD_LEVEL).unwrap();
31812×
            assert_eq!(retired_keys(&s).len(), 14);
31822×
            let listed = s
31832×
                .backend()
31842×
                .list("pack-")
31852×
                .unwrap()
31862×
                .into_iter()
318714×
                .filter(|e| e.key.ends_with(".idx"))
31882×
                .count();
31892×
            assert_eq!(listed, 3, "{which}");
31902×
            let fresh = Store::open_on(s.backend().clone(), s.dir()).unwrap();
31912×
            assert_eq!(
31922×
                fresh.pack_count(),
3193
                1,
3194
                "{which}: retired packs are not loaded"
3195
            );
31962×
            let report = s.expire_retired(RETIRED_FOR, &mut silent()).unwrap();
31972×
            assert_eq!(
3198
                report,
3199
                RetiredReport {
3200
                    packs: 1,
3201
                    loose: 11,
3202
                    kept: 2
3203
                },
3204
                "{which}"
3205
            );
32062×
            let young = retired_keys(&s);
32072×
            assert_eq!(young.len(), 2, "{which}: {young:?}");
32084×
            assert!(young.iter().all(|k| !old.contains(k)));
320924×
            for k in old.iter().filter(|k| k.ends_with(".pack")) {
32102×
                assert!(!s.backend().has(k).unwrap(), "{which}: {k}");
3211
            }
32124×
            for k in &young {
32134×
                assert!(s.backend().has(k).unwrap(), "{which}: {k} kept");
3214
            }
321524×
            for i in 0..12 {
321624×
                let r = hash_object(ObjectType::Chunk, &chunk(i));
321724×
                assert_eq!(fresh.get(&r).unwrap().1, chunk(i), "{which}");
3218
            }
32192×
        });
32201×
    }
3221
3222
    /// A retired loose object whose pack copy was pruned since is not
3223
    /// lost when the retired files go: it is loose again.
3224
    #[test]
32251×
    fn a_retired_loose_object_no_pack_holds_is_kept() {
32262×
        each_pair(|which, s, _| {
32272×
            let a = s.put(ObjectType::Chunk, b"a").unwrap();
32282×
            let b = s.put(ObjectType::Chunk, b"b").unwrap();
32292×
            s.pack_loose().unwrap();
32302×
            assert_eq!(s.loose_count().unwrap(), 0);
3231
            // The pack rewritten without b, as pruning does.
32322×
            let first = s.packs.borrow()[0].key.clone();
32332×
            s.write_pack(&[a], &[], &[first], ZSTD_LEVEL, &mut silent())
32342×
                .unwrap();
32352×
            let report = s
32362×
                .expire_retired(std::time::Duration::ZERO, &mut silent())
32372×
                .unwrap();
32382×
            assert_eq!((report.packs, report.loose), (1, 1), "{which}: a's copy");
32392×
            assert!(retired_keys(&s).is_empty());
32402×
            assert_eq!(s.get(&b).unwrap().1, b"b", "{which}");
32412×
            assert_eq!(s.get(&a).unwrap().1, b"a");
32422×
            assert_eq!(s.loose_count().unwrap(), 1, "b is loose again");
3243
            // And pruning it, unreachable, deletes it.
32442×
            let keep = [a].into_iter().collect();
32452×
            let r = s.prune(&keep, std::time::Duration::ZERO).unwrap();
32462×
            assert_eq!(r.objects, 1);
32472×
            assert!(!s.has(&b));
32482×
        });
32491×
    }
3250
3251
    /// A backend whose first pack listing names an index that is deleted
3252
    /// before it is read, as a collection running meanwhile would.
3253
    pub(crate) struct Racing {
3254
        inner: MemoryStore,
3255
        lists: std::sync::atomic::AtomicUsize,
3256
        /// Deleted right after the first listing returns.
3257
        pub(crate) doomed: std::sync::Mutex<Option<String>>,
3258
        /// Named in every listing, never stored.
3259
        ghost: bool,
3260
    }
3261
3262
    impl crate::strata::Store for Racing {
32630×
        fn describe(&self) -> String {
32640×
            "racing".into()
32650×
        }
326611×
        fn has(&self, k: &str) -> crate::strata::Result<bool> {
326711×
            self.inner.has(k)
326811×
        }
326969×
        fn get(&self, k: &str) -> crate::strata::Result<Vec<u8>> {
327069×
            self.inner.get(k)
327169×
        }
32726×
        fn get_range(&self, k: &str, r: std::ops::Range<u64>) -> crate::strata::Result<Vec<u8>> {
32736×
            self.inner.get_range(k, r)
32746×
        }
327520×
        fn put_new(&self, k: &str, b: &[u8]) -> crate::strata::Result<()> {
327620×
            self.inner.put_new(k, b)
327720×
        }
327854×
        fn list(&self, p: &str) -> crate::strata::Result<Vec<crate::strata::Entry>> {
327954×
            let mut out = self.inner.list(p)?;
328054×
            if p == "pack-" {
328134×
                self.lists
328234×
                    .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
328334×
                if let Some(k) = self.doomed.lock().unwrap().take() {
32842×
                    self.inner.delete(&k)?;
328532×
                }
328634×
                if self.ghost {
328750×
                    for k in ["pack-0000.idx", "pack-0000.pack"] {
328850×
                        out.push(crate::strata::Entry {
328950×
                            key: k.into(),
329050×
                            size: 100,
329150×
                            modified_ms: None,
329250×
                        });
329350×
                    }
3294142×
                    out.sort_by(|a, b| a.key.cmp(&b.key));
32959×
                }
329620×
            }
329754×
            Ok(out)
329854×
        }
32990×
        fn delete(&self, k: &str) -> crate::strata::Result<()> {
33000×
            self.inner.delete(k)
33010×
        }
330224×
        fn read_ref(
330324×
            &self,
330424×
            k: &str,
330524×
        ) -> crate::strata::Result<Option<(Vec<u8>, crate::strata::Version)>> {
330624×
            self.inner.read_ref(k)
330724×
        }
33086×
        fn write_ref(
33096×
            &self,
33106×
            k: &str,
33116×
            b: &[u8],
33126×
            e: Option<&crate::strata::Version>,
33136×
        ) -> crate::strata::Result<crate::strata::Version> {
33146×
            self.inner.write_ref(k, b, e)
33156×
        }
3316
    }
3317
33183×
    pub(crate) fn racing(ghost: bool) -> Arc<Racing> {
33193×
        Arc::new(Racing {
33203×
            inner: MemoryStore::new(),
33213×
            lists: Default::default(),
33223×
            doomed: Default::default(),
33233×
            ghost,
33243×
        })
33253×
    }
3326
3327
    #[test]
33281×
    fn an_index_deleted_between_listing_and_loading_is_passed_over() {
33291×
        let dir = tempfile::tempdir().unwrap();
33301×
        let backend = racing(false);
33311×
        let s = Store::open_on(backend.clone(), dir.path()).unwrap();
33321×
        let a = s.put(ObjectType::Chunk, b"a").unwrap();
33331×
        s.pack_loose().unwrap();
33341×
        let b = s.put(ObjectType::Chunk, b"b").unwrap();
33351×
        s.pack_loose().unwrap();
33361×
        let idx = {
33371×
            let packs = s.packs.borrow();
33381×
            let (i, _) = s.find_packed(&a).unwrap();
33391×
            format!("{}.idx", packs[i].key.trim_end_matches(".pack"))
3340
        };
33411×
        *backend.doomed.lock().unwrap() = Some(idx);
33421×
        let before = backend.lists.load(std::sync::atomic::Ordering::Relaxed);
33431×
        let again = Store::open_on(backend.clone(), dir.path()).unwrap();
33441×
        assert_eq!(again.pack_count(), 1, "the deleted one is passed over");
33451×
        assert_eq!(
33461×
            backend.lists.load(std::sync::atomic::Ordering::Relaxed) - before,
3347
            2,
3348
            "and the store listed again"
3349
        );
33501×
        assert_eq!(again.get(&b).unwrap().1, b"b");
3351
        // A listing that names an index never there opens too.
33521×
        let ghosted = racing(true);
33531×
        let s = Store::open_on(ghosted.clone(), dir.path().join("g")).unwrap();
33541×
        let c = s.put(ObjectType::Chunk, b"c").unwrap();
33551×
        s.pack_loose().unwrap();
33561×
        let again = Store::open_on(ghosted, dir.path().join("g")).unwrap();
33571×
        assert_eq!(again.pack_count(), 1);
33581×
        assert_eq!(again.get(&c).unwrap().1, b"c");
33591×
    }
3360
}
3361
3362
#[cfg(test)]
3363
mod batch_tests {
3364
    use super::*;
3365
3366
    #[test]
33671×
    fn a_large_batch_packs_every_object_and_indexes_them_all() {
33681×
        let dir = tempfile::tempdir().unwrap();
33691×
        let store = Store::open(dir.path()).unwrap();
33701×
        let mut refs = Vec::new();
3371
        {
33721×
            let batch = store.begin_batch().unwrap();
33731000×
            for i in 0..1000u32 {
33741000×
                let body = format!("object {i} {}", "x".repeat(20_000 + i as usize)).into_bytes();
33751000×
                refs.push(store.put(ObjectType::Blob, &body).unwrap());
33761000×
            }
33771×
            batch.finish().unwrap();
3378
        }
33791×
        let again = Store::open(dir.path()).unwrap();
33801000×
        for r in &refs {
33811000×
            again
33821000×
                .get(r)
33831000×
                .unwrap_or_else(|e| panic!("{}: {e}", r.short(12)));
3384
        }
33851×
    }
3386
3387
    #[test]
33881×
    fn a_pack_a_stopped_batch_left_is_adopted_and_its_torn_end_cut() {
3389
        // oxygen#79: a batch never finished (a process killed part way)
3390
        // leaves its objects in an unindexed .tmp pack.
33911×
        let dir = tempfile::tempdir().unwrap();
33921×
        let mut refs = Vec::new();
3393
        {
33941×
            let store = Store::open(dir.path()).unwrap();
33951×
            let batch = store.begin_batch().unwrap();
3396300×
            for i in 0..300u32 {
3397300×
                let body = format!("object {i} {}", "y".repeat(20_000 + i as usize)).into_bytes();
3398300×
                refs.push(store.put(ObjectType::Blob, &body).unwrap());
3399300×
            }
34001×
            std::mem::forget(batch);
3401
        }
34021×
        let tmp: Vec<_> = fs::read_dir(dir.path())
34031×
            .unwrap()
34041×
            .flatten()
34052×
            .filter(|e| e.file_name().to_string_lossy().starts_with(".tmp-"))
34061×
            .map(|e| e.path())
34071×
            .collect();
34081×
        assert_eq!(tmp.len(), 1, "one unfinished pack");
3409
        // A torn last entry: a header and part of its body.
34101×
        let mut f = fs::OpenOptions::new().append(true).open(&tmp[0]).unwrap();
34111×
        f.write_all(&[0, 1, 0xff, 0, 0, 0, 9, 9, 0, 0]).unwrap();
34121×
        f.write_all(&[7; 40]).unwrap();
34131×
        drop(f);
34141×
        let store = Store::open(dir.path()).unwrap();
34151×
        assert!(store.get(&refs[0]).is_err(), "unreadable before");
34161×
        let young = store
34171×
            .adopt_unfinished_packs(std::time::Duration::from_secs(3600), &|_| {})
34181×
            .unwrap();
34191×
        assert_eq!(young, (0, 0), "a pack written to lately is left alone");
34201×
        let (packs, objects) = store
34212×
            .adopt_unfinished_packs(std::time::Duration::ZERO, &|_| {})
34221×
            .unwrap();
34231×
        assert_eq!((packs, objects), (1, 300));
3424300×
        for r in &refs {
3425300×
            store
3426300×
                .get(r)
3427300×
                .unwrap_or_else(|e| panic!("{}: {e}", r.short(12)));
3428
        }
34291×
        let again = Store::open(dir.path()).unwrap();
34301×
        assert!(again.get(&refs[299]).is_ok(), "indexed for good");
34311×
    }
3432
}
3433
3434
#[cfg(test)]
3435
mod size_tests {
3436
    use super::*;
3437
3438
    #[test]
34391×
    fn an_object_over_the_limit_is_refused_and_nothing_is_written() {
34401×
        let dir = tempfile::tempdir().unwrap();
34411×
        let store = Store::open(dir.path()).unwrap();
34421×
        let big = vec![b'x'; MAX_OBJECT_LEN as usize + 1];
3443
        {
34441×
            let batch = store.begin_batch().unwrap();
34451×
            store.put(ObjectType::Blob, b"small").unwrap();
34461×
            let e = store.put(ObjectType::Change, &big).unwrap_err();
34471×
            assert!(
34481×
                matches!(e, StoreError::TooLarge { kind: "change", .. }),
3449
                "{e}"
3450
            );
34511×
            assert!(e.to_string().contains("commit it in parts"), "{e}");
34521×
            batch.finish().unwrap();
3453
        }
3454
        // The store still opens and reads what was written.
34551×
        let again = Store::open(dir.path()).unwrap();
34561×
        again.get(&hash_object(ObjectType::Blob, b"small")).unwrap();
34571×
    }
3458
}
3459
3460
#[cfg(test)]
3461
mod text_tests {
3462
    use super::*;
3463
    use crate::objects::{AtomRef, Edit, CHANGE_VERSION};
3464
34652×
    fn large_change() -> Change {
34662×
        let mut c = crate::objects::tests::sample_change();
34672×
        assert_eq!(c.version, CHANGE_VERSION);
34682×
        c.edits[0] = Edit::Insert {
34692×
            node: crate::ids::NodeId([7; 16]),
34702×
            after: AtomRef::Start,
34712×
            before: AtomRef::End,
34722×
            lines: (0..2000)
34734000×
                .map(|i| format!("line {i}\n").into_bytes())
34742×
                .collect(),
3475
        };
34762×
        c
34772×
    }
3478
3479
    #[test]
34801×
    fn a_large_insert_is_stored_beside_its_change_and_read_back() {
34811×
        let dir = tempfile::tempdir().unwrap();
34821×
        let s = Store::open(dir.path()).unwrap();
34831×
        let c = large_change();
34841×
        let r = s.put_change(&c).unwrap();
34851×
        assert_eq!(r, c.hash());
34861×
        let (_, body) = s.get(&r).unwrap();
34871×
        assert!(
34881×
            body.len() < 1024,
3489
            "the change holds a reference, not {} bytes of lines",
34900×
            body.len()
3491
        );
34921×
        let (back, parts) = s.get_change_parts(&r).unwrap();
34931×
        assert_eq!(back, c);
3494
        // The text blob and its chunk.
34951×
        assert_eq!(parts.len(), 2);
34962×
        for p in &parts {
34972×
            assert!(s.has(p));
3498
        }
3499
        // Storing it again writes nothing new.
35001×
        let n = s.list().unwrap().len();
35011×
        s.put_change(&c).unwrap();
35021×
        assert_eq!(s.list().unwrap().len(), n);
35031×
    }
3504
3505
    #[test]
35061×
    fn a_change_whose_text_blob_is_missing_is_reported_not_misread() {
35071×
        if !super::files_on_disk() {
35080×
            return;
35091×
        }
35101×
        let dir = tempfile::tempdir().unwrap();
35111×
        let s = Store::open(dir.path()).unwrap();
35121×
        let c = large_change();
35131×
        let r = s.put_change(&c).unwrap();
35141×
        let (_, parts) = s.get_change_parts(&r).unwrap();
35151×
        fs::remove_file(s.path_for(&parts[0])).unwrap();
35161×
        let e = s.get_change(&r).unwrap_err();
35171×
        assert!(
35181×
            matches!(&e, StoreError::Corrupt(h, why) if *h == r && why.contains("missing")),
3519
            "{e}"
3520
        );
35211×
    }
3522
}
3523
3524
#[cfg(test)]
3525
mod put_many_tests {
3526
    use super::*;
3527
3528
    #[test]
35291×
    fn many_objects_in_a_batch_are_compressed_together_and_all_read_back() {
35301×
        let dir = tempfile::tempdir().unwrap();
35311×
        let s = Store::open(dir.path()).unwrap();
35321×
        let before = s.put(ObjectType::Chunk, b"already here").unwrap();
35331×
        let bodies: Vec<Vec<u8>> = (0..200u32)
3534200×
            .map(|i| format!("chunk {i} ").repeat(6000).into_bytes())
35351×
            .collect();
35361×
        let mut objects: Vec<(ObjectType, &[u8])> = bodies
35371×
            .iter()
3538200×
            .map(|b| (ObjectType::Chunk, b.as_slice()))
35391×
            .collect();
35401×
        objects.push((ObjectType::Chunk, bodies[0].as_slice()));
35411×
        objects.push((ObjectType::Chunk, b"already here"));
35421×
        let refs = {
35431×
            let batch = s.begin_batch().unwrap();
35441×
            let refs = s.put_many(&objects).unwrap();
35451×
            batch.finish().unwrap();
35461×
            refs
3547
        };
35481×
        assert_eq!(refs.len(), objects.len());
35491×
        assert_eq!(refs[200], refs[0], "a repeat is the same object");
35501×
        assert_eq!(refs[201], before);
35511×
        let again = Store::open(dir.path()).unwrap();
3552202×
        for (r, b) in refs.iter().zip(&objects) {
3553202×
            assert_eq!(again.get(r).unwrap().1, b.1);
3554
        }
35551×
    }
3556
}