OXYGENHUB

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

3356 of 3520 lines run, 513 of 713 functions · run #375 on main · 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
//! A store that is not this disk's directory, where finding an object's
31
//! entries is a LIST request, also keeps every entry key in the
32
//! `pack-entries` journal and, once a collection folds them, in entry
33
//! lists (`pack-<id>.entries`), read once per open (spec §13.7, #116).
34
35
use crate::attest::Attest;
36
use crate::hash::{hash_object, header, ObjectRef, ObjectType};
37
use crate::identity::{Identity, Signature};
38
use crate::objects::{Blob, Change, ObjectError, StateObject};
39
use crate::receipt::Receipt;
40
use crate::strata::{self, Error as BackendError, LocalStore, MemoryStore};
41
use rand::RngCore;
42
use std::cell::RefCell;
43
use std::fs;
44
use std::io::{self, Write};
45
use std::path::{Path, PathBuf};
46
use std::sync::Arc;
47
48
const PACK_MAGIC: &[u8; 8] = b"OXYPACK1";
49
const IDX_MAGIC: &[u8; 8] = b"OXYIDX01";
50
const IDX_ENTRY: usize = 32 + 8 + 4 + 4 + 1 + 1;
51
const FLAG_ZSTD: u8 = 1;
52
/// Compressed with the pack's trained dictionary (`pack-ID.dict`), which a
53
/// full repack makes from a sample of the objects it packs: small files
54
/// share most of their vocabulary, which each one alone cannot use.
55
const FLAG_DICT: u8 = 2;
56
/// The size of a pack's dictionary, and the least a repack must pack for
57
/// one to pay.
58
const DICT_BYTES: usize = 110 << 10;
59
const DICT_MIN_OBJECTS: usize = 1000;
60
/// Objects up to this size are compressed with the dictionary.
61
const DICT_OBJECT_MAX: usize = 64 << 10;
62
const ZSTD_LEVEL: i32 = 3;
63
/// The level of a full repack (`oxy gc`): slower, once, for a smaller store.
64
pub const REPACK_ZSTD_LEVEL: i32 = 9;
65
/// Largest object a pack entry may claim; anything bigger is corruption,
66
/// and refusing it keeps a bad index from driving huge allocations.
67
const MAX_OBJECT_LEN: u32 = 1 << 30;
68
/// What reading an entry of a pack whose dictionary is gone says. A pack
69
/// being deleted can lose its dictionary first, so a read that meets this
70
/// looks at the packs again before it is believed.
71
const NO_DICT: &str = "packed with a dictionary the pack lacks";
72
73
/// The record of what repacks replaced (oxygenhub#249): packs and loose
74
/// objects whose contents a newer pack holds, kept for readers that
75
/// still use them until garbage collection deletes them, at least
76
/// [`RETIRED_FOR`] later. One line per key, `<retired at, wall ms> <key>`.
77
/// It is a reference record (updated by compare-and-swap, so concurrent
78
/// repacks and collections never lose each other's lines), named under
79
/// `pack-` so the pack listing every open makes shows whether it holds
80
/// anything.
81
const RETIRED_KEY: &str = "pack-retired";
82
/// How long a retired pack or loose object stays in the store: a reader
83
/// that loaded its list of packs before a repack reads on from the old
84
/// packs this long, and after that finds the objects in the new pack by
85
/// looking again (see [`Store::get`]).
86
pub const RETIRED_FOR: std::time::Duration = std::time::Duration::from_secs(60 * 60);
87
/// Compare-and-swap rounds for the retired record before giving up.
88
const RETIRED_TRIES: usize = 32;
89
/// After a look at the pack listing found nothing new, how long a missed
90
/// [`Store::has`] answers without looking again: a writer probing for
91
/// thousands of objects it is about to store would otherwise list the
92
/// packs for each.
93
const HAS_RELIST_AFTER: std::time::Duration = std::time::Duration::from_secs(1);
94
95
/// The journal of signatures and index entries (spec §13.7): every entry
96
/// key a writer on a store that is not this disk's directory wrote since
97
/// the last fold, one line each, `<folded at, wall ms; 0 until folded>
98
/// <key>`, as the retired record's lines are. A reference record updated
99
/// by compare-and-swap, so concurrent writers never lose each other's
100
/// lines; read once per open, instead of a listing per lookup.
101
pub const JOURNAL_KEY: &str = "pack-entries";
102
/// A complete entry list (spec §13.7): every entry a fold found.
103
const LIST_SUFFIX: &str = ".entries";
104
/// A partial entry list: journal lines a writer folded out.
105
const PART_SUFFIX: &str = ".entries-part";
106
/// The first line of an entry list.
107
const LIST_MAGIC: &str = "oxy-entries 1";
108
/// Written once a collection has copied the older one-list keys into
109
/// per-object entries: readers stop looking for them.
110
const LEGACY_FOLDED_KEY: &str = "pack-legacy-folded";
111
/// Unfolded journal lines a writer lets the journal hold before it folds
112
/// them into a partial list.
113
const JOURNAL_FOLD_AT: usize = 1000;
114
/// How long a handle answers lookups from the entries it read before it
115
/// reads the pack listing and the journal again (spec §13.7): the
116
/// interval a cloud client keeps the pack listing for, so another
117
/// writer's entry is seen as soon as its pack would be.
118
pub const ENTRIES_FRESH: std::time::Duration = std::time::Duration::from_secs(30);
119
/// The per-object entry indexes, each with the one-list name older
120
/// stores kept it under.
121
const ENTRY_INDEXES: [(&str, &str); 6] = [
122
    ("sig", "sigs"),
123
    ("seal", "seals"),
124
    ("receipt", "receipts"),
125
    ("approval", "approvals"),
126
    ("cert", "certs"),
127
    ("accepted", "accepted-lists"),
128
];
129
130
/// A pack listing as compared: each key with its size and time.
131
type Listed = Vec<(String, u64, Option<u64>)>;
132
133
/// The retired record's lines: each key with the time it was retired.
1341115×
fn parse_retired(bytes: &[u8]) -> std::collections::BTreeMap<String, u64> {
1351115×
    String::from_utf8_lossy(bytes)
1361115×
        .lines()
13772130×
        .filter_map(|l| {
13872130×
            let (t, k) = l.split_once(' ')?;
13972130×
            Some((k.to_string(), t.parse().ok()?))
14072130×
        })
1411115×
        .collect()
1421115×
}
143
144461×
fn encode_retired(m: &std::collections::BTreeMap<String, u64>) -> Vec<u8> {
145461×
    m.iter()
14636589×
        .map(|(k, t)| format!("{t} {k}\n"))
147461×
        .collect::<String>()
148461×
        .into_bytes()
149461×
}
150
151
/// What deleting retired packs and loose objects did.
152
#[derive(Debug, Clone, Default, PartialEq, Eq)]
153
pub struct RetiredReport {
154
    /// Retired packs deleted, each with its index and dictionary.
155
    pub packs: usize,
156
    /// Retired loose objects deleted.
157
    pub loose: usize,
158
    /// Retired entries younger than [`RETIRED_FOR`], left for the next
159
    /// collection.
160
    pub kept: usize,
161
}
162
163
/// An object read from the store: its type and body.
164
pub type Object = (ObjectType, Vec<u8>);
165
166
/// Where a long operation reports what it is doing (oxygen#66): a phase,
167
/// or a count within one, as a line for a spinner or a log.
168
pub type Progress<'a> = &'a mut dyn FnMut(&str);
169
170
/// The storage under a store: objects by key, see [`crate::strata`].
171
pub type Backend = dyn strata::Store;
172
173105×
fn silent() -> impl FnMut(&str) {
174201×
    |_| {}
175105×
}
176
177
/// A backend error as a store error: what is not found or already there
178
/// is handled where it matters; the rest is an I/O failure.
1794×
fn backend_err(e: BackendError) -> StoreError {
1804×
    match e {
1814×
        BackendError::Io(e) => StoreError::Io(e),
1820×
        other => StoreError::Io(io::Error::other(other.to_string())),
183
    }
1844×
}
185
186
/// `put_new` where an object already there is what we wanted.
18710649×
fn put_new_or_existing(backend: &Backend, key: &str, bytes: &[u8]) -> Result<()> {
18810649×
    match backend.put_new(key, bytes) {
18910649×
        Ok(()) | Err(BackendError::Exists(_)) => Ok(()),
1900×
        Err(e) => Err(backend_err(e)),
191
    }
19210649×
}
193
194
/// `get` as an option: `None` when there is no such key.
19566436×
fn get_optional(backend: &Backend, key: &str) -> Result<Option<Vec<u8>>> {
19666436×
    match backend.get(key) {
19757757×
        Ok(b) => Ok(Some(b)),
1988679×
        Err(BackendError::NotFound(_)) => Ok(None),
1990×
        Err(e) => Err(backend_err(e)),
200
    }
20166436×
}
202
2032859×
fn delete(backend: &Backend, key: &str) -> Result<()> {
2042859×
    backend.delete(key).map_err(backend_err)
2052859×
}
206
207
/// `loose/<hh>/<rest>`.
208100823×
fn loose_key(r: &ObjectRef) -> String {
209100823×
    let h = r.to_hex();
210100823×
    format!("loose/{}/{}", &h[..2], &h[2..])
211100823×
}
212
213
/// The object a loose key names, if it is one.
21425727×
fn loose_ref(key: &str) -> Option<ObjectRef> {
21525727×
    let rest = key.strip_prefix("loose/")?;
21625727×
    let (hh, tail) = rest.split_once('/')?;
21725727×
    if hh.len() != 2 || tail.contains('/') {
2180×
        return None;
21925727×
    }
22025727×
    ObjectRef::from_hex(&format!("{hh}{tail}"))
22125727×
}
222
223
/// `<index>/<hh>/<rest>/` for the entries kept under an object.
2249227×
fn fanout_prefix(index: &str, r: &ObjectRef) -> String {
2259227×
    let h = r.to_hex();
2269227×
    format!("{index}/{}/{}/", &h[..2], &h[2..])
2279227×
}
228
229
/// Whether a key names an entry list (spec §13.7), complete or partial.
2305073×
fn is_entry_list(key: &str) -> bool {
2315073×
    key.starts_with("pack-") && (key.ends_with(LIST_SUFFIX) || key.ends_with(PART_SUFFIX))
2325073×
}
233
234
/// An entry list's bytes: zstd-compressed text, [`LIST_MAGIC`] and then
235
/// one key per line, sorted.
23615×
fn encode_entry_list<'a>(keys: impl Iterator<Item = &'a String>) -> Result<Vec<u8>> {
23715×
    let mut text = String::from(LIST_MAGIC);
23815×
    text.push('\n');
2392230×
    for k in keys {
2402230×
        text.push_str(k);
2412230×
        text.push('\n');
2422230×
    }
24315×
    Ok(zstd::bulk::compress(text.as_bytes(), ZSTD_LEVEL)?)
24415×
}
245
246
/// The keys an entry list holds.
24720×
fn decode_entry_list(key: &str, bytes: &[u8]) -> Result<Vec<String>> {
24820×
    let bad = |why: &str| StoreError::Io(io::Error::other(format!("{key}: {why}")));
24920×
    let text = zstd::stream::decode_all(bytes).map_err(|_| bad("not zstd"))?;
25020×
    let text = String::from_utf8(text).map_err(|_| bad("not UTF-8"))?;
25120×
    let mut lines = text.lines();
25220×
    if lines.next() != Some(LIST_MAGIC) {
2530×
        return Err(bad("not an entry list"));
25420×
    }
25520×
    Ok(lines
2562243×
        .filter(|l| !l.is_empty())
25720×
        .map(str::to_string)
25820×
        .collect())
25920×
}
260
261
/// Drop journal lines folded at least [`RETIRED_FOR`] before `now`:
262
/// every reader that could need them has a list holding them by then.
263332×
fn expire_folded(m: &mut std::collections::BTreeMap<String, u64>, now: u64) {
264332×
    let hour = RETIRED_FOR.as_millis() as u64;
26530147×
    m.retain(|_, t| *t == 0 || now.saturating_sub(*t) < hour);
266332×
}
267
268
/// `<index>/<hh>/<rest>`: the one-file-per-object layout of older stores.
2696804×
fn legacy_key(index: &str, r: &ObjectRef) -> String {
2706804×
    let h = r.to_hex();
2716804×
    format!("{index}/{}/{}", &h[..2], &h[2..])
2726804×
}
273
274407×
fn now_ms() -> u64 {
275407×
    std::time::SystemTime::now()
276407×
        .duration_since(std::time::UNIX_EPOCH)
277407×
        .map(|d| d.as_millis() as u64)
278407×
        .unwrap_or(0)
279407×
}
280
281
#[derive(Clone, Copy, Debug)]
282
struct PackEntry {
283
    offset: u64,
284
    stored_len: u32,
285
    raw_len: u32,
286
    ty: ObjectType,
287
    flags: u8,
288
}
289
290
/// One pack's index, fully in memory and sorted by ObjectRef.
291
struct PackIndex {
292
    /// The pack's key, `pack-<id>.pack`.
293
    key: String,
294
    refs: Vec<ObjectRef>,
295
    entries: Vec<PackEntry>,
296
    /// The pack's dictionary, prepared once, when it has one.
297
    dict: Option<Arc<zstd::dict::DecoderDictionary<'static>>>,
298
}
299
300
impl std::fmt::Debug for PackIndex {
3010×
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
3020×
        f.debug_struct("PackIndex")
3030×
            .field("key", &self.key)
3040×
            .field("objects", &self.refs.len())
3050×
            .field("dict", &self.dict.is_some())
3060×
            .finish()
3070×
    }
308
}
309
310
/// Decode a stored pack entry and check it against its reference.
31117678×
fn decode_entry(
31217678×
    r: &ObjectRef,
31317678×
    e: PackEntry,
31417678×
    stored: Vec<u8>,
31517678×
    dict: Option<&Arc<zstd::dict::DecoderDictionary<'static>>>,
31617678×
) -> Result<Object> {
31717678×
    let body = if e.flags & FLAG_DICT != 0 {
3184502×
        let dict = dict.ok_or_else(|| StoreError::Corrupt(*r, NO_DICT.into()))?;
3194502×
        zstd::bulk::Decompressor::with_prepared_dictionary(dict)
3204502×
            .and_then(|mut d| d.decompress(&stored, e.raw_len as usize))
3214502×
            .map_err(|err| StoreError::Corrupt(*r, format!("zstd: {err}")))?
32213176×
    } else if e.flags & FLAG_ZSTD != 0 {
3237502×
        zstd::bulk::decompress(&stored, e.raw_len as usize)
3247502×
            .map_err(|err| StoreError::Corrupt(*r, format!("zstd: {err}")))?
325
    } else {
3265674×
        stored
327
    };
32817673×
    if body.len() != e.raw_len as usize || hash_object(e.ty, &body) != *r {
329119×
        return Err(StoreError::Corrupt(
330119×
            *r,
331119×
            "packed object does not match its reference".into(),
332119×
        ));
33317554×
    }
33417554×
    Ok((e.ty, body))
33517678×
}
336
337
impl PackIndex {
338
    /// Load a pack's index; `pack_len` is the pack's size, from the listing
339
    /// that found it (a stat of it would be a request of its own). `None`
340
    /// when the index is gone since the listing: the pack was deleted.
341499×
    fn load(
342499×
        backend: &Backend,
343499×
        idx_key: &str,
344499×
        pack_key: &str,
345499×
        pack_len: u64,
346499×
    ) -> Result<Option<PackIndex>> {
347499×
        let data = match backend.get(idx_key) {
348475×
            Ok(d) => d,
34924×
            Err(BackendError::NotFound(_)) => return Ok(None),
3500×
            Err(e) => return Err(backend_err(e)),
351
        };
352475×
        let bad = |m: &str| StoreError::Corrupt(ObjectRef::ZERO, format!("{idx_key}: {m}"));
353475×
        if data.len() < 12 || &data[..8] != IDX_MAGIC {
3542×
            return Err(bad("bad index header"));
355473×
        }
356473×
        let count = u32::from_le_bytes(data[8..12].try_into().unwrap()) as usize;
357473×
        if data.len() != 12 + count * IDX_ENTRY {
35842×
            return Err(bad("index length mismatch"));
359431×
        }
360431×
        let mut refs = Vec::with_capacity(count);
361431×
        let mut entries = Vec::with_capacity(count);
36224451×
        for i in 0..count {
36324451×
            let e = &data[12 + i * IDX_ENTRY..12 + (i + 1) * IDX_ENTRY];
36424451×
            let r = ObjectRef::from_slice(&e[..32]).unwrap();
36524451×
            if let Some(prev) = refs.last() {
36624020×
                if r <= *prev {
3673×
                    return Err(bad("index not sorted"));
36824017×
                }
369431×
            }
37024448×
            refs.push(r);
37124443×
            let entry = PackEntry {
37224448×
                offset: u64::from_le_bytes(e[32..40].try_into().unwrap()),
37324448×
                stored_len: u32::from_le_bytes(e[40..44].try_into().unwrap()),
37424448×
                raw_len: u32::from_le_bytes(e[44..48].try_into().unwrap()),
37524448×
                ty: ObjectType::ALL
37624448×
                    .get(e[48] as usize)
37724448×
                    .copied()
37824448×
                    .ok_or_else(|| bad("unknown object type"))?,
37924443×
                flags: e[49],
380
            };
38124443×
            if entry.stored_len > MAX_OBJECT_LEN || entry.raw_len > MAX_OBJECT_LEN {
3823×
                return Err(StoreError::Corrupt(
3833×
                    r,
3843×
                    format!(
3853×
                        "{idx_key}: the index claims an object of {} bytes ({} stored), more than the {MAX_OBJECT_LEN} an object may be",
3863×
                        entry.raw_len, entry.stored_len
3873×
                    ),
3883×
                ));
38924440×
            }
39024440×
            if entry.offset.saturating_add(entry.stored_len as u64) > pack_len {
39154×
                return Err(bad("index entry points outside the pack"));
39224386×
            }
39324386×
            entries.push(entry);
394
        }
395366×
        let dict_key = format!("{}.dict", pack_key.trim_end_matches(".pack"));
396366×
        let dict = get_optional(backend, &dict_key)?
397366×
            .map(|d| Arc::new(zstd::dict::DecoderDictionary::copy(&d)));
398366×
        Ok(Some(PackIndex {
399366×
            key: pack_key.to_string(),
400366×
            refs,
401366×
            entries,
402366×
            dict,
403366×
        }))
404499×
    }
405
40623359×
    fn find(&self, r: &ObjectRef) -> Option<PackEntry> {
40723359×
        self.refs.binary_search(r).ok().map(|i| self.entries[i])
40823359×
    }
409
410
    /// Read one entry from the pack: one ranged read, then decode.
41117358×
    fn fetch(&self, backend: &Backend, r: &ObjectRef, e: PackEntry) -> Result<Object> {
41217358×
        let stored = backend
41317358×
            .get_range(&self.key, e.offset..e.offset + e.stored_len as u64)
41417358×
            .map_err(|err| match err {
415162×
                BackendError::NotFound(_) => StoreError::NotFound(*r),
4160×
                other => backend_err(other),
417162×
            })?;
41817196×
        decode_entry(r, e, stored, self.dict.as_ref())
41917358×
    }
420
}
421
42215794×
fn type_index(ty: ObjectType) -> u8 {
42322432×
    ObjectType::ALL.iter().position(|t| *t == ty).unwrap() as u8
42415794×
}
425
426
/// What `pack_loose` did.
427
#[derive(Debug, Clone, Default, PartialEq, Eq)]
428
pub struct PackReport {
429
    pub objects: usize,
430
    pub raw_bytes: u64,
431
    pub packed_bytes: u64,
432
}
433
434
/// Packs tolerated before `merge_packs` folds them into one (spec §13.4).
435
pub const MAX_PACKS: usize = 16;
436
437
/// What `prune` did.
438
#[derive(Debug, Clone, Default, PartialEq, Eq)]
439
pub struct PruneReport {
440
    pub objects: usize,
441
    pub bytes: u64,
442
    /// Unreachable objects left alone because they are younger than the grace period.
443
    pub deferred: usize,
444
}
445
446
#[derive(Debug, thiserror::Error)]
447
pub enum StoreError {
448
    #[error("io: {0}")]
449
    Io(#[from] io::Error),
450
    #[error("corrupt object {0}: {1}")]
451
    Corrupt(ObjectRef, String),
452
    #[error("object {0}")]
453
    Object(#[from] ObjectError),
454
    #[error("object {0} is a {1}, expected {2}")]
455
    WrongType(ObjectRef, &'static str, &'static str),
456
    #[error("object {0} not found")]
457
    NotFound(ObjectRef),
458
    #[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)")]
459
    TooLarge {
460
        kind: &'static str,
461
        size: usize,
462
        limit: u32,
463
    },
464
}
465
466
pub type Result<T> = std::result::Result<T, StoreError>;
467
468
pub struct Store {
469
    backend: Arc<Backend>,
470
    /// Temporary files (a pack being streamed) and derived caches (`metas`).
471
    /// For a store in a directory it is that directory.
472
    scratch: PathBuf,
473
    /// The backend is that directory, so the old file paths still hold.
474
    local: bool,
475
    packs: RefCell<Vec<Arc<PackIndex>>>,
476
    /// The pack listing `packs` was loaded from (key, size, time), to tell
477
    /// whether a look at the store would find anything new.
478
    packs_listed: RefCell<Listed>,
479
    /// When a look at the pack listing for a missed [`Store::has`] last
480
    /// found nothing new.
481
    has_relisted: std::cell::Cell<Option<std::time::Instant>>,
482
    /// An open write batch (spec §13.4): objects go here instead of to loose
483
    /// files, and are written out with one `fsync` when the batch finishes.
484
    pending: RefCell<Option<Pending>>,
485
    batch_depth: std::cell::Cell<u32>,
486
    /// Change metadata already read, from the `metas` cache file.
487
    metas: RefCell<Option<crate::fasthash::FastMap<ObjectRef, crate::state::ChangeMeta>>>,
488
    /// The live entry lists (spec §13.7) the pack listing showed: listed
489
    /// and not retired.
490
    entry_lists: RefCell<Vec<String>>,
491
    /// Whether the store is folded: a live entry list is complete.
492
    folded: std::cell::Cell<bool>,
493
    /// Whether the pack listing showed `pack-legacy-folded`.
494
    legacy_folded: std::cell::Cell<bool>,
495
    /// The entry keys of a folded store, read from its lists and journal
496
    /// on the first lookup after the packs were loaded.
497
    entries: RefCell<Option<std::collections::BTreeSet<String>>>,
498
    /// When `entries` was read; past [`ENTRIES_FRESH`] it is read again.
499
    entries_read: std::cell::Cell<Option<std::time::Instant>>,
500
    /// Entry lists already read, by key: written once, so never read
501
    /// twice; those no longer live are let go.
502
    lists_read: RefCell<std::collections::HashMap<String, Arc<Vec<String>>>>,
503
    /// Entry keys written in an open batch, added to the journal when the
504
    /// batch is flushed.
505
    journal_queue: RefCell<Vec<String>>,
506
}
507
508
impl std::fmt::Debug for Store {
5090×
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
5100×
        f.debug_struct("Store")
5110×
            .field("backend", &self.backend.describe())
5120×
            .field("scratch", &self.scratch)
5130×
            .field("packs", &self.packs.borrow().len())
5140×
            .finish()
5150×
    }
516
}
517
518
/// Objects written since `begin_batch`. Small batches stay in memory and
519
/// become loose objects; once `SPILL_OBJECTS` or `SPILL_BYTES` is exceeded
520
/// the batch streams into a pack file instead.
521
#[derive(Debug)]
522
struct Pending {
523
    buffered: Vec<(ObjectType, Vec<u8>)>,
524
    buffered_bytes: usize,
525
    lookup: std::collections::HashMap<ObjectRef, PendingEntry>,
526
    pack: Option<PendingPack>,
527
    report: PackReport,
528
}
529
530
#[derive(Debug, Clone, Copy)]
531
enum PendingEntry {
532
    Buffered(usize),
533
    Packed(PackEntry),
534
}
535
536
#[derive(Debug)]
537
struct PendingPack {
538
    id: String,
539
    tmp_path: PathBuf,
540
    writer: io::BufWriter<fs::File>,
541
    offset: u64,
542
    index: Vec<(ObjectRef, PackEntry)>,
543
}
544
545
const SPILL_OBJECTS: usize = 256;
546
const SPILL_BYTES: usize = 8 << 20;
547
548
/// Finishes the enclosing batch when dropped; call `finish` to see errors.
549
pub struct Batch<'a> {
550
    store: &'a Store,
551
    done: bool,
552
}
553
554
impl Batch<'_> {
5553008×
    pub fn finish(mut self) -> Result<()> {
5563008×
        self.done = true;
5573008×
        self.store.finish_batch()
5583008×
    }
559
}
560
561
impl Drop for Batch<'_> {
5623010×
    fn drop(&mut self) {
5633010×
        if !self.done {
5642×
            let _ = self.store.finish_batch();
5653008×
        }
5663010×
    }
567
}
568
569
/// Append one stored object to a pending pack and index it.
5703226×
fn write_pack_entry(
5713226×
    pack: &mut PendingPack,
5723226×
    r: &ObjectRef,
5733226×
    ty: ObjectType,
5743226×
    raw_len: usize,
5753226×
    flags: u8,
5763226×
    stored: &[u8],
5773226×
    report: &mut PackReport,
5783226×
) -> Result<PackEntry> {
5793226×
    let header_len = 1 + 1 + 4 + 4 + 32;
5803226×
    pack.writer.write_all(&[type_index(ty), flags])?;
5813226×
    pack.writer
5823226×
        .write_all(&(stored.len() as u32).to_le_bytes())?;
5833226×
    pack.writer.write_all(&(raw_len as u32).to_le_bytes())?;
5843226×
    pack.writer.write_all(&r.0)?;
5853226×
    pack.writer.write_all(stored)?;
5863226×
    let entry = PackEntry {
5873226×
        offset: pack.offset + header_len,
5883226×
        stored_len: stored.len() as u32,
5893226×
        raw_len: raw_len as u32,
5903226×
        ty,
5913226×
        flags,
5923226×
    };
5933226×
    pack.offset += header_len + stored.len() as u64;
5943226×
    pack.index.push((*r, entry));
5953226×
    report.objects += 1;
5963226×
    report.raw_bytes += raw_len as u64;
5973226×
    report.packed_bytes += stored.len() as u64;
5983226×
    Ok(entry)
5993226×
}
600
601
/// A random name for a temporary file or a pack.
602124×
fn fresh_id() -> String {
603124×
    let mut nonce = [0u8; 8];
604124×
    rand::rngs::OsRng.fill_bytes(&mut nonce);
605124×
    hex::encode(nonce)
606124×
}
607
608
/// Under `OXY_STORE_BACKEND=memory` every store opens on a store in
609
/// memory shared by directory, so the whole test suite can run without a
610
/// directory store, as it would over a bucket. Objects live for the
611
/// process.
6123082×
fn memory_backend_for(dir: &Path) -> Option<Arc<Backend>> {
613
    use std::collections::HashMap;
614
    use std::sync::{Mutex, OnceLock};
6153082×
    if std::env::var_os("OXY_STORE_BACKEND").as_deref() != Some(std::ffi::OsStr::new("memory")) {
6163082×
        return None;
6170×
    }
618
    static STORES: OnceLock<Mutex<HashMap<PathBuf, Arc<MemoryStore>>>> = OnceLock::new();
6190×
    let key = dir.canonicalize().unwrap_or_else(|_| dir.to_path_buf());
6200×
    let mut map = STORES
6210×
        .get_or_init(Default::default)
6220×
        .lock()
6230×
        .unwrap_or_else(|p| p.into_inner());
6240×
    let store = map
6250×
        .entry(key)
6260×
        .or_insert_with(|| Arc::new(MemoryStore::new()))
6270×
        .clone();
6280×
    Some(store)
6293082×
}
630
631
#[cfg(test)]
632
thread_local! {
633
    /// A backend every store opened on this thread uses instead of its
634
    /// directory: tests of a repository kept on a store that is not this
635
    /// disk's directory, with several handles on one store.
636
    pub(crate) static TEST_BACKEND: RefCell<Option<Arc<Backend>>> = const { RefCell::new(None) };
637
}
638
639
impl Store {
640
    /// Open the store in `dir`, which is also its scratch directory.
6413091×
    pub fn open(dir: impl Into<PathBuf>) -> io::Result<Store> {
6423091×
        let dir = dir.into();
6433091×
        fs::create_dir_all(dir.join("loose"))?;
644
        #[cfg(test)]
6451428×
        if let Some(backend) = TEST_BACKEND.with(|t| t.borrow().clone()) {
6469×
            return Store::open_on(backend, dir);
6471419×
        }
6483082×
        if let Some(backend) = memory_backend_for(&dir) {
6490×
            return Store::open_on(backend, dir);
6503082×
        }
6513082×
        let backend: Arc<Backend> =
6523082×
            Arc::new(LocalStore::open(&dir).map_err(|e| io::Error::other(e.to_string()))?);
6533082×
        let mut store = Store::open_on(backend, dir)?;
6542973×
        store.local = true;
6552973×
        Ok(store)
6563091×
    }
657
658
    /// Open the store kept on `backend`, with `scratch` for what must be
659
    /// on this disk: a pack while it is being written, and caches.
6603170×
    pub fn open_on(backend: Arc<Backend>, scratch: impl Into<PathBuf>) -> io::Result<Store> {
6613170×
        let scratch = scratch.into();
6623170×
        fs::create_dir_all(&scratch)?;
6633170×
        let store = Store {
6643170×
            backend,
6653170×
            scratch,
6663170×
            local: false,
6673170×
            packs: RefCell::new(Vec::new()),
6683170×
            packs_listed: RefCell::new(Vec::new()),
6693170×
            has_relisted: std::cell::Cell::new(None),
6703170×
            pending: RefCell::new(None),
6713170×
            batch_depth: std::cell::Cell::new(0),
6723170×
            metas: RefCell::new(None),
6733170×
            entry_lists: RefCell::new(Vec::new()),
6743170×
            folded: std::cell::Cell::new(false),
6753170×
            legacy_folded: std::cell::Cell::new(false),
6763170×
            entries: RefCell::new(None),
6773170×
            entries_read: std::cell::Cell::new(None),
6783170×
            lists_read: RefCell::new(Default::default()),
6793170×
            journal_queue: RefCell::new(Vec::new()),
6803170×
        };
6813170×
        store
6823170×
            .reload_packs()
6833170×
            .map_err(|e| io::Error::other(e.to_string()))?;
6843061×
        Ok(store)
6853170×
    }
686
687
    /// The storage under this store.
68858×
    pub fn backend(&self) -> &Arc<Backend> {
68958×
        &self.backend
69058×
    }
691
692
    /// Whether the store is the directory `dir()` names, where every key
693
    /// is a file.
69414454×
    pub fn is_local(&self) -> bool {
69514454×
        self.local
69614454×
    }
697
698
    /// Load the list of packs again, from a new listing: what other
699
    /// writers packed since becomes readable, and packs since retired or
700
    /// deleted are let go. Packs already loaded are kept, not read again.
701497×
    pub fn refresh(&self) -> Result<()> {
702497×
        *self.entries.borrow_mut() = None;
703497×
        self.reload_packs()
704497×
    }
705
706
    /// The pack listing, with how `packs_listed` records it.
7076225×
    fn list_packs(&self) -> Result<(Vec<strata::Entry>, Listed)> {
7086225×
        let listed = self.backend.list("pack-").map_err(backend_err)?;
7096225×
        let sig = listed
7106225×
            .iter()
7116225×
            .map(|e| (e.key.clone(), e.size, e.modified_ms))
7126225×
            .collect();
7136225×
        Ok((listed, sig))
7146225×
    }
715
716
    /// Load every pack the listing names that is not retired. An index
717
    /// that is gone by the time it is read (its pack deleted after the
718
    /// listing) is passed over, and the store listed once more, since
719
    /// what replaced it may be new since the first listing.
7203936×
    fn reload_packs(&self) -> Result<()> {
7213936×
        let (listed, sig) = self.list_packs()?;
7223936×
        self.load_listed(listed, sig)
7233936×
    }
724
7253945×
    fn load_listed(&self, mut listed: Vec<strata::Entry>, mut sig: Listed) -> Result<()> {
7263945×
        let had: std::collections::HashMap<String, Arc<PackIndex>> = self
7273945×
            .packs
7283945×
            .borrow()
7293945×
            .iter()
7303945×
            .map(|p| (p.key.clone(), p.clone()))
7313945×
            .collect();
7323945×
        let mut loaded: std::collections::HashMap<String, Arc<PackIndex>> = Default::default();
7333957×
        for attempt in 0..2 {
7343957×
            let retired = self.retired_in(&listed)?;
7353957×
            let mut idx_keys: Vec<&str> = listed
7363957×
                .iter()
7373957×
                .map(|e| e.key.as_str())
7383957×
                .filter(|k| k.ends_with(".idx"))
7393957×
                .collect();
7403957×
            idx_keys.sort();
7413957×
            let mut packs = Vec::new();
7423957×
            let mut vanished = false;
7433957×
            for idx in idx_keys {
7441500×
                let pack = format!("{}.pack", idx.trim_end_matches(".idx"));
7451500×
                if retired.contains_key(&pack) {
746174×
                    continue;
7471326×
                }
74811391×
                let Some(p) = listed.iter().find(|e| e.key == pack) else {
7490×
                    continue;
750
                };
7511326×
                if let Some(kept) = had.get(&pack).or_else(|| loaded.get(&pack)) {
752827×
                    packs.push(kept.clone());
753827×
                    continue;
754499×
                }
755499×
                match PackIndex::load(self.backend.as_ref(), idx, &pack, p.size)? {
756366×
                    Some(index) => {
757366×
                        let index = Arc::new(index);
758366×
                        loaded.insert(pack, index.clone());
759366×
                        packs.push(index);
760366×
                    }
76124×
                    None => vanished = true,
762
                }
763
            }
7643848×
            if vanished && attempt == 0 {
76512×
                (listed, sig) = self.list_packs()?;
76612×
                continue;
7673836×
            }
7683836×
            let lists: Vec<String> = listed
7693836×
                .iter()
7703836×
                .map(|e| e.key.as_str())
7713836×
                .filter(|k| is_entry_list(k) && !retired.contains_key(*k))
7723836×
                .map(str::to_string)
7733836×
                .collect();
7743836×
            self.folded
7753836×
                .set(lists.iter().any(|k| k.ends_with(LIST_SUFFIX)));
7763836×
            self.legacy_folded
7773836×
                .set(listed.iter().any(|e| e.key == LEGACY_FOLDED_KEY));
7783836×
            *self.entry_lists.borrow_mut() = lists;
7793836×
            if *self.packs_listed.borrow() != sig {
780535×
                *self.entries.borrow_mut() = None;
7813301×
            }
7823836×
            *self.packs.borrow_mut() = packs;
7833836×
            *self.packs_listed.borrow_mut() = sig;
7843836×
            return Ok(());
785
        }
7860×
        unreachable!("the second round always returns")
7873945×
    }
788
789
    /// Load the packs again when the pack listing differs from the one
790
    /// they were loaded from; whether it did. What a miss calls before it
791
    /// is believed (oxygenhub#249): another writer may have packed the
792
    /// object, or a repack moved it, since this store loaded its packs.
793589×
    fn reload_if_changed(&self) -> Result<bool> {
794589×
        let (listed, sig) = self.list_packs()?;
795589×
        if *self.packs_listed.borrow() == sig {
796580×
            return Ok(false);
7979×
        }
7989×
        self.load_listed(listed, sig)?;
7999×
        Ok(true)
800589×
    }
801
802
    /// The retired record, read when `listed` (a pack listing) shows it
803
    /// holds anything.
8045645×
    fn retired_in(
8055645×
        &self,
8065645×
        listed: &[strata::Entry],
8075645×
    ) -> Result<std::collections::BTreeMap<String, u64>> {
8085645×
        match listed.iter().find(|e| e.key == RETIRED_KEY) {
809693×
            Some(e) if e.size > 0 => Ok(self.read_retired()?.0),
8104990×
            _ => Ok(Default::default()),
811
        }
8125645×
    }
813
814
    /// The retired record and its version (`None`: never written).
815667×
    fn read_retired(
816667×
        &self,
817667×
    ) -> Result<(
818667×
        std::collections::BTreeMap<String, u64>,
819667×
        Option<strata::Version>,
820667×
    )> {
821667×
        self.read_record(RETIRED_KEY)
822667×
    }
823
824
    /// A record of `<wall ms> <key>` lines (the retired record, the
825
    /// journal) and its version (`None`: never written).
8261171×
    fn read_record(
8271171×
        &self,
8281171×
        key: &str,
8291171×
    ) -> Result<(
8301171×
        std::collections::BTreeMap<String, u64>,
8311171×
        Option<strata::Version>,
8321171×
    )> {
8331171×
        match self.backend.read_ref(key).map_err(backend_err)? {
8341115×
            Some((bytes, v)) => Ok((parse_retired(&bytes), Some(v))),
83556×
            None => Ok((Default::default(), None)),
836
        }
8371171×
    }
838
839
    /// Change the retired record by compare-and-swap, reading it again
840
    /// whenever another writer changed it first. `edit` must give the
841
    /// same result however often it runs.
842127×
    fn update_retired(
843127×
        &self,
844127×
        edit: impl FnMut(&mut std::collections::BTreeMap<String, u64>),
845127×
    ) -> Result<()> {
846127×
        self.update_record(RETIRED_KEY, "the record of retired packs", edit)
847127×
    }
848
849
    /// Change a record of `<wall ms> <key>` lines by compare-and-swap, as
850
    /// [`Store::update_retired`] does; `what` names it in the error when
851
    /// other writers kept winning.
852366×
    fn update_record(
853366×
        &self,
854366×
        key: &str,
855366×
        what: &str,
856366×
        mut edit: impl FnMut(&mut std::collections::BTreeMap<String, u64>),
857366×
    ) -> Result<()> {
858366×
        for _ in 0..RETIRED_TRIES {
859462×
            let (mut m, version) = self.read_record(key)?;
860462×
            let before = m.clone();
861462×
            edit(&mut m);
862
            // Unchanged; a record never written stays unwritten.
863462×
            if m == before && (version.is_some() || m.is_empty()) {
8642×
                return Ok(());
865460×
            }
866460×
            match self
867460×
                .backend
868460×
                .write_ref(key, &encode_retired(&m), version.as_ref())
869
            {
870364×
                Ok(_) => return Ok(()),
87196×
                Err(BackendError::Conflict(_)) => continue,
8720×
                Err(e) => return Err(backend_err(e)),
873
            }
874
        }
8750×
        Err(StoreError::Io(io::Error::other(format!(
8760×
            "other writers kept changing {what}; run the command again"
8770×
        ))))
878366×
    }
879
880
    /// Record `keys` (packs, by their `.pack` key, and loose objects) as
881
    /// retired now: kept for the readers still using them, and deleted by
882
    /// a collection at least [`RETIRED_FOR`] later. A key already retired
883
    /// keeps its first time.
884114×
    fn retire(&self, keys: &[String]) -> Result<()> {
885114×
        if keys.is_empty() {
88610×
            return Ok(());
887104×
        }
888104×
        let now = now_ms();
889104×
        self.update_retired(|m| {
8902533×
            for k in keys {
8912533×
                m.entry(k.clone()).or_insert(now);
8922533×
            }
893104×
        })
894114×
    }
895
896
    /// Delete what was retired at least `older_than` ago (a collection
897
    /// does, with [`RETIRED_FOR`]): each pack with its index and
898
    /// dictionary, and each loose object a pack now holds. A retired loose
899
    /// object no pack holds any more (the pack that had it was pruned, and
900
    /// the object stored again since) is kept, as an ordinary loose object.
901747×
    pub fn expire_retired(
902747×
        &self,
903747×
        older_than: std::time::Duration,
904747×
        progress: Progress,
905747×
    ) -> Result<RetiredReport> {
906747×
        let (listed, _) = self.list_packs()?;
907747×
        let retired = self.retired_in(&listed)?;
908747×
        let mut report = RetiredReport::default();
909747×
        if retired.is_empty() {
910722×
            return Ok(report);
91125×
        }
91225×
        let now = now_ms();
91325×
        let older_than = older_than.as_millis() as u64;
91425×
        let due: Vec<(String, u64)> = retired
91525×
            .iter()
9161970×
            .filter(|(_, t)| now.saturating_sub(**t) >= older_than)
9171723×
            .map(|(k, t)| (k.clone(), *t))
91825×
            .collect();
91925×
        report.kept = retired.len() - due.len();
92025×
        if due.is_empty() {
92112×
            return Ok(report);
92213×
        }
92313×
        progress(&format!("deleting {} retired files", due.len()));
924
        // Which objects the live packs hold.
92513×
        self.reload_packs()?;
9261723×
        for (key, _) in &due {
9271723×
            if key.ends_with(".pack") {
92812×
                self.remove_pack(key)?;
92912×
                report.packs += 1;
9301711×
            } else if is_entry_list(key) {
931
                // An entry list a fold replaced (spec §13.7).
9321×
                delete(self.backend.as_ref(), key)?;
9331710×
            } else if let Some(r) = loose_ref(key) {
9341710×
                if self.find_packed(&r).is_some() {
9351708×
                    delete(self.backend.as_ref(), key)?;
9361708×
                    report.loose += 1;
9372×
                }
9380×
            }
939
        }
94013×
        self.update_retired(|m| {
9411723×
            for (k, t) in &due {
9421723×
                if m.get(k) == Some(t) {
9431723×
                    m.remove(k);
9441723×
                }
945
            }
94613×
        })?;
94713×
        self.reload_packs()?;
94813×
        Ok(report)
949747×
    }
950
951
    /// The loose objects retired and not yet deleted, by key.
952941×
    fn retired_loose(&self) -> Result<std::collections::HashSet<String>> {
953941×
        let (listed, _) = self.list_packs()?;
954941×
        Ok(self
955941×
            .retired_in(&listed)?
956941×
            .into_keys()
9575365×
            .filter(|k| k.starts_with("loose/"))
958941×
            .collect())
959941×
    }
960
96135×
    pub fn pack_count(&self) -> usize {
96235×
        self.packs.borrow().len()
96335×
    }
964
965105783×
    fn find_packed(&self, r: &ObjectRef) -> Option<(usize, PackEntry)> {
966105783×
        self.packs
967105783×
            .borrow()
968105783×
            .iter()
969105783×
            .enumerate()
970105783×
            .find_map(|(i, p)| p.find(r).map(|e| (i, e)))
971105783×
    }
972
973
    /// Loose objects, with their keys and ages; not those retired, whose
974
    /// objects a pack holds.
975909×
    fn loose_objects(&self) -> Result<Vec<(ObjectRef, strata::Entry)>> {
976909×
        let retired = self.retired_loose()?;
977909×
        let mut out = self.loose_objects_all()?;
97823466×
        out.retain(|(_, e)| !retired.contains(&e.key));
979909×
        Ok(out)
980909×
    }
981
982
    /// Every loose object, retired or not.
983941×
    fn loose_objects_all(&self) -> Result<Vec<(ObjectRef, strata::Entry)>> {
984941×
        let mut out: Vec<(ObjectRef, strata::Entry)> = self
985941×
            .backend
986941×
            .list("loose/")
987941×
            .map_err(backend_err)?
988941×
            .into_iter()
98924017×
            .filter_map(|e| loose_ref(&e.key).map(|r| (r, e)))
990941×
            .collect();
991941×
        out.sort_by_key(|(r, _)| *r);
992941×
        Ok(out)
993941×
    }
994
995729×
    pub fn loose_count(&self) -> io::Result<usize> {
996729×
        self.loose_objects()
997729×
            .map(|l| l.len())
998729×
            .map_err(|e| io::Error::other(e.to_string()))
999729×
    }
1000
1001
    /// Move every loose object into a new pack (spec §13.4). Loose objects
1002
    /// are deleted only after the pack and its index are durable.
100354×
    pub fn pack_loose(&self) -> Result<PackReport> {
100454×
        let loose = self.loose_objects()?;
100554×
        self.write_pack(
100654×
            &loose.iter().map(|(r, _)| *r).collect::<Vec<_>>(),
1007541×
            &loose.iter().map(|(_, e)| e.key.clone()).collect::<Vec<_>>(),
100854×
            &[],
1009
            ZSTD_LEVEL,
101054×
            &mut silent(),
1011
        )
101254×
    }
1013
1014
    /// Fold every pack and loose object into one pack compressed at
1015
    /// `level` (spec §13.4): what an explicit `oxy gc` does, as `git gc`
1016
    /// repacks. Returns the report and how many packs were folded in.
101713×
    pub fn repack_all(&self, level: i32) -> Result<(PackReport, usize)> {
101813×
        self.repack_all_with(level, &mut silent())
101913×
    }
1020
1021
    /// [`Store::repack_all`], reporting progress.
102237×
    pub fn repack_all_with(&self, level: i32, progress: Progress) -> Result<(PackReport, usize)> {
102337×
        let loose = self.loose_objects()?;
102437×
        let (mut refs, keys, trained): (Vec<ObjectRef>, Vec<String>, bool) = {
102537×
            let packs = self.packs.borrow();
1026
            (
102737×
                packs.iter().flat_map(|p| p.refs.iter().copied()).collect(),
102837×
                packs.iter().map(|p| p.key.clone()).collect(),
102937×
                packs.first().is_some_and(|p| p.dict.is_some()),
1030
            )
1031
        };
1032
        // Nothing loose and one pack already made by a full repack (only
1033
        // such a pack has a trained dictionary): rewriting it would give
1034
        // the same pack. A hub's nightly gc rewrote andymccall/oxygen's
1035
        // 48 000 objects (37 s under its write lock) every time.
103637×
        if loose.is_empty() && keys.len() <= 1 && (level == ZSTD_LEVEL || trained) {
10371×
            return Ok((PackReport::default(), 0));
103836×
        }
103936×
        refs.extend(loose.iter().map(|(r, _)| *r));
104036×
        let loose_keys: Vec<String> = loose.into_iter().map(|(_, e)| e.key).collect();
104136×
        let n = keys.len();
104236×
        let report = self.write_pack(&refs, &loose_keys, &keys, level, progress)?;
104336×
        Ok((report, n))
104437×
    }
1045
1046
    /// Fold every pack into one when there are more than `MAX_PACKS`
1047
    /// (spec §13.4). Returns how many packs were merged.
10484×
    pub fn merge_packs(&self) -> Result<usize> {
10492×
        let (refs, keys): (Vec<ObjectRef>, Vec<String>) = {
10504×
            let packs = self.packs.borrow();
10514×
            if packs.len() <= MAX_PACKS {
10522×
                return Ok(0);
10532×
            }
1054
            (
105536×
                packs.iter().flat_map(|p| p.refs.iter().copied()).collect(),
105636×
                packs.iter().map(|p| p.key.clone()).collect(),
1057
            )
1058
        };
10592×
        let n = keys.len();
10602×
        self.write_pack(&refs, &[], &keys, ZSTD_LEVEL, &mut silent())?;
10612×
        Ok(n)
10624×
    }
1063
1064
    /// Delete every object not in `keep` (spec §21). Loose objects and packs
1065
    /// younger than `grace` are left alone, since a concurrent writer may not
1066
    /// have recorded its references yet. Packs holding unreachable objects
1067
    /// are rewritten without them. Signatures, seal and receipt entries of
1068
    /// deleted objects go too.
10698×
    pub fn prune(
10708×
        &self,
10718×
        keep: &std::collections::HashSet<ObjectRef>,
10728×
        grace: std::time::Duration,
10738×
    ) -> Result<PruneReport> {
10748×
        self.prune_with(keep, grace, &mut silent())
10758×
    }
1076
1077
    /// [`Store::prune`], reporting progress.
107832×
    pub fn prune_with(
107932×
        &self,
108032×
        keep: &std::collections::HashSet<ObjectRef>,
108132×
        grace: std::time::Duration,
108232×
        progress: Progress,
108332×
    ) -> Result<PruneReport> {
108432×
        progress("pruning loose objects");
108532×
        let now = now_ms();
108632×
        let grace_ms = grace.as_millis() as u64;
1087
        // An object whose age the store does not know is old enough.
108832×
        let old_enough =
108987×
            |modified: Option<u64>| modified.is_none_or(|t| now.saturating_sub(t) >= grace_ms);
109032×
        let mut report = PruneReport::default();
109132×
        let mut gone: Vec<String> = Vec::new();
1092
        // Loose objects. A retired one (its object packed) whose object is
1093
        // unreachable goes with it, uncounted: the pack's copy is counted.
109432×
        let retired = self.retired_loose()?;
109532×
        let mut unretired: Vec<String> = Vec::new();
1096551×
        for (r, entry) in self.loose_objects_all()? {
1097551×
            if keep.contains(&r) {
1098473×
                continue;
109978×
            }
110078×
            if retired.contains(&entry.key) {
110145×
                if old_enough(entry.modified_ms) {
110241×
                    delete(self.backend.as_ref(), &entry.key)?;
110341×
                    unretired.push(entry.key);
11044×
                }
110545×
                continue;
110633×
            }
110733×
            if !old_enough(entry.modified_ms) {
11084×
                report.deferred += 1;
11094×
                continue;
111029×
            }
111129×
            report.bytes += entry.size;
111229×
            report.objects += 1;
111329×
            delete(self.backend.as_ref(), &entry.key)?;
111429×
            self.remove_sidecars(&r, &mut gone)?;
1115
        }
111632×
        if !unretired.is_empty() {
11176×
            self.update_retired(|m| {
111841×
                for k in &unretired {
111941×
                    m.remove(k);
112041×
                }
11216×
            })?;
112226×
        }
1123
        // Packs: rewrite those that hold unreachable objects.
112432×
        let mut keep_refs: Vec<ObjectRef> = Vec::new();
112532×
        let mut old_keys: Vec<String> = Vec::new();
112632×
        let mut rewrite = false;
112732×
        progress("pruning packs");
112832×
        for p in self.packs.borrow().iter() {
1129213×
            let unreachable = p.refs.iter().filter(|r| !keep.contains(r)).count();
113015×
            if unreachable == 0 {
11316×
                continue;
11329×
            }
11339×
            let modified = self
11349×
                .backend
11359×
                .stat(&p.key)
11369×
                .map_err(backend_err)?
11379×
                .and_then(|e| e.modified_ms);
11389×
            if !old_enough(modified) {
11393×
                report.deferred += unreachable;
11403×
                continue;
11416×
            }
11426×
            rewrite = true;
11436×
            old_keys.push(p.key.clone());
1144102×
            for (r, e) in p.refs.iter().zip(p.entries.iter()) {
1145102×
                if keep.contains(r) {
114661×
                    keep_refs.push(*r);
114761×
                } else {
114841×
                    report.objects += 1;
114941×
                    report.bytes += e.stored_len as u64;
115041×
                }
1151
            }
1152
        }
115332×
        if rewrite {
11546×
            let dropped: Vec<ObjectRef> = {
11556×
                let packs = self.packs.borrow();
11566×
                packs
11576×
                    .iter()
11586×
                    .filter(|p| old_keys.contains(&p.key))
11596×
                    .flat_map(|p| p.refs.iter().copied())
1160102×
                    .filter(|r| !keep.contains(r))
11616×
                    .collect()
1162
            };
11636×
            self.write_pack(&keep_refs, &[], &old_keys, ZSTD_LEVEL, progress)?;
116441×
            for r in dropped {
116541×
                self.remove_sidecars(&r, &mut gone)?;
1166
            }
116726×
        }
116832×
        self.journal_forget(&gone)?;
116932×
        Ok(report)
117032×
    }
1171
1172
    /// Everything kept beside an object: its signatures and the seal and
1173
    /// receipt entries under it, in both layouts.
1174
    /// The entry keys deleted are added to `gone`, for the journal.
117570×
    fn remove_sidecars(&self, r: &ObjectRef, gone: &mut Vec<String>) -> Result<()> {
1176210×
        for index in ["sigs", "seals", "receipts"] {
1177210×
            delete(self.backend.as_ref(), &legacy_key(index, r))?;
1178
        }
1179210×
        for index in ["sig", "seal", "receipt"] {
1180210×
            for e in self
1181210×
                .backend
1182210×
                .list(&fanout_prefix(index, r))
1183210×
                .map_err(backend_err)?
1184
            {
118515×
                delete(self.backend.as_ref(), &e.key)?;
118615×
                gone.push(e.key);
1187
            }
1188
        }
118970×
        Ok(())
119070×
    }
1191
1192
    /// Take entry keys a collection deleted out of the journal, so the
1193
    /// next complete fold leaves them out (spec §13.7).
119432×
    fn journal_forget(&self, gone: &[String]) -> Result<()> {
119532×
        if self.local || gone.is_empty() {
119630×
            return Ok(());
11972×
        }
11982×
        if let Some(set) = self.entries.borrow_mut().as_mut() {
11990×
            for k in gone {
12000×
                set.remove(k);
12010×
            }
12022×
        }
12032×
        self.update_record(JOURNAL_KEY, "the journal of entries", |m| {
12043×
            for k in gone {
12053×
                m.remove(k);
12063×
            }
12072×
        })
120832×
    }
1209
1210
    /// Put a complete, synced temporary pack file and its index into the
1211
    /// store; `index` must be sorted by ObjectRef. The pack goes first: an
1212
    /// index never names a pack that is not there.
1213109×
    fn finish_pack_files(
1214109×
        &self,
1215109×
        id: &str,
1216109×
        tmp_pack: &Path,
1217109×
        index: &[(ObjectRef, PackEntry)],
1218109×
    ) -> Result<()> {
1219109×
        let mut idx = Vec::with_capacity(12 + index.len() * IDX_ENTRY);
1220109×
        idx.extend_from_slice(IDX_MAGIC);
1221109×
        idx.extend_from_slice(&(index.len() as u32).to_le_bytes());
12227897×
        for (r, e) in index {
12237897×
            idx.extend_from_slice(&r.0);
12247897×
            idx.extend_from_slice(&e.offset.to_le_bytes());
12257897×
            idx.extend_from_slice(&e.stored_len.to_le_bytes());
12267897×
            idx.extend_from_slice(&e.raw_len.to_le_bytes());
12277897×
            idx.extend_from_slice(&[type_index(e.ty), e.flags]);
12287897×
        }
1229109×
        let put = self
1230109×
            .backend
1231109×
            .put_new_file(&format!("pack-{id}.pack"), tmp_pack)
1232109×
            .map_err(backend_err);
1233
        // A store on another disk copied it; the one in this directory
1234
        // moved it and this finds nothing.
1235109×
        let _ = fs::remove_file(tmp_pack);
1236109×
        put?;
1237109×
        self.backend
1238109×
            .put_new(&format!("pack-{id}.idx"), &idx)
1239109×
            .map_err(backend_err)?;
1240109×
        self.reload_packs()
1241109×
    }
1242
1243
    /// Remove a pack with its index and dictionary: the index first, so
1244
    /// a store opened meanwhile does not load a pack that is going.
124512×
    fn remove_pack(&self, pack_key: &str) -> Result<()> {
124612×
        let stem = pack_key.trim_end_matches(".pack");
124712×
        delete(self.backend.as_ref(), &format!("{stem}.idx"))?;
124812×
        delete(self.backend.as_ref(), pack_key)?;
124912×
        delete(self.backend.as_ref(), &format!("{stem}.dict"))
125012×
    }
1251
1252
    /// Write `refs` into a new pack, then retire `loose_keys` and
1253
    /// `old_packs`: they stay in the store, unused by stores opened from
1254
    /// now on, until a collection [`RETIRED_FOR`] later deletes them, so a
1255
    /// reader that loaded its packs before goes on reading (oxygenhub#249).
1256100×
    fn write_pack(
1257100×
        &self,
1258100×
        refs: &[ObjectRef],
1259100×
        loose_keys: &[String],
1260100×
        old_packs: &[String],
1261100×
        level: i32,
1262100×
        progress: Progress,
1263100×
    ) -> Result<PackReport> {
1264100×
        if refs.is_empty() {
12650×
            self.retire(old_packs)?;
12660×
            self.reload_packs()?;
12670×
            return Ok(PackReport::default());
1268100×
        }
1269100×
        let mut refs: Vec<ObjectRef> = refs.to_vec();
1270100×
        refs.sort();
1271100×
        refs.dedup();
1272100×
        let id = fresh_id();
1273100×
        let tmp_pack = self.scratch.join(format!(".tmp-{id}.pack"));
1274100×
        let mut report = PackReport::default();
1275100×
        let mut index: Vec<(ObjectRef, PackEntry)> = Vec::with_capacity(refs.len());
1276
        // A full repack of enough objects trains a dictionary for them.
1277100×
        let dict = if level == REPACK_ZSTD_LEVEL && refs.len() >= DICT_MIN_OBJECTS {
12782×
            let _t = crate::timing::phase("pack: train a dictionary");
12792×
            progress(&format!("training a dictionary on {} objects", refs.len()));
12802×
            self.train_dictionary(&refs)
1281
        } else {
128298×
            None
1283
        };
1284100×
        let encoder = dict
1285100×
            .as_ref()
1286100×
            .map(|d| zstd::dict::EncoderDictionary::copy(d, level));
1287100×
        let nocompress = self.nocompress_set()?;
1288
        {
1289100×
            let mut f = io::BufWriter::new(fs::File::create(&tmp_pack)?);
1290100×
            f.write_all(PACK_MAGIC)?;
1291100×
            let mut offset = PACK_MAGIC.len() as u64;
1292
            // Read a batch (the store is not shared between threads),
1293
            // compress it on every core, write it in order.
1294100×
            let mut batches = refs.chunks(1024);
1295100×
            let mut compressed_batch = Vec::new();
1296100×
            let total = refs.len();
1297100×
            let mut reported = usize::MAX;
1298
            loop {
12994771×
                if compressed_batch.is_empty() {
1300202×
                    let Some(batch) = batches.next() else { break };
1301
                    // Every tenth, and the first batch.
1302102×
                    let tenth = report.objects * 10 / total.max(1);
1303102×
                    if tenth != reported {
1304102×
                        reported = tenth;
1305102×
                        progress(&format!("packing {} of {total} objects", report.objects));
1306102×
                    }
1307102×
                    let mut read: Vec<(ObjectRef, ObjectType, Vec<u8>, bool)> =
1308102×
                        Vec::with_capacity(batch.len());
13094671×
                    for r in batch {
13104671×
                        let (ty, body) = self.get(r)?;
13114671×
                        read.push((*r, ty, body, nocompress.contains(r)));
1312
                    }
1313
                    use rayon::prelude::*;
1314102×
                    compressed_batch = read
1315102×
                        .into_par_iter()
1316102×
                        .map_init(
13171447×
                            || {
13181447×
                                encoder.as_ref().and_then(|e| {
1319369×
                                    zstd::bulk::Compressor::with_prepared_dictionary(e).ok()
1320369×
                                })
13211447×
                            },
13224671×
                            |with_dict, (r, ty, body, raw)| {
13234671×
                                let packed = if raw {
13242×
                                    None
1325
                                } else {
1326
                                    // Small objects with the dictionary;
1327
                                    // a large one has context enough.
13284669×
                                    let dicted = with_dict
13294669×
                                        .as_mut()
13304669×
                                        .filter(|_| body.len() <= DICT_OBJECT_MAX)
13314669×
                                        .and_then(|c| c.compress(&body).ok())
13324669×
                                        .map(|c| (FLAG_ZSTD | FLAG_DICT, c));
13334669×
                                    dicted
13344669×
                                        .or_else(|| {
13351668×
                                            zstd::bulk::compress(&body, level)
13361668×
                                                .ok()
13371668×
                                                .map(|c| (FLAG_ZSTD, c))
13381668×
                                        })
13394669×
                                        .filter(|(_, c)| c.len() < body.len())
1340
                                };
13414671×
                                (r, ty, body, packed)
13424671×
                            },
1343
                        )
1344102×
                        .collect::<Vec<_>>();
1345102×
                    compressed_batch.reverse();
13464569×
                }
13474671×
                let Some((r, ty, body, packed)) = compressed_batch.pop() else {
13480×
                    continue;
1349
                };
13504671×
                let r = &r;
13514671×
                let (flags, stored) = match packed {
13523807×
                    Some(c) => c,
1353864×
                    None => (0, body.clone()),
1354
                };
13554671×
                let header_len = 1 + 1 + 4 + 4 + 32;
13564671×
                f.write_all(&[type_index(ty), flags])?;
13574671×
                f.write_all(&(stored.len() as u32).to_le_bytes())?;
13584671×
                f.write_all(&(body.len() as u32).to_le_bytes())?;
13594671×
                f.write_all(&r.0)?;
13604671×
                f.write_all(&stored)?;
13614671×
                index.push((
13624671×
                    *r,
13634671×
                    PackEntry {
13644671×
                        offset: offset + header_len,
13654671×
                        stored_len: stored.len() as u32,
13664671×
                        raw_len: body.len() as u32,
13674671×
                        ty,
13684671×
                        flags,
13694671×
                    },
13704671×
                ));
13714671×
                offset += header_len + stored.len() as u64;
13724671×
                report.objects += 1;
13734671×
                report.raw_bytes += body.len() as u64;
13744671×
                report.packed_bytes += stored.len() as u64;
1375
            }
1376100×
            f.flush()?;
1377100×
            f.get_ref().sync_all()?;
1378
        }
1379
        // The dictionary before the pack that needs it becomes visible.
1380100×
        if let Some(d) = &dict {
13812×
            self.backend
13822×
                .put_new(&format!("pack-{id}.dict"), d)
13832×
                .map_err(backend_err)?;
138498×
        }
1385100×
        progress(&format!("writing the index of {} objects", index.len()));
1386100×
        self.finish_pack_files(&id, &tmp_pack, &index)?;
1387100×
        let mut retired = old_packs.to_vec();
1388100×
        retired.extend_from_slice(loose_keys);
1389100×
        self.retire(&retired)?;
1390100×
        self.reload_packs()?;
1391100×
        Ok(report)
1392100×
    }
1393
1394
    /// A dictionary for these objects, trained on a sample of them (at
1395
    /// most 10 000, 32 MB); `None` when training fails.
13962×
    fn train_dictionary(&self, refs: &[ObjectRef]) -> Option<Vec<u8>> {
13972×
        let step = (refs.len() / 10_000).max(1);
13982×
        let mut samples: Vec<Vec<u8>> = Vec::new();
13992×
        let mut total = 0usize;
14003001×
        for r in refs.iter().step_by(step) {
14013001×
            if total > 32 << 20 {
14020×
                break;
14033001×
            }
14043001×
            let Ok((ty, body)) = self.get(r) else {
14050×
                continue;
1406
            };
14070×
            if matches!(
14083001×
                ty,
1409
                ObjectType::Chunk | ObjectType::Blob | ObjectType::Change
14103001×
            ) && !body.is_empty()
14113001×
                && body.len() < 1 << 20
14123001×
            {
14133001×
                total += body.len();
14143001×
                samples.push(body);
14153001×
            }
1416
        }
14172×
        zstd::dict::from_samples(&samples, DICT_BYTES).ok()
14182×
    }
1419
1420
    /// The scratch directory: for a store in a directory, that directory.
142119×
    pub fn dir(&self) -> &Path {
142219×
        &self.scratch
142319×
    }
1424
1425
    /// Where a loose object is kept, for a store in a directory (tests
1426
    /// tamper with and remove objects there).
1427
    #[cfg(test)]
14284×
    pub(crate) fn path_for(&self, r: &ObjectRef) -> PathBuf {
14294×
        self.scratch.join(loose_key(r))
14304×
    }
1431
1432
    /// Remove a loose object, whatever the store is on (tests make an
1433
    /// object go missing).
1434
    #[cfg(test)]
14353×
    pub(crate) fn remove_loose_for_test(&self, r: &ObjectRef) -> Result<()> {
14363×
        delete(self.backend.as_ref(), &loose_key(r))
14373×
    }
1438
1439
    /// Whether the store holds `r`. A miss looks at the pack listing
1440
    /// again before it is believed, as [`Store::get`]'s does; a writer
1441
    /// asking about many objects it is about to store looks at most once
1442
    /// in [`HAS_RELIST_AFTER`] while nothing new turns up.
144312952×
    pub fn has(&self, r: &ObjectRef) -> bool {
144412952×
        if self.has_now(r) {
144512228×
            return true;
1446724×
        }
1447724×
        let recent = self
1448724×
            .has_relisted
1449724×
            .get()
1450724×
            .is_some_and(|t| t.elapsed() < HAS_RELIST_AFTER);
1451724×
        if recent {
1452192×
            return false;
1453532×
        }
1454532×
        match self.reload_if_changed() {
1455
            Ok(true) => {
14565×
                self.has_relisted.set(None);
14575×
                self.has_now(r)
1458
            }
1459
            _ => {
1460527×
                self.has_relisted.set(Some(std::time::Instant::now()));
1461527×
                false
1462
            }
1463
        }
146412952×
    }
1465
1466
    /// [`Store::has`] against the packs as loaded.
146712957×
    fn has_now(&self, r: &ObjectRef) -> bool {
146812957×
        self.find_packed(r).is_some()
146912377×
            || self
147012377×
                .pending
147112377×
                .borrow()
147212377×
                .as_ref()
147312377×
                .is_some_and(|p| p.lookup.contains_key(r))
147412096×
            || self.backend.has(&loose_key(r)).unwrap_or(false)
147512957×
    }
1476
1477
    /// Start a write batch: until the matching `finish`, `put` collects
1478
    /// objects and writes them out together, with one `fsync` for a large
1479
    /// batch (a pack) instead of one per object. Batches nest; only the
1480
    /// outermost does the work. Objects are readable while pending.
14813011×
    pub fn begin_batch(&self) -> Result<Batch<'_>> {
14823011×
        let depth = self.batch_depth.get();
14833011×
        if depth == 0 {
14842429×
            *self.pending.borrow_mut() = Some(Pending {
14852429×
                buffered: vec![],
14862429×
                buffered_bytes: 0,
14872429×
                lookup: Default::default(),
14882429×
                pack: None,
14892429×
                report: PackReport::default(),
14902429×
            });
14912429×
        }
14923011×
        self.batch_depth.set(depth + 1);
14933011×
        Ok(Batch {
14943011×
            store: self,
14953011×
            done: false,
14963011×
        })
14973011×
    }
1498
14993010×
    fn finish_batch(&self) -> Result<()> {
15003010×
        let depth = self.batch_depth.get();
15013010×
        if depth == 0 {
15020×
            return Ok(());
15033010×
        }
15043010×
        self.batch_depth.set(depth - 1);
15053010×
        if depth > 1 {
1506582×
            return Ok(());
15072428×
        }
15082428×
        let Some(pending) = self.pending.borrow_mut().take() else {
15092×
            return self.flush_journal();
1510
        };
15112426×
        self.write_pending(pending)?;
15122424×
        self.flush_journal()
15133010×
    }
1514
1515
    /// Make every object written so far durable, even inside an open
1516
    /// batch, which stays open. Call before a reference that names them
1517
    /// is moved (spec §13.6): the operation head, a bookmark pointer.
15183855×
    pub fn flush(&self) -> Result<()> {
15193855×
        let Some(pending) = self.pending.borrow_mut().take() else {
15202775×
            return self.flush_journal();
1521
        };
15221080×
        self.write_pending(pending)?;
15231078×
        self.flush_journal()?;
15241078×
        *self.pending.borrow_mut() = Some(Pending {
15251078×
            buffered: vec![],
15261078×
            buffered_bytes: 0,
15271078×
            lookup: Default::default(),
15281078×
            pack: None,
15291078×
            report: PackReport::default(),
15301078×
        });
15311078×
        Ok(())
15323855×
    }
1533
15343506×
    fn write_pending(&self, pending: Pending) -> Result<()> {
15353506×
        match pending.pack {
1536
            None => {
15373498×
                let objects: Vec<(ObjectType, &[u8])> = pending
15383498×
                    .buffered
15393498×
                    .iter()
15405576×
                    .map(|(ty, body)| (*ty, body.as_slice()))
15413498×
                    .collect();
15423498×
                self.put_loose_many(&objects)
1543
            }
15448×
            Some(mut pack) => {
15458×
                pack.writer.flush()?;
15468×
                pack.writer.get_ref().sync_all()?;
15478×
                drop(pack.writer);
15488×
                pack.index.sort_by_key(|(r, _)| *r);
15498×
                self.finish_pack_files(&pack.id, &pack.tmp_path, &pack.index)?;
15508×
                Ok(())
1551
            }
1552
        }
15533506×
    }
1554
1555
    /// Whether a batch is open and holds `r`.
1556923×
    fn pending_get(&self, r: &ObjectRef) -> Option<Result<Object>> {
1557923×
        let mut guard = self.pending.borrow_mut();
1558923×
        let pending = guard.as_mut()?;
1559867×
        let entry = *pending.lookup.get(r)?;
1560867×
        Some(match entry {
1561685×
            PendingEntry::Buffered(i) => {
1562685×
                let (ty, body) = &pending.buffered[i];
1563685×
                Ok((*ty, body.clone()))
1564
            }
1565182×
            PendingEntry::Packed(e) => {
1566182×
                let Some(pack) = pending.pack.as_mut() else {
15670×
                    return Some(Err(StoreError::NotFound(*r)));
1568
                };
1569
                // Read from the pack file being written, on this disk.
1570182×
                (|| {
1571182×
                    pack.writer.flush()?;
1572
                    use std::io::{Read, Seek, SeekFrom};
1573182×
                    let mut f = fs::File::open(&pack.tmp_path)?;
1574182×
                    f.seek(SeekFrom::Start(e.offset))?;
1575182×
                    let mut stored = vec![0u8; e.stored_len as usize];
1576182×
                    f.read_exact(&mut stored)?;
1577182×
                    decode_entry(r, e, stored, None)
1578
                })()
1579
            }
1580
        })
1581923×
    }
1582
1583
    /// Append one object to the pending pack file.
15843026×
    fn pending_pack_write(
15853026×
        &self,
15863026×
        pack: &mut PendingPack,
15873026×
        r: &ObjectRef,
15883026×
        ty: ObjectType,
15893026×
        body: &[u8],
15903026×
        report: &mut PackReport,
15913026×
    ) -> Result<PackEntry> {
15923026×
        let (flags, stored) = if self.is_nocompress(r) {
1593120×
            (0, body.to_vec())
1594
        } else {
15952906×
            let compressed = zstd::bulk::compress(body, ZSTD_LEVEL)
15962906×
                .map_err(|e| StoreError::Corrupt(*r, format!("zstd: {e}")))?;
15972906×
            if compressed.len() < body.len() {
15982292×
                (FLAG_ZSTD, compressed)
1599
            } else {
1600614×
                (0, body.to_vec())
1601
            }
1602
        };
16033026×
        write_pack_entry(pack, r, ty, body.len(), flags, &stored, report)
16043026×
    }
1605
16068752×
    fn put_pending(&self, r: ObjectRef, ty: ObjectType, body: &[u8]) -> Result<()> {
16078752×
        let mut guard = self.pending.borrow_mut();
16088752×
        let pending = guard.as_mut().expect("batch open");
16098752×
        if pending.lookup.contains_key(&r) {
1610150×
            return Ok(());
16118602×
        }
16128602×
        if pending.pack.is_none()
16137538×
            && pending.buffered.len() < SPILL_OBJECTS
16147532×
            && pending.buffered_bytes + body.len() <= SPILL_BYTES
1615
        {
16167530×
            pending.buffered.push((ty, body.to_vec()));
16177530×
            pending.buffered_bytes += body.len();
16187530×
            pending
16197530×
                .lookup
16207530×
                .insert(r, PendingEntry::Buffered(pending.buffered.len() - 1));
16217530×
            return Ok(());
16221072×
        }
16231072×
        self.ensure_pack(pending)?;
16241072×
        let mut report = std::mem::take(&mut pending.report);
16251072×
        let pack = pending.pack.as_mut().unwrap();
16261072×
        let e = self.pending_pack_write(pack, &r, ty, body, &mut report)?;
16271072×
        pending.report = report;
16281072×
        pending.lookup.insert(r, PendingEntry::Packed(e));
16291072×
        Ok(())
16308752×
    }
1631
1632
    /// Open the pending batch's pack, moving the objects buffered so far
1633
    /// into it (a spill).
16341073×
    fn ensure_pack(&self, pending: &mut Pending) -> Result<()> {
16351073×
        if pending.pack.is_none() {
1636
            // Spill: open the pack and move the buffered objects into it.
16379×
            let id = fresh_id();
16389×
            let tmp_path = self.scratch.join(format!(".tmp-{id}.pack"));
16399×
            let mut writer = io::BufWriter::new(fs::File::create(&tmp_path)?);
16409×
            writer.write_all(PACK_MAGIC)?;
16419×
            let mut pack = PendingPack {
16429×
                id,
16439×
                tmp_path,
16449×
                writer,
16459×
                offset: PACK_MAGIC.len() as u64,
16469×
                index: vec![],
16479×
            };
16489×
            let buffered = std::mem::take(&mut pending.buffered);
16499×
            pending.buffered_bytes = 0;
16509×
            let mut report = std::mem::take(&mut pending.report);
16511954×
            for (bty, bbody) in &buffered {
16521954×
                let br = hash_object(*bty, bbody);
16531954×
                let e = self.pending_pack_write(&mut pack, &br, *bty, bbody, &mut report)?;
16541954×
                pending.lookup.insert(br, PendingEntry::Packed(e));
1655
            }
16569×
            pending.report = report;
16579×
            pending.pack = Some(pack);
16581064×
        }
16591073×
        Ok(())
16601073×
    }
1661
1662
    /// Store many objects at once. Inside a large write batch they are
1663
    /// compressed on every core and appended in order; otherwise each is
1664
    /// stored as [`Store::put`] does.
1665885×
    pub fn put_many(&self, objects: &[(ObjectType, &[u8])]) -> Result<Vec<ObjectRef>> {
1666
        use rayon::prelude::*;
16671109×
        let total: usize = objects.iter().map(|(_, b)| b.len()).sum();
1668885×
        if self.pending.borrow().is_none() || total < SPILL_BYTES {
1669907×
            return objects.iter().map(|(ty, b)| self.put(*ty, b)).collect();
16701×
        }
1671202×
        for (ty, b) in objects {
1672202×
            if b.len() > MAX_OBJECT_LEN as usize {
16730×
                return Err(StoreError::TooLarge {
16740×
                    kind: ty.name(),
16750×
                    size: b.len(),
16760×
                    limit: MAX_OBJECT_LEN,
16770×
                });
1678202×
            }
1679
        }
16801×
        let refs: Vec<ObjectRef> = objects
16811×
            .par_iter()
1682202×
            .map(|(ty, b)| hash_object(*ty, b))
16831×
            .collect();
1684
        // What is new: not stored already, not earlier in this call.
16851×
        let mut seen: std::collections::HashSet<ObjectRef> = std::collections::HashSet::new();
16861×
        let mut todo: Vec<usize> = Vec::new();
1687202×
        for (i, r) in refs.iter().enumerate() {
1688202×
            let known = self
1689202×
                .pending
1690202×
                .borrow()
1691202×
                .as_ref()
1692202×
                .is_some_and(|p| p.lookup.contains_key(r));
1693202×
            if seen.insert(*r)
1694201×
                && !known
1695201×
                && self.find_packed(r).is_none()
1696201×
                && !self.backend.has(&loose_key(r)).unwrap_or(false)
1697200×
            {
1698200×
                todo.push(i);
1699200×
            }
1700
        }
17011×
        let raw = self.nocompress_set()?;
17021×
        let packed: Vec<(u8, Vec<u8>)> = todo
17031×
            .par_iter()
1704200×
            .map(|i| {
1705200×
                let body = objects[*i].1;
1706200×
                if raw.contains(&refs[*i]) {
17070×
                    return (0, body.to_vec());
1708200×
                }
1709200×
                match zstd::bulk::compress(body, ZSTD_LEVEL) {
1710200×
                    Ok(c) if c.len() < body.len() => (FLAG_ZSTD, c),
17110×
                    _ => (0, body.to_vec()),
1712
                }
1713200×
            })
17141×
            .collect();
17151×
        let mut guard = self.pending.borrow_mut();
17161×
        let pending = guard.as_mut().expect("batch open");
17171×
        self.ensure_pack(pending)?;
17181×
        let mut report = std::mem::take(&mut pending.report);
17191×
        let pack = pending.pack.as_mut().unwrap();
1720200×
        for (i, (flags, stored)) in todo.iter().zip(packed) {
1721200×
            let (ty, body) = objects[*i];
1722200×
            let e = write_pack_entry(pack, &refs[*i], ty, body.len(), flags, &stored, &mut report)?;
1723200×
            pending.lookup.insert(refs[*i], PendingEntry::Packed(e));
1724
        }
17251×
        pending.report = report;
17261×
        Ok(refs)
1727885×
    }
1728
1729
    /// Store `body` as an object of type `ty`; idempotent.
173018939×
    pub fn put(&self, ty: ObjectType, body: &[u8]) -> Result<ObjectRef> {
1731
        // Refuse what the store could not read back (spec §13.2), before
1732
        // anything is written: a pack must never hold an entry its index
1733
        // would reject.
173418939×
        if body.len() > MAX_OBJECT_LEN as usize {
17351×
            return Err(StoreError::TooLarge {
17361×
                kind: ty.name(),
17371×
                size: body.len(),
17381×
                limit: MAX_OBJECT_LEN,
17391×
            });
174018938×
        }
174118938×
        let r = hash_object(ty, body);
174218938×
        if self.find_packed(&r).is_some() || self.backend.has(&loose_key(&r)).unwrap_or(false) {
17432186×
            return Ok(r);
174416752×
        }
174516752×
        if self.pending.borrow().is_some() {
17468752×
            self.put_pending(r, ty, body)?;
17478752×
            return Ok(r);
17488000×
        }
17498000×
        self.put_loose(ty, body)?;
17507995×
        Ok(r)
175118939×
    }
1752
1753
    /// Write one loose object, durable when this returns.
17548241×
    fn put_loose(&self, ty: ObjectType, body: &[u8]) -> Result<ObjectRef> {
17558241×
        let r = hash_object(ty, body);
17568241×
        let mut bytes = header(ty, body.len());
17578241×
        bytes.extend_from_slice(body);
17588241×
        crate::durable::step("write loose object")?;
17598236×
        put_new_or_existing(self.backend.as_ref(), &loose_key(&r), &bytes)?;
17608236×
        Ok(r)
17618241×
    }
1762
1763
    /// Write several loose objects with as few waits on the disk as the
1764
    /// guarantees allow (a directory store writes them all, then syncs
1765
    /// each directory once). Durable when this returns, as `put_loose`
1766
    /// one at a time would be (spec §13.6).
17673498×
    fn put_loose_many(&self, objects: &[(ObjectType, &[u8])]) -> Result<()> {
17683498×
        if objects.len() <= 1 {
17692024×
            for (ty, body) in objects {
1770241×
                self.put_loose(*ty, body)?;
1771
            }
17722024×
            return Ok(());
17731474×
        }
17741474×
        let mut seen = std::collections::HashSet::new();
17751474×
        let items: Vec<(String, Vec<u8>)> = objects
17761474×
            .iter()
17775335×
            .filter_map(|(ty, body)| {
17785335×
                let r = hash_object(*ty, body);
17795335×
                seen.insert(r).then(|| {
17805335×
                    let mut bytes = header(*ty, body.len());
17815335×
                    bytes.extend_from_slice(body);
17825335×
                    (loose_key(&r), bytes)
17835335×
                })
17845335×
            })
17851474×
            .collect();
17861474×
        self.backend.put_many_new(&items).map_err(backend_err)
17873498×
    }
1788
1789
    /// Read and verify an object, packed, loose, or pending in a batch.
1790
    /// The packs' indexes are in memory, so they are asked first: asking
1791
    /// for a loose copy of a packed object costs a request that finds
1792
    /// nothing on a store in a bucket, every time (oxygenhub#228).
1793
    ///
1794
    /// A miss is not believed at once (oxygenhub#249): when the pack it
1795
    /// was in is gone (deleted after a repack retired it) the packs are
1796
    /// loaded again, and when the object is nowhere the pack listing is
1797
    /// looked at, and the packs loaded again if it changed (another
1798
    /// writer packed it, or a repack moved it); then it is read once more.
179970408×
    pub fn get(&self, r: &ObjectRef) -> Result<Object> {
180070408×
        match self.get_now(r) {
1801178×
            Err(e) if is_miss(&e) => {
1802
                // Read from a pack this store loaded, which is gone.
180351×
                let packed_miss =
180451×
                    !matches!(e, StoreError::NotFound(_)) || self.find_packed(r).is_some();
180551×
                let changed = if packed_miss {
18062×
                    self.reload_packs()?;
18072×
                    true
1808
                } else {
180949×
                    self.reload_if_changed()?
1810
                };
181151×
                if changed {
18125×
                    self.get_now(r)
1813
                } else {
181446×
                    Err(e)
1815
                }
1816
            }
181770357×
            other => other,
1818
        }
181970408×
    }
1820
1821
    /// [`Store::get`] against the packs as loaded.
182270413×
    fn get_now(&self, r: &ObjectRef) -> Result<Object> {
182370413×
        if let Some((i, e)) = self.find_packed(r) {
182416290×
            let pack = self.packs.borrow()[i].clone();
182516290×
            return pack.fetch(self.backend.as_ref(), r, e);
182654123×
        }
182754123×
        match get_optional(self.backend.as_ref(), &loose_key(r))? {
182853207×
            Some(bytes) => parse_loose(&bytes, r),
1829916×
            None => self.pending_get(r).unwrap_or(Err(StoreError::NotFound(*r))),
1830
        }
183170413×
    }
1832
1833
    /// Many objects at once, read and verified in parallel: packed ones
1834
    /// from their packs, the others loose (as [`Store::get`], packs first),
1835
    /// then what is pending in a batch. Results are in the order of `refs`.
1836
    ///
1837
    /// Misses are looked for again as [`Store::get`]'s are, together.
18381054×
    pub fn get_many(&self, refs: &[ObjectRef]) -> Vec<Result<Object>> {
18391054×
        let mut out = self.get_many_now(refs);
18401054×
        let missed: Vec<usize> = out
18411054×
            .iter()
18421054×
            .enumerate()
18432816×
            .filter(|(_, g)| g.as_ref().is_err_and(is_miss))
18441054×
            .map(|(i, _)| i)
18451054×
            .collect();
18461054×
        if missed.is_empty() {
18471045×
            return out;
18489×
        }
18499×
        let packed_miss = missed.iter().any(|i| {
18509×
            !matches!(out[*i], Err(StoreError::NotFound(_)))
18519×
                || self.find_packed(&refs[*i]).is_some()
18529×
        });
18539×
        let changed = if packed_miss {
18542×
            self.reload_packs().map(|_| true)
1855
        } else {
18567×
            self.reload_if_changed()
1857
        };
18589×
        match changed {
1859
            Ok(true) => {
1860160×
                let again: Vec<ObjectRef> = missed.iter().map(|i| refs[*i]).collect();
1861160×
                for (i, g) in missed.into_iter().zip(self.get_many_now(&again)) {
1862160×
                    out[i] = g;
1863160×
                }
1864
            }
18657×
            Ok(false) => {}
18660×
            Err(e) => {
18670×
                let why = e.to_string();
18680×
                for i in missed {
18690×
                    out[i] = Err(StoreError::Io(io::Error::other(why.clone())));
18700×
                }
1871
            }
1872
        }
18739×
        out
18741054×
    }
1875
1876
    /// [`Store::get_many`] against the packs as loaded.
18771056×
    fn get_many_now(&self, refs: &[ObjectRef]) -> Vec<Result<Object>> {
1878
        use rayon::prelude::*;
18791056×
        let backend = self.backend.as_ref();
18801056×
        let packs = self.packs.borrow();
18811056×
        let place: Vec<Option<(usize, PackEntry)>> = refs
18821056×
            .iter()
18832976×
            .map(|r| {
18842976×
                packs
18852976×
                    .iter()
18862976×
                    .enumerate()
18872976×
                    .find_map(|(p, pack)| pack.find(r).map(|e| (p, e)))
18882976×
            })
18891056×
            .collect();
18901056×
        let indexes: &[Arc<PackIndex>] = &packs;
18911056×
        let found: Vec<Option<Result<Object>>> = refs
18921056×
            .par_iter()
18931056×
            .zip(place.par_iter())
18942976×
            .map(|(r, at)| match at {
18951067×
                Some((p, e)) => Some(indexes[*p].fetch(backend, r, *e)),
18961909×
                None => match get_optional(backend, &loose_key(r)) {
18971902×
                    Ok(Some(bytes)) => Some(parse_loose(&bytes, r)),
18987×
                    Ok(None) => None,
18990×
                    Err(e) => Some(Err(e)),
1900
                },
19012976×
            })
19021056×
            .collect();
19031056×
        drop(packs);
19041056×
        found
19051056×
            .into_iter()
19061056×
            .zip(refs)
19072976×
            .map(|(f, r)| {
19082976×
                f.unwrap_or_else(|| self.pending_get(r).unwrap_or(Err(StoreError::NotFound(*r))))
19092976×
            })
19101056×
            .collect()
19111056×
    }
1912
191349759×
    pub fn get_typed(&self, r: &ObjectRef, want: ObjectType) -> Result<Vec<u8>> {
191449759×
        let (ty, body) = self.get(r)?;
191549715×
        if ty != want {
19161×
            return Err(StoreError::WrongType(*r, ty.name(), want.name()));
191749714×
        }
191849714×
        Ok(body)
191949759×
    }
1920
1921
    /// Store a change and the text blobs (and their chunks) its large
1922
    /// inserts are stored as (spec §6.1).
19233173×
    pub fn put_change(&self, c: &Change) -> Result<ObjectRef> {
19243173×
        let encoding = crate::timing::phase("store: encode the change and its texts");
19253173×
        let bytes = c.encode_streaming(|batch| self.put_texts(&batch))?;
19263173×
        drop(encoding);
19273173×
        let _t = crate::timing::phase("store: put the encoding");
19283173×
        self.put(ObjectType::Change, &bytes)
19293173×
    }
1930
1931
    /// Store text blobs and their chunks: every chunk at once (compressed
1932
    /// on every core in a large batch), then the blobs.
193362×
    pub fn put_texts(&self, texts: &[crate::objects::StoredText]) -> Result<()> {
193462×
        let mut chunks: Vec<(ObjectType, &[u8])> = Vec::new();
193563×
        for t in texts {
193663×
            if !self.has(&t.blob.hash()) {
193752×
                chunks.extend(
193852×
                    crate::chunk::chunks(&t.content)
193952×
                        .into_iter()
194093×
                        .map(|c| (ObjectType::Chunk, c)),
1941
                );
194211×
            }
1943
        }
194462×
        self.put_many(&chunks)?;
194563×
        for t in texts {
194663×
            self.put(ObjectType::Blob, &t.blob.encode())?;
1947
        }
194862×
        Ok(())
194962×
    }
1950
19515091×
    pub fn get_change(&self, r: &ObjectRef) -> Result<Change> {
19525091×
        Ok(self.get_change_parts(r)?.0)
19535091×
    }
1954
1955
    /// A change and the objects its stored inserts are made of: each text
1956
    /// blob and its chunks, which travel and are kept with it.
19575255×
    pub fn get_change_parts(&self, r: &ObjectRef) -> Result<(Change, Vec<ObjectRef>)> {
19585255×
        let bytes = self.get_typed(r, ObjectType::Change)?;
19595242×
        let mut parts = Vec::new();
19605242×
        let mut failed = None;
19615242×
        let decoded = Change::decode_with(&bytes, &mut |text, count, form| match self
196259×
            .read_text(text)
1963
        {
196458×
            Ok((blob, content)) => {
196558×
                parts.push(*text);
196658×
                parts.extend(blob.chunks);
196758×
                crate::objects::text_lines(&content, count, form)
1968
            }
19691×
            Err(e) => {
19701×
                let why = e.to_string();
19711×
                failed = Some(e);
19721×
                Err(ObjectError::Schema(format!(
19731×
                    "text blob {}: {why}",
19741×
                    text.short(12)
19751×
                )))
1976
            }
197759×
        });
19785242×
        match (decoded, failed) {
19795241×
            (Ok(c), _) => Ok((c, parts)),
19801×
            (Err(_), Some(StoreError::NotFound(t))) => Err(StoreError::Corrupt(
19811×
                *r,
19821×
                format!("its text blob {} is missing", t.short(12)),
19831×
            )),
19840×
            (Err(e), _) => Err(StoreError::Corrupt(*r, e.to_string())),
1985
        }
19865255×
    }
1987
1988
    /// A change with every insert left without its lines (empty
1989
    /// placeholders), and where each one's lines are, in order: for
1990
    /// history-only states, which read text only for the graphs they build
1991
    /// (see [`Change::decode_skeleton`]).
1992572×
    pub fn get_change_lazy(&self, r: &ObjectRef) -> Result<(Change, Vec<crate::objects::TextRef>)> {
1993572×
        let bytes = self.get_typed(r, ObjectType::Change)?;
1994570×
        Change::decode_skeleton(&bytes, *r).map_err(|e| StoreError::Corrupt(*r, e.to_string()))
1995572×
    }
1996
1997
    /// A change's metadata (id, dependencies, author, clock, message, kind)
1998
    /// without its contents: read from its skeleton, so neither its text
1999
    /// blobs nor its inline lines are read (spec §11).
2000
    ///
2001
    /// Served from the `metas` cache when this store read it before (a
2002
    /// change object can be as large as the tree it adds), as long as the
2003
    /// store still has the change.
20044630×
    pub fn get_change_meta(&self, r: &ObjectRef) -> Result<crate::state::ChangeMeta> {
20054630×
        if self.metas.borrow().is_none() {
20061023×
            *self.metas.borrow_mut() = Some(self.read_metas_cache());
20073607×
        }
20084630×
        let cached = self.metas.borrow().as_ref().and_then(|m| m.get(r).cloned());
20094630×
        if let Some(meta) = cached {
20103346×
            if self.has(r) {
2011
                #[cfg(test)]
20121916×
                assert_eq!(
2013
                    meta,
20141916×
                    Change::decode_meta(&self.get_typed(r, ObjectType::Change)?).unwrap(),
2015
                    "the metadata cache holds a change's metadata"
2016
                );
20173346×
                return Ok(meta);
20180×
            }
20191284×
        }
20201284×
        let bytes = self.get_typed(r, ObjectType::Change)?;
20211284×
        let meta =
20221284×
            Change::decode_meta(&bytes).map_err(|e| StoreError::Corrupt(*r, e.to_string()))?;
20231284×
        self.remember_meta(r, &meta);
20241284×
        Ok(meta)
20254630×
    }
2026
2027
    /// The metadata cache: records of a change hash, its metadata in the
2028
    /// flat layout, and the first 8 bytes of BLAKE3 over both, each
2029
    /// appended in one write. Derived only: a record that does not check
2030
    /// ends the reading (a torn write), and the cache can be deleted. It
2031
    /// lives in the scratch directory, on this disk.
20321023×
    fn read_metas_cache(&self) -> crate::fasthash::FastMap<ObjectRef, crate::state::ChangeMeta> {
20331023×
        let mut out = crate::fasthash::FastMap::default();
20341023×
        let Ok(bytes) = fs::read(self.scratch.join("metas")) else {
2035397×
            return out;
2036
        };
2037626×
        let mut r = crate::flat::Reader::new(&bytes);
2038626×
        let mut good = 0;
20392057×
        while !r.at_end() {
20401432×
            let mut record = || -> std::result::Result<_, ObjectError> {
20411432×
                let h = r.object()?;
20421431×
                let body = r.bytes()?;
20431431×
                let check = r.fixed::<8>()?;
20441431×
                Ok((h, body, check))
20451432×
            };
20461432×
            let Ok((h, body, check)) = record() else {
20471×
                break;
2048
            };
20491431×
            let mut sum = blake3::Hasher::new();
20501431×
            sum.update(&h.0).update(body);
20511431×
            if sum.finalize().as_bytes()[..8] != check {
20520×
                break;
20531431×
            }
20541431×
            match crate::state::ChangeMeta::read_flat(&mut crate::flat::Reader::new(body)) {
20551431×
                Ok(m) => out.insert(h, m),
20560×
                Err(_) => break,
2057
            };
20581431×
            good += 32 + 4 + body.len() + 8;
2059
        }
2060
        // Cut a bad tail, so records appended later are read.
2061626×
        if good < bytes.len() {
20621×
            let _ = fs::OpenOptions::new()
20631×
                .write(true)
20641×
                .open(self.scratch.join("metas"))
20651×
                .and_then(|f| f.set_len(good as u64));
2066625×
        }
2067626×
        out
20681023×
    }
2069
2070
    /// Add a change's metadata to the cache; quietly nothing when it
2071
    /// cannot be written (it is only a cache).
20721284×
    fn remember_meta(&self, r: &ObjectRef, meta: &crate::state::ChangeMeta) {
20731284×
        let mut body = crate::flat::Writer::default();
20741284×
        meta.write_flat(&mut body);
20751284×
        let mut w = crate::flat::Writer::default();
20761284×
        w.object(r);
20771284×
        w.bytes(&body.0);
20781284×
        let mut sum = blake3::Hasher::new();
20791284×
        sum.update(&r.0).update(&body.0);
20801284×
        w.fixed(&sum.finalize().as_bytes()[..8]);
20811284×
        let _ = fs::OpenOptions::new()
20821284×
            .create(true)
20831284×
            .append(true)
20841284×
            .open(self.scratch.join("metas"))
20851284×
            .and_then(|mut f| f.write_all(&w.0));
20861284×
        if let Some(m) = self.metas.borrow_mut().as_mut() {
20871284×
            m.insert(*r, meta.clone());
20881284×
        }
20891284×
    }
2090
209159×
    fn read_text(&self, r: &ObjectRef) -> Result<(Blob, Vec<u8>)> {
209259×
        let blob = Blob::decode(&self.get_typed(r, ObjectType::Blob)?)?;
209358×
        let mut content = Vec::with_capacity(blob.length as usize);
209498×
        for c in &blob.chunks {
209598×
            content.extend(self.get_typed(c, ObjectType::Chunk)?);
2096
        }
209758×
        Ok((blob, content))
209859×
    }
2099
2100258×
    pub fn put_chunk(&self, bytes: &[u8]) -> Result<ObjectRef> {
2101258×
        self.put(ObjectType::Chunk, bytes)
2102258×
    }
2103
2104
    /// Store a chunk that packs will keep uncompressed (`nocompress`, spec §13.3).
2105123×
    pub fn put_chunk_nocompress(&self, bytes: &[u8]) -> Result<ObjectRef> {
2106123×
        let r = hash_object(ObjectType::Chunk, bytes);
2107123×
        put_new_or_existing(
2108123×
            self.backend.as_ref(),
2109123×
            &format!("nocompress/{}", r.to_hex()),
2110123×
            b"",
21110×
        )?;
2112123×
        self.put(ObjectType::Chunk, bytes)
2113123×
    }
2114
21153026×
    fn is_nocompress(&self, r: &ObjectRef) -> bool {
21163026×
        self.backend
21173026×
            .has(&format!("nocompress/{}", r.to_hex()))
21183026×
            .unwrap_or(false)
21193026×
    }
2120
2121
    /// Every chunk marked to be kept raw.
2122101×
    fn nocompress_set(&self) -> Result<std::collections::HashSet<ObjectRef>> {
2123101×
        Ok(self
2124101×
            .backend
2125101×
            .list("nocompress/")
2126101×
            .map_err(backend_err)?
2127101×
            .into_iter()
2128101×
            .filter_map(|e| ObjectRef::from_hex(e.key.trim_start_matches("nocompress/")))
2129101×
            .collect())
2130101×
    }
2131
21323276×
    pub fn put_state(&self, st: &StateObject) -> Result<ObjectRef> {
21333276×
        self.put(ObjectType::State, &st.encode())
21343276×
    }
2135
21365958×
    pub fn get_state(&self, r: &ObjectRef) -> Result<StateObject> {
21375958×
        Ok(StateObject::decode(&self.get_typed(r, ObjectType::State)?)?)
21385958×
    }
2139
2140547×
    pub fn put_attest(&self, a: &Attest) -> Result<ObjectRef> {
2141547×
        self.put(ObjectType::Attest, &a.encode())
2142547×
    }
2143
2144969×
    pub fn get_attest(&self, r: &ObjectRef) -> Result<Attest> {
2145969×
        Ok(Attest::decode(&self.get_typed(r, ObjectType::Attest)?)?)
2146969×
    }
2147
21483×
    pub fn put_receipt(&self, r: &Receipt) -> Result<ObjectRef> {
21493×
        self.put(ObjectType::Receipt, &r.encode())
21503×
    }
2151
215212×
    pub fn get_receipt(&self, r: &ObjectRef) -> Result<Receipt> {
215312×
        Ok(Receipt::decode(&self.get_typed(r, ObjectType::Receipt)?)?)
215412×
    }
2155
2156
    /// Record `value` under `key` in `index`: one write-once entry, so two
2157
    /// writers of a shared store never lose each other's.
21581458×
    fn index_add(&self, index: &str, key: &ObjectRef, value: &ObjectRef) -> Result<()> {
21593137×
        debug_assert!(ENTRY_INDEXES.iter().any(|(i, _)| *i == index));
21601458×
        let entry = format!("{}{}", fanout_prefix(index, key), value.to_hex());
21611458×
        put_new_or_existing(self.backend.as_ref(), &entry, b"")?;
21621458×
        self.journal_add(entry)
21631458×
    }
2164
2165
    /// The values recorded under `key` in `index`, with those an older
2166
    /// store kept as one list in `legacy`, sorted.
21672831×
    fn index_get(&self, index: &str, legacy: &str, key: &ObjectRef) -> Result<Vec<ObjectRef>> {
21682831×
        let mut values: Vec<ObjectRef> = self
21692831×
            .entry_keys(&fanout_prefix(index, key))?
21702831×
            .into_iter()
21712831×
            .filter_map(|k| k.rsplit('/').next().and_then(ObjectRef::from_hex))
21722831×
            .collect();
21732831×
        if !self.reads_legacy() {
217442×
            // Older lists folded into entries (spec §13.7).
21752789×
        } else if let Some(bytes) = get_optional(self.backend.as_ref(), &legacy_key(legacy, key))? {
21767×
            values.extend(
21777×
                String::from_utf8_lossy(&bytes)
21787×
                    .lines()
21797×
                    .filter_map(ObjectRef::from_hex),
21807×
            );
21812782×
        }
21822831×
        values.sort();
21832831×
        values.dedup();
21842831×
        Ok(values)
21852831×
    }
2186
2187
    /// Record that seal attestation `attest` covers `change`.
21881384×
    pub fn index_seal(&self, change: &ObjectRef, attest: &ObjectRef) -> Result<()> {
21891384×
        self.index_add("seal", change, attest)
21901384×
    }
2191
2192
    /// Seal attestations known to cover `change`.
21931545×
    pub fn seals_for(&self, change: &ObjectRef) -> Result<Vec<ObjectRef>> {
21941545×
        self.index_get("seal", "seals", change)
21951545×
    }
2196
2197
    /// Record receipt attestation `attest` for `state` (spec §14.6).
2198
    /// Keep a short-lived key's certificate (spec §14.7), with its issuer's
2199
    /// signature, indexed by the key it certifies.
220053×
    pub fn put_certificate(&self, c: &crate::cert::SignedCertificate) -> Result<ObjectRef> {
220153×
        let r = self.put(ObjectType::Certificate, &c.cert.encode())?;
220253×
        self.put_signature(&r, &c.sig)?;
220353×
        self.index_add("cert", &c.cert.key_id(), &r)?;
220453×
        Ok(r)
220553×
    }
2206
2207
    /// A kept certificate with its issuer's signature.
2208173×
    pub fn get_certificate(&self, r: &ObjectRef) -> Result<crate::cert::SignedCertificate> {
2209173×
        let cert = crate::cert::Certificate::decode(&self.get_typed(r, ObjectType::Certificate)?)?;
2210173×
        let sig = self
2211173×
            .signatures(r)?
2212173×
            .into_iter()
2213173×
            .find(|s| s.key == cert.issuer)
2214173×
            .ok_or(StoreError::NotFound(*r))?;
2215173×
        Ok(crate::cert::SignedCertificate { cert, sig })
2216173×
    }
2217
2218
    /// The certificate kept for short-lived key `key`, if any.
2219902×
    pub fn certificate_of(&self, key: &ObjectRef) -> Result<Option<ObjectRef>> {
2220902×
        Ok(self
2221902×
            .index_get("cert", "certs", key)?
2222902×
            .into_iter()
2223902×
            .find(|r| self.has(r)))
2224902×
    }
2225
2226
    /// Record `value` under `key` in index `index` (for repository code).
222713×
    pub(crate) fn index_add_pub(
222813×
        &self,
222913×
        index: &str,
223013×
        key: &ObjectRef,
223113×
        value: &ObjectRef,
223213×
    ) -> Result<()> {
223313×
        self.index_add(index, key, value)
223413×
    }
2235
2236
    /// The values recorded under `key` in index `index` (which has no
2237
    /// older one-list form: the legacy name only has to differ from it).
2238360×
    pub(crate) fn index_get_pub(&self, index: &str, key: &ObjectRef) -> Result<Vec<ObjectRef>> {
2239360×
        self.index_get(index, &format!("{index}-lists"), key)
2240360×
    }
2241
22426×
    pub fn index_receipt(&self, state: &ObjectRef, attest: &ObjectRef) -> Result<()> {
22436×
        self.index_add("receipt", state, attest)
22446×
    }
2245
2246
    /// Receipt attestations recorded for `state`.
224717×
    pub fn receipts_for(&self, state: &ObjectRef) -> Result<Vec<ObjectRef>> {
224817×
        self.index_get("receipt", "receipts", state)
224917×
    }
2250
22512×
    pub fn index_approval(&self, state: &ObjectRef, attest: &ObjectRef) -> Result<()> {
22522×
        self.index_add("approval", state, attest)
22532×
    }
2254
2255
    /// Approval attestations recorded for `state` (OxygenFlow approval
2256
    /// gates, oxygenhub#235).
22574×
    pub fn approvals_for(&self, state: &ObjectRef) -> Result<Vec<ObjectRef>> {
22584×
        self.index_get("approval", "approvals", state)
22594×
    }
2260
2261
    /// Every state with an approval, for garbage collection.
226224×
    pub fn approval_states(&self) -> Result<Vec<ObjectRef>> {
226324×
        self.indexed_states("approval", "approvals")
226424×
    }
2265
2266
    /// Every state with a receipt, for garbage collection.
226733×
    pub fn receipt_states(&self) -> Result<Vec<ObjectRef>> {
226833×
        self.indexed_states("receipt", "receipts")
226933×
    }
2270
2271
    /// The states an index (`one/<hh>/<rest>/<attest>` entries, and the
2272
    /// packed `many/<hh>/<rest>` lists) holds attestations for.
227357×
    fn indexed_states(&self, one: &str, many: &str) -> Result<Vec<ObjectRef>> {
227457×
        let mut out: Vec<ObjectRef> = Vec::new();
227557×
        for key in self.entry_keys(&format!("{one}/"))? {
2276
            // receipt/<hh>/<rest>/<attest>
22777×
            let mut parts = key.splitn(4, '/');
22787×
            let (Some(_), Some(hh), Some(rest), Some(_)) =
22797×
                (parts.next(), parts.next(), parts.next(), parts.next())
2280
            else {
22810×
                continue;
2282
            };
22837×
            out.extend(ObjectRef::from_hex(&format!("{hh}{rest}")));
2284
        }
228557×
        let legacy = if self.reads_legacy() {
228655×
            self.backend
228755×
                .list(&format!("{many}/"))
228855×
                .map_err(backend_err)?
2289
        } else {
22902×
            Vec::new()
2291
        };
229257×
        for e in legacy {
22932×
            let mut parts = e.key.splitn(3, '/');
22942×
            let (Some(_), Some(hh), Some(rest)) = (parts.next(), parts.next(), parts.next()) else {
22950×
                continue;
2296
            };
22972×
            out.extend(ObjectRef::from_hex(&format!("{hh}{rest}")));
2298
        }
229957×
        out.sort();
230057×
        out.dedup();
230157×
        Ok(out)
230257×
    }
2303
2304440×
    pub fn put_identity(&self, id: &Identity) -> Result<ObjectRef> {
2305440×
        self.put(ObjectType::Identity, &id.encode())
2306440×
    }
2307
23081140×
    pub fn get_identity(&self, r: &ObjectRef) -> Result<Identity> {
23091140×
        Ok(Identity::decode(&self.get_typed(r, ObjectType::Identity)?)?)
23101140×
    }
2311
2312
    /// Signatures stored alongside object `r` (spec §5.4), one per signing
2313
    /// key, sorted by key. Empty if none.
23143843×
    pub fn signatures(&self, r: &ObjectRef) -> Result<Vec<Signature>> {
23153843×
        let mut sigs: Vec<Signature> = Vec::new();
23163843×
        for key in self.entry_keys(&fanout_prefix("sig", r))? {
2317
            // Gone since it was listed or journalled: a collection deleted
2318
            // it with its object, or a new signature by the same signer is
2319
            // replacing it.
23202550×
            let Some(bytes) = get_optional(self.backend.as_ref(), &key)? else {
23210×
                continue;
2322
            };
23232550×
            sigs.extend(Signature::decode_list(&bytes)?);
2324
        }
2325
        // An older store's one list per object; a signer kept there and
2326
        // here counts once, as written here.
23273843×
        let legacy = if self.reads_legacy() {
23283799×
            get_optional(self.backend.as_ref(), &legacy_key("sigs", r))?
2329
        } else {
233044×
            None
2331
        };
23323843×
        if let Some(bytes) = legacy {
23334×
            for sig in Signature::decode_list(&bytes)? {
23344×
                if !sigs.iter().any(|s| s.key == sig.key) {
23353×
                    sigs.push(sig);
23363×
                }
2337
            }
23383839×
        }
23393843×
        sigs.sort_by_key(|s| s.key);
23403843×
        sigs.dedup_by_key(|s| s.key);
23413843×
        Ok(sigs)
23423843×
    }
2343
2344
    /// Attach a signature to object `r`, replacing any earlier one by the
2345
    /// same key: one entry per signer, so the latest by a key is the one
2346
    /// kept, and two signers never touch each other's.
2347873×
    pub fn put_signature(&self, r: &ObjectRef, sig: &Signature) -> Result<()> {
2348873×
        let key = format!("{}{}", fanout_prefix("sig", r), sig.key.to_hex());
2349873×
        let bytes = Signature::encode_list(std::slice::from_ref(sig));
2350873×
        if get_optional(self.backend.as_ref(), &key)?.as_deref() != Some(bytes.as_slice()) {
2351816×
            delete(self.backend.as_ref(), &key)?;
2352816×
            put_new_or_existing(self.backend.as_ref(), &key, &bytes)?;
235357×
        }
2354
        // Journalled even when the key was there: an older writer's
2355
        // entry is not in the journal or any list.
2356873×
        self.journal_add(key)
2357873×
    }
2358
2359
    // ---- the journal and entry lists (spec §13.7, #116)
2360
2361
    /// Whether lookups still read the one-list keys of older stores: until
2362
    /// a complete fold has copied them into per-object entries and the
2363
    /// pack listing shows its marker (spec §13.7).
23646731×
    fn reads_legacy(&self) -> bool {
23656731×
        !self.legacy_folded.get()
23666731×
    }
2367
2368
    /// Copy every older one-list key's contents into per-object entries,
2369
    /// keeping an entry already there, then write the marker that tells
2370
    /// readers to stop reading those keys. The keys themselves are left
2371
    /// for older readers. A list that cannot be read is passed over, and
2372
    /// said.
237310×
    fn fold_legacy(&self, progress: Progress) -> Result<()> {
237410×
        let mut copied = 0usize;
237560×
        for (index, legacy) in ENTRY_INDEXES {
237660×
            for e in self
237760×
                .backend
237860×
                .list(&format!("{legacy}/"))
237960×
                .map_err(backend_err)?
2380
            {
2381
                // <legacy>/<hh>/<rest>
23826×
                let mut parts = e.key.splitn(3, '/');
23836×
                let (Some(_), Some(hh), Some(rest)) = (parts.next(), parts.next(), parts.next())
2384
                else {
23850×
                    continue;
2386
                };
23876×
                let Some(r) = ObjectRef::from_hex(&format!("{hh}{rest}")) else {
23880×
                    continue;
2389
                };
23906×
                let Some(bytes) = get_optional(self.backend.as_ref(), &e.key)? else {
23910×
                    continue;
2392
                };
23936×
                if index == "sig" {
23942×
                    let Ok(sigs) = Signature::decode_list(&bytes) else {
23950×
                        progress(&format!("{}: not a list of signatures, passed over", e.key));
23960×
                        continue;
2397
                    };
23982×
                    for sig in sigs {
23992×
                        let key = format!("{}{}", fanout_prefix("sig", &r), sig.key.to_hex());
24002×
                        let one = Signature::encode_list(std::slice::from_ref(&sig));
24012×
                        put_new_or_existing(self.backend.as_ref(), &key, &one)?;
24022×
                        copied += 1;
2403
                    }
2404
                } else {
24054×
                    for v in String::from_utf8_lossy(&bytes)
24064×
                        .lines()
24074×
                        .filter_map(ObjectRef::from_hex)
2408
                    {
24094×
                        let key = format!("{}{}", fanout_prefix(index, &r), v.to_hex());
24104×
                        put_new_or_existing(self.backend.as_ref(), &key, b"")?;
24114×
                        copied += 1;
2412
                    }
2413
                }
2414
            }
2415
        }
241610×
        if copied > 0 {
24172×
            progress(&format!("{copied} entries copied out of older lists"));
24188×
        }
241910×
        put_new_or_existing(
242010×
            self.backend.as_ref(),
242110×
            LEGACY_FOLDED_KEY,
242210×
            format!("{}\n", now_ms()).as_bytes(),
24230×
        )?;
242410×
        self.legacy_folded.set(true);
242510×
        Ok(())
242610×
    }
2427
2428
    /// The entry keys under `prefix` (an index's `<index>/` or one
2429
    /// object's `<index>/<hh>/<rest>/`), sorted: on a folded store from
2430
    /// its lists and journal, with no request; otherwise by listing.
24316731×
    fn entry_keys(&self, prefix: &str) -> Result<Vec<String>> {
24326731×
        if let Some(keys) = self.with_entries(|set| {
243384×
            set.range(prefix.to_string()..)
24342300×
                .take_while(|k| k.starts_with(prefix))
243584×
                .cloned()
243684×
                .collect::<Vec<String>>()
243784×
        })? {
243884×
            return Ok(keys);
24396647×
        }
24406647×
        Ok(self
24416647×
            .backend
24426647×
            .list(prefix)
24436647×
            .map_err(backend_err)?
24446647×
            .into_iter()
24456647×
            .map(|e| e.key)
24466647×
            .collect())
24476731×
    }
2448
2449
    /// `f` over a folded store's entry keys, read on first use; `None`
2450
    /// when the store is not folded (or is this disk's directory), and
2451
    /// lookups list.
24526732×
    fn with_entries<T>(
24536732×
        &self,
24546732×
        f: impl FnOnce(&std::collections::BTreeSet<String>) -> T,
24556732×
    ) -> Result<Option<T>> {
24566732×
        if self.local || !self.folded.get() {
24576647×
            return Ok(None);
245885×
        }
2459
        // Read over the interval ago: list the packs again (a cloud
2460
        // client keeps that listing as long), loading what changed, and
2461
        // read the journal again, so another writer's entry is seen as
2462
        // its pack would be.
246385×
        if self
246485×
            .entries_read
246585×
            .get()
246685×
            .is_some_and(|t| t.elapsed() >= ENTRIES_FRESH)
2467
        {
24681×
            *self.entries.borrow_mut() = None;
24691×
            self.reload_if_changed()?;
24701×
            if !self.folded.get() {
24710×
                return Ok(None);
24721×
            }
247384×
        }
247485×
        if self.entries.borrow().is_none() {
247518×
            let Some(set) = self.read_entries()? else {
24760×
                return Ok(None);
2477
            };
247818×
            *self.entries.borrow_mut() = Some(set);
247967×
        }
248085×
        Ok(self.entries.borrow().as_ref().map(f))
24816732×
    }
2482
2483
    /// Every live entry list's keys and every journal line's, folded or
2484
    /// not, with what this handle has queued. A list deleted since the
2485
    /// packs were listed (retired over an hour ago) has the packs listed
2486
    /// again; `None` when one is still missing then, or the store is no
2487
    /// longer folded: lookups list.
248818×
    fn read_entries(&self) -> Result<Option<std::collections::BTreeSet<String>>> {
248919×
        'attempt: for attempt in 0..2 {
249019×
            if !self.folded.get() {
24910×
                return Ok(None);
249219×
            }
249319×
            let mut set = std::collections::BTreeSet::new();
249419×
            let lists = self.entry_lists.borrow().clone();
249519×
            self.lists_read
249619×
                .borrow_mut()
249719×
                .retain(|k, _| lists.contains(k));
249820×
            for key in &lists {
249920×
                let known = self.lists_read.borrow().get(key).cloned();
250020×
                let keys = match known {
25012×
                    Some(keys) => keys,
250218×
                    None => match get_optional(self.backend.as_ref(), key)? {
250317×
                        Some(bytes) => {
250417×
                            let keys = Arc::new(decode_entry_list(key, &bytes)?);
250517×
                            self.lists_read
250617×
                                .borrow_mut()
250717×
                                .insert(key.clone(), keys.clone());
250817×
                            keys
2509
                        }
25101×
                        None if attempt == 0 => {
25111×
                            self.reload_packs()?;
25121×
                            continue 'attempt;
2513
                        }
25140×
                        None => return Ok(None),
2515
                    },
2516
                };
251719×
                set.extend(keys.iter().cloned());
2518
            }
251918×
            set.extend(self.read_record(JOURNAL_KEY)?.0.into_keys());
252018×
            set.extend(self.journal_queue.borrow().iter().cloned());
252118×
            self.entries_read.set(Some(std::time::Instant::now()));
252218×
            return Ok(Some(set));
2523
        }
25240×
        Ok(None)
252518×
    }
2526
2527
    /// Add an entry key this handle wrote to the journal: at once, or
2528
    /// when the open batch is flushed. One the handle already knows from
2529
    /// the lists or the journal is not added again. Nothing on this
2530
    /// disk's directory.
25312331×
    fn journal_add(&self, key: String) -> Result<()> {
25322331×
        if self.local {
25331103×
            return Ok(());
25341228×
        }
25351228×
        if let Some(set) = self.entries.borrow_mut().as_mut() {
25360×
            if !set.insert(key.clone()) {
25370×
                return Ok(());
25380×
            }
25391228×
        }
25401228×
        if self.batch_depth.get() > 0 {
25411001×
            self.journal_queue.borrow_mut().push(key);
25421001×
            return Ok(());
2543227×
        }
2544227×
        self.journal_append(std::slice::from_ref(&key))
25452331×
    }
2546
2547
    /// Add what the open batch queued to the journal.
25486279×
    fn flush_journal(&self) -> Result<()> {
25496279×
        let queued = std::mem::take(&mut *self.journal_queue.borrow_mut());
25506279×
        self.journal_append(&queued)
25516279×
    }
2552
2553
    /// Add `keys` to the journal, unfolded, in one compare-and-swap
2554
    /// update, dropping lines folded over an hour ago; past
2555
    /// [`JOURNAL_FOLD_AT`] unfolded lines, fold them into a partial list.
25566506×
    fn journal_append(&self, keys: &[String]) -> Result<()> {
25576506×
        if keys.is_empty() {
25586278×
            return Ok(());
2559228×
        }
2560228×
        let now = now_ms();
2561228×
        let mut unfolded = 0;
2562324×
        self.update_record(JOURNAL_KEY, "the journal of entries", |m| {
2563324×
            expire_folded(m, now);
25641324×
            for k in keys {
25651324×
                m.entry(k.clone()).or_insert(0);
25661324×
            }
256730252×
            unfolded = m.values().filter(|t| **t == 0).count();
2568324×
        })?;
2569228×
        if unfolded > JOURNAL_FOLD_AT {
25701×
            self.fold_journal()?;
2571227×
        }
2572228×
        Ok(())
25736506×
    }
2574
2575
    /// Fold the journal's unfolded lines into a partial entry list, and
2576
    /// mark them folded. Lines added meanwhile stay unfolded.
25771×
    fn fold_journal(&self) -> Result<()> {
25781×
        let (journal, _) = self.read_record(JOURNAL_KEY)?;
25791×
        let keys: Vec<String> = journal
25801×
            .into_iter()
25811001×
            .filter(|(_, t)| *t == 0)
25821×
            .map(|(k, _)| k)
25831×
            .collect();
25841×
        if keys.is_empty() {
25850×
            return Ok(());
25861×
        }
25871×
        self.put_entry_list(PART_SUFFIX, keys.iter())?;
25881×
        self.mark_folded(&keys)
25891×
    }
2590
2591
    /// Mark these journal lines folded now, if they are not yet; drop
2592
    /// those folded over an hour ago.
25938×
    fn mark_folded(&self, keys: &[String]) -> Result<()> {
25948×
        let now = now_ms();
25958×
        self.update_record(JOURNAL_KEY, "the journal of entries", |m| {
25968×
            expire_folded(m, now);
25971214×
            for k in keys {
25981214×
                if let Some(t) = m.get_mut(k) {
25991214×
                    if *t == 0 {
26001214×
                        *t = now;
26011214×
                    }
26020×
                }
2603
            }
26048×
        })
26058×
    }
2606
2607
    /// Write a new entry list of `keys` (sorted), complete or partial by
2608
    /// `suffix`; its key.
260915×
    fn put_entry_list<'a>(
261015×
        &self,
261115×
        suffix: &str,
261215×
        keys: impl Iterator<Item = &'a String>,
261315×
    ) -> Result<String> {
261415×
        let key = format!("pack-{}{suffix}", fresh_id());
261515×
        let bytes = encode_entry_list(keys)?;
261615×
        self.backend.put_new(&key, &bytes).map_err(backend_err)?;
261715×
        Ok(key)
261815×
    }
2619
2620
    /// Fold every signature and index entry into one complete entry list
2621
    /// (spec §13.7), what a collection does last: first the older
2622
    /// one-list keys into per-object entries, once; then the journal's lines and
2623
    /// a listing of every entry prefix, written as a new list, every list
2624
    /// the packs' listing showed retired, and the journal lines read
2625
    /// marked folded. Nothing when the store is this disk's directory, or
2626
    /// when the one live list already holds exactly these entries and the
2627
    /// journal has nothing unfolded. Whether a list was written.
262838×
    pub fn fold_entries(&self, progress: Progress) -> Result<bool> {
262938×
        if self.local {
263023×
            return Ok(false);
263115×
        }
263215×
        self.reload_packs()?;
263315×
        if !self.legacy_folded.get() {
263410×
            progress("folding older lists of signatures and index entries");
263510×
            self.fold_legacy(progress)?;
26365×
        }
263715×
        let (journal, _) = self.read_record(JOURNAL_KEY)?;
263815×
        progress("folding signatures and index entries");
263915×
        let mut all: std::collections::BTreeSet<String> = journal.keys().cloned().collect();
264090×
        for (index, _) in ENTRY_INDEXES {
26411229×
            for e in self
264290×
                .backend
264390×
                .list(&format!("{index}/"))
264490×
                .map_err(backend_err)?
26451229×
            {
26461229×
                all.insert(e.key);
26471229×
            }
2648
        }
264915×
        let lists = self.entry_lists.borrow().clone();
265015×
        let unfolded: Vec<String> = journal
265115×
            .iter()
26521217×
            .filter(|(_, t)| **t == 0)
2653213×
            .map(|(k, _)| k.clone())
265415×
            .collect();
265515×
        if unfolded.is_empty() && lists.len() == 1 && lists[0].ends_with(LIST_SUFFIX) {
26563×
            if let Some(bytes) = get_optional(self.backend.as_ref(), &lists[0])? {
26573×
                let held: std::collections::BTreeSet<String> =
26583×
                    decode_entry_list(&lists[0], &bytes)?.into_iter().collect();
26593×
                if held == all {
26601×
                    return Ok(false);
26612×
                }
26620×
            }
266312×
        }
266414×
        self.put_entry_list(LIST_SUFFIX, all.iter())?;
266514×
        self.retire(&lists)?;
266614×
        if !unfolded.is_empty() {
26677×
            self.mark_folded(&unfolded)?;
26687×
        }
266914×
        *self.entries.borrow_mut() = None;
267014×
        self.reload_packs()?;
267114×
        Ok(true)
267238×
    }
2673
2674
    /// Finish packs a writer left unfinished: a batch writes its objects to
2675
    /// `.tmp-ID.pack` and only indexes it at the end, so a process that
2676
    /// stopped part way (an import killed after hours) leaves every object
2677
    /// in a pack nothing can read. Each such pack untouched for `min_age`
2678
    /// is read through: every whole entry checked against its reference,
2679
    /// a torn last one cut off, then indexed as a pack of the store.
2680
    /// Returns the packs and objects adopted.
26812×
    pub fn adopt_unfinished_packs(
26822×
        &self,
26832×
        min_age: std::time::Duration,
26842×
        progress: &dyn Fn(&str),
26852×
    ) -> Result<(usize, usize)> {
2686
        use std::io::{BufReader, Read};
26872×
        let mut packs = 0;
26882×
        let mut objects = 0;
26892×
        let Ok(dir) = fs::read_dir(&self.scratch) else {
26900×
            return Ok((0, 0));
2691
        };
26924×
        for entry in dir.flatten() {
26934×
            let name = entry.file_name().to_string_lossy().to_string();
26944×
            let Some(id) = name
26954×
                .strip_prefix(".tmp-")
26964×
                .and_then(|n| n.strip_suffix(".pack"))
2697
            else {
26982×
                continue;
2699
            };
27002×
            let path = entry.path();
27012×
            let meta = fs::metadata(&path)?;
27022×
            let age = meta
27032×
                .modified()
27042×
                .ok()
27052×
                .and_then(|m| m.elapsed().ok())
27062×
                .unwrap_or_default();
27072×
            if age < min_age {
27081×
                progress(&format!(
27091×
                    "{name}: written to {}s ago, left alone",
27101×
                    age.as_secs()
27111×
                ));
27121×
                continue;
27131×
            }
27141×
            let mut f = BufReader::with_capacity(1 << 20, fs::File::open(&path)?);
27151×
            let mut magic = [0u8; 8];
27161×
            if f.read_exact(&mut magic).is_err() || &magic != PACK_MAGIC {
27170×
                progress(&format!("{name}: not a pack, left alone"));
27180×
                continue;
27191×
            }
27201×
            let mut offset = PACK_MAGIC.len() as u64;
27211×
            let mut index: Vec<(ObjectRef, PackEntry)> = Vec::new();
2722
            loop {
2723301×
                let mut head = [0u8; 42];
2724301×
                if f.read_exact(&mut head).is_err() {
27250×
                    break;
2726301×
                }
2727301×
                let Some(ty) = ObjectType::ALL.get(head[0] as usize).copied() else {
27280×
                    break;
2729
                };
2730301×
                let flags = head[1];
2731301×
                let stored_len = u32::from_le_bytes(head[2..6].try_into().unwrap());
2732301×
                let raw_len = u32::from_le_bytes(head[6..10].try_into().unwrap());
2733301×
                let r = ObjectRef(head[10..42].try_into().unwrap());
2734301×
                let mut stored = vec![0u8; stored_len as usize];
2735301×
                if flags & FLAG_DICT != 0 || f.read_exact(&mut stored).is_err() {
27361×
                    break;
2737300×
                }
2738300×
                let e = PackEntry {
2739300×
                    offset: offset + 42,
2740300×
                    stored_len,
2741300×
                    raw_len,
2742300×
                    ty,
2743300×
                    flags,
2744300×
                };
2745300×
                if decode_entry(&r, e, stored, None).is_err() {
27460×
                    break;
2747300×
                }
2748300×
                index.push((r, e));
2749300×
                offset += 42 + stored_len as u64;
2750300×
                if index.len().is_multiple_of(200_000) {
27510×
                    progress(&format!(
27520×
                        "{name}: {} objects checked, {} MiB",
27530×
                        index.len(),
27540×
                        offset >> 20
27550×
                    ));
2756300×
                }
2757
            }
27581×
            if index.is_empty() {
27590×
                progress(&format!("{name}: no whole object, left alone"));
27600×
                continue;
27611×
            }
27621×
            if offset < meta.len() {
27631×
                progress(&format!(
27641×
                    "{name}: {} torn bytes at the end cut off",
27651×
                    meta.len() - offset
27661×
                ));
27671×
                fs::OpenOptions::new()
27681×
                    .write(true)
27691×
                    .open(&path)?
27701×
                    .set_len(offset)?;
27710×
            }
27721×
            index.sort_by_key(|(r, _)| *r);
27731×
            index.dedup_by_key(|(r, _)| *r);
27741×
            objects += index.len();
27751×
            packs += 1;
27761×
            progress(&format!("{name}: {} objects indexed", index.len()));
27771×
            self.finish_pack_files(id, &path, &index)?;
2778
        }
27792×
        Ok((packs, objects))
27802×
    }
2781
2782
    /// Every object reference in the store, loose and packed (slow; for gc and tests).
278389×
    pub fn list(&self) -> io::Result<Vec<ObjectRef>> {
278489×
        let mut out: Vec<ObjectRef> = self
278589×
            .loose_objects()
278689×
            .map_err(|e| io::Error::other(e.to_string()))?
278789×
            .into_iter()
278889×
            .map(|(r, _)| r)
278989×
            .collect();
279089×
        for p in self.packs.borrow().iter() {
279110×
            out.extend(p.refs.iter().copied());
279210×
        }
279389×
        out.sort();
279489×
        out.dedup();
279589×
        Ok(out)
279689×
    }
2797
}
2798
2799
/// Whether a read failed in a way a newer list of packs may cure: the
2800
/// object was not found, or its pack's dictionary is gone.
2801346×
fn is_miss(e: &StoreError) -> bool {
2802346×
    match e {
2803218×
        StoreError::NotFound(_) => true,
2804128×
        StoreError::Corrupt(_, why) => why == NO_DICT,
28050×
        _ => false,
2806
    }
2807346×
}
2808
2809
/// A loose object's bytes, verified against `r`.
281055109×
fn parse_loose(data: &[u8], r: &ObjectRef) -> Result<Object> {
281155109×
    let nul = data
281255109×
        .iter()
2813785616×
        .position(|&b| b == 0)
281455109×
        .ok_or_else(|| StoreError::Corrupt(*r, "no header".into()))?;
281555109×
    let head = std::str::from_utf8(&data[..nul])
281655109×
        .map_err(|_| StoreError::Corrupt(*r, "header not UTF-8".into()))?;
281755109×
    let mut parts = head.split(' ');
281855109×
    let (Some("oxy1"), Some(ty), Some(len), None) =
281955109×
        (parts.next(), parts.next(), parts.next(), parts.next())
2820
    else {
28210×
        return Err(StoreError::Corrupt(*r, format!("bad header {head:?}")));
2822
    };
282355109×
    let ty = ObjectType::from_name(ty)
282455109×
        .ok_or_else(|| StoreError::Corrupt(*r, format!("unknown type {ty}")))?;
282555109×
    let len: usize = len
282655109×
        .parse()
282755109×
        .map_err(|_| StoreError::Corrupt(*r, "bad length".into()))?;
282855109×
    let body = &data[nul + 1..];
282955109×
    if body.len() != len {
28300×
        return Err(StoreError::Corrupt(
28310×
            *r,
28320×
            format!("length {} != header {len}", body.len()),
28330×
        ));
283455109×
    }
283555109×
    if hash_object(ty, body) != *r {
28364×
        return Err(StoreError::Corrupt(*r, "hash mismatch".into()));
283755105×
    }
283855105×
    Ok((ty, body.to_vec()))
283955109×
}
2840
2841
/// Whether the objects are files that a test may tamper with: not under
2842
/// `OXY_STORE_BACKEND=memory`.
2843
#[cfg(test)]
28446×
fn files_on_disk() -> bool {
28456×
    std::env::var_os("OXY_STORE_BACKEND").is_none()
28466×
}
2847
2848
#[cfg(test)]
2849
mod tests {
2850
    use super::*;
2851
2852
    /// The store under test twice: in a directory, and on a store in
2853
    /// memory with a scratch directory, as over a bucket.
28546×
    fn each_store(dir: &Path, f: impl Fn(&str, Store)) {
28556×
        f("directory", Store::open(dir.join("objects")).unwrap());
28566×
        f(
28576×
            "memory",
28586×
            Store::open_on(Arc::new(MemoryStore::new()), dir.join("scratch")).unwrap(),
28596×
        );
28606×
    }
2861
2862
    #[test]
28631×
    fn put_get_verify() {
28641×
        if !super::files_on_disk() {
28650×
            return;
28661×
        }
28671×
        let dir = tempfile::tempdir().unwrap();
28681×
        let s = Store::open(dir.path().join("objects")).unwrap();
28691×
        let r = s.put(ObjectType::Chunk, b"hello").unwrap();
28701×
        assert!(s.has(&r));
28711×
        assert_eq!(s.get(&r).unwrap(), (ObjectType::Chunk, b"hello".to_vec()));
28721×
        assert_eq!(s.put(ObjectType::Chunk, b"hello").unwrap(), r);
28731×
        assert_eq!(s.list().unwrap(), vec![r]);
28741×
        assert!(matches!(
28751×
            s.get_typed(&r, ObjectType::Blob),
2876
            Err(StoreError::WrongType(..))
2877
        ));
28781×
        assert!(matches!(
28791×
            s.get(&ObjectRef([9; 32])),
2880
            Err(StoreError::NotFound(_))
2881
        ));
2882
2883
        // Tamper: flip a byte, read must fail.
28841×
        let path = s.path_for(&r);
28851×
        let mut data = fs::read(&path).unwrap();
28861×
        let last = data.len() - 1;
28871×
        data[last] ^= 1;
28881×
        fs::write(&path, data).unwrap();
28891×
        assert!(matches!(s.get(&r), Err(StoreError::Corrupt(..))));
28901×
    }
2891
2892
    #[test]
28931×
    fn packing_keeps_every_object_readable() {
28941×
        if !super::files_on_disk() {
28950×
            return;
28961×
        }
28971×
        let dir = tempfile::tempdir().unwrap();
28981×
        let s = Store::open(dir.path()).unwrap();
28991×
        let mut refs = Vec::new();
2900300×
        for i in 0..300u32 {
2901300×
            let body = format!("object {i} ").repeat(1 + (i % 7) as usize);
2902300×
            refs.push((s.put(ObjectType::Chunk, body.as_bytes()).unwrap(), body));
2903300×
        }
29041×
        let c = crate::objects::tests::sample_change();
29051×
        let cr = s.put_change(&c).unwrap();
29061×
        assert_eq!(s.loose_count().unwrap(), 301);
29071×
        let report = s.pack_loose().unwrap();
29081×
        assert_eq!(report.objects, 301);
29091×
        assert!(report.packed_bytes < report.raw_bytes, "{report:?}");
29101×
        assert_eq!(s.loose_count().unwrap(), 0);
29111×
        assert_eq!(s.pack_count(), 1);
2912300×
        for (r, body) in &refs {
2913300×
            assert!(s.has(r));
2914300×
            assert_eq!(
2915300×
                s.get(r).unwrap(),
2916300×
                (ObjectType::Chunk, body.as_bytes().to_vec())
2917
            );
2918
        }
29191×
        assert_eq!(s.get_change(&cr).unwrap(), c);
29201×
        assert_eq!(s.list().unwrap().len(), 301);
2921
        // Re-putting a packed object is a no-op; new objects go loose and a
2922
        // second pack coexists with the first.
29231×
        assert_eq!(
29241×
            s.put(ObjectType::Chunk, refs[0].1.as_bytes()).unwrap(),
29251×
            refs[0].0
2926
        );
29271×
        assert_eq!(s.loose_count().unwrap(), 0);
29281×
        let extra = s.put(ObjectType::Chunk, b"after packing").unwrap();
29291×
        s.pack_loose().unwrap();
29301×
        assert_eq!(s.pack_count(), 2);
29311×
        assert_eq!(s.get(&extra).unwrap().1, b"after packing");
2932
        // A reopened store sees the packs.
29331×
        let s2 = Store::open(dir.path()).unwrap();
29341×
        assert_eq!(s2.pack_count(), 2);
29351×
        assert_eq!(s2.get(&refs[5].0).unwrap().1, refs[5].1.as_bytes());
29361×
        assert!(matches!(
29371×
            s2.get(&ObjectRef([9; 32])),
2938
            Err(StoreError::NotFound(_))
2939
        ));
2940
        // Corrupting an object's body inside the pack is detected on read.
29411×
        let target = refs[5].0;
29421×
        let (pack_path, offset) = {
29431×
            let packs = s2.packs.borrow();
29441×
            let (i, e) = s2.find_packed(&target).unwrap();
29451×
            (dir.path().join(&packs[i].key), e.offset as usize)
29461×
        };
29471×
        let mut data = fs::read(&pack_path).unwrap();
29481×
        data[offset] ^= 0xff;
29491×
        fs::write(&pack_path, data).unwrap();
29501×
        let s3 = Store::open(dir.path()).unwrap();
29511×
        assert!(matches!(s3.get(&target), Err(StoreError::Corrupt(..))));
29521×
        assert_eq!(
29531×
            s3.get(&refs[6].0).unwrap().1,
29541×
            refs[6].1.as_bytes(),
2955
            "other objects are unaffected"
2956
        );
29571×
    }
2958
2959
    #[test]
29601×
    fn a_full_repack_trains_a_dictionary_and_reads_back() {
29611×
        if !super::files_on_disk() {
29620×
            return;
29631×
        }
29641×
        let dir = tempfile::tempdir().unwrap();
29651×
        let s = Store::open(dir.path()).unwrap();
2966
        // Many small, alike files: what a dictionary is for.
29671×
        let bodies: Vec<Vec<u8>> = (0..1500)
29681500×
            .map(|i| {
29691500×
                format!(
2970
                    "// SPDX-License-Identifier: GPL-2.0\n#include <linux/module.h>\n\
2971
                     static int probe_{i}(struct device *dev)\n{{\n\treturn {i};\n}}\n"
2972
                )
29731500×
                .into_bytes()
29741500×
            })
29751×
            .collect();
29761×
        let refs: Vec<ObjectRef> = bodies
29771×
            .iter()
29781500×
            .map(|b| s.put(ObjectType::Chunk, b).unwrap())
29791×
            .collect();
29801×
        s.repack_all(REPACK_ZSTD_LEVEL).unwrap();
29811×
        let dicts = fs::read_dir(dir.path())
29821×
            .unwrap()
29831×
            .flatten()
29845×
            .filter(|e| e.path().extension().is_some_and(|x| x == "dict"))
29851×
            .count();
29861×
        assert_eq!(dicts, 1, "the pack has a dictionary");
29871×
        let with_dict = refs
29881×
            .iter()
29891500×
            .filter(|r| s.find_packed(r).unwrap().1.flags & FLAG_DICT != 0)
29901×
            .count();
29911×
        assert!(with_dict > 1000, "most use it: {with_dict}");
29921×
        let again = Store::open(dir.path()).unwrap();
29931500×
        for (r, b) in refs.iter().zip(&bodies) {
29941500×
            assert_eq!(&again.get(r).unwrap().1, b);
2995
        }
29964×
        let packs = |dir: &Path| {
29974×
            let mut v: Vec<String> = fs::read_dir(dir)
29984×
                .unwrap()
29994×
                .flatten()
300022×
                .map(|e| e.file_name().to_string_lossy().into_owned())
300122×
                .filter(|n| n.starts_with("pack-"))
30024×
                .collect();
30034×
            v.sort();
30044×
            v
30054×
        };
3006
        // With nothing new, a repack leaves that pack as it is: it would
3007
        // only write the same again.
30081×
        let before = packs(dir.path());
30091×
        let (report, merged) = again.repack_all(REPACK_ZSTD_LEVEL).unwrap();
30101×
        assert_eq!((report.objects, merged), (0, 0));
30111×
        assert_eq!(packs(dir.path()), before);
3012
        // One new object: a repack replaces the pack and its dictionary,
3013
        // retiring the old ones, which a collection an hour on deletes.
30141×
        let extra = again.put(ObjectType::Chunk, b"one more\n").unwrap();
30151×
        again.repack_all(REPACK_ZSTD_LEVEL).unwrap();
30161×
        let files = packs(dir.path());
30171×
        assert_eq!(
30181×
            files.len(),
3019
            7,
3020
            "two of pack, index and dictionary, and the retired record: {files:?}"
3021
        );
30221×
        assert_eq!(again.pack_count(), 1, "the old pack is not read");
30231×
        assert_eq!(&again.get(&refs[7]).unwrap().1, &bodies[7]);
30241×
        assert_eq!(again.get(&extra).unwrap().1, b"one more\n");
30251×
        again
30261×
            .expire_retired(std::time::Duration::ZERO, &mut silent())
30271×
            .unwrap();
30281×
        let files = packs(dir.path());
30291×
        assert_eq!(files.len(), 4, "one pack, index, dictionary: {files:?}");
30301×
        assert!(before
30311×
            .iter()
30324×
            .filter(|f| *f != RETIRED_KEY)
30333×
            .all(|f| !files.contains(f)));
30341×
        assert_eq!(&again.get(&refs[8]).unwrap().1, &bodies[8]);
30351×
    }
3036
3037
    #[test]
30381×
    fn nocompress_chunks_pack_raw() {
30391×
        let dir = tempfile::tempdir().unwrap();
30401×
        let s = Store::open(dir.path()).unwrap();
30411×
        let body = vec![b'a'; 10_000];
30421×
        let raw = s.put_chunk_nocompress(&body).unwrap();
30431×
        let compressible = s.put(ObjectType::Chunk, &vec![b'b'; 10_000]).unwrap();
30441×
        s.pack_loose().unwrap();
30451×
        let packs = s.packs.borrow();
30461×
        let (i, e) = s.find_packed(&raw).unwrap();
30471×
        assert_eq!((e.flags, e.stored_len as usize), (0, 10_000));
30481×
        let (_, c) = s.find_packed(&compressible).unwrap();
30491×
        assert_eq!(c.flags, FLAG_ZSTD);
30501×
        assert!(c.stored_len < 10_000);
30511×
        assert_eq!(packs[i].fetch(s.backend.as_ref(), &raw, e).unwrap().1, body);
30521×
    }
3053
3054
    #[test]
30551×
    fn many_packs_merge_into_one() {
30561×
        let dir = tempfile::tempdir().unwrap();
30572×
        each_store(dir.path(), |which, s| {
30582×
            let mut refs = Vec::new();
305936×
            for i in 0..(MAX_PACKS + 2) {
306036×
                refs.push(
306136×
                    s.put(ObjectType::Chunk, format!("pack {i}").as_bytes())
306236×
                        .unwrap(),
306336×
                );
306436×
                s.pack_loose().unwrap();
306536×
            }
30662×
            assert_eq!(s.pack_count(), MAX_PACKS + 2, "{which}");
30672×
            assert_eq!(s.merge_packs().unwrap(), MAX_PACKS + 2);
30682×
            assert_eq!(s.pack_count(), 1);
30692×
            assert_eq!(s.merge_packs().unwrap(), 0);
307036×
            for (i, r) in refs.iter().enumerate() {
307136×
                assert_eq!(s.get(r).unwrap().1, format!("pack {i}").as_bytes());
3072
            }
30732×
            let again = Store::open_on(s.backend().clone(), s.dir()).unwrap();
30742×
            assert_eq!(again.pack_count(), 1, "{which}: reopened");
30752×
        });
30761×
    }
3077
3078
    #[test]
30791×
    fn prune_removes_unreachable_loose_and_packed_objects() {
3080
        use std::collections::HashSet;
30811×
        let dir = tempfile::tempdir().unwrap();
30822×
        each_store(dir.path(), |which, s| {
30832×
            let keep_a = s.put(ObjectType::Chunk, b"keep a").unwrap();
30842×
            let drop_a = s.put(ObjectType::Chunk, b"drop a").unwrap();
30852×
            s.pack_loose().unwrap();
30862×
            let keep_b = s.put(ObjectType::Chunk, b"keep b").unwrap();
30872×
            let drop_b = s.put(ObjectType::Chunk, b"drop b").unwrap();
30882×
            let sk = crate::identity::SecretKey {
30892×
                ed_seed: [1; 32],
30902×
                ml_seed: [2; 32],
30912×
            };
30922×
            let id = sk.identity("a", "b", 0);
30932×
            let key = s.put_identity(&id).unwrap();
30942×
            s.put_signature(
30952×
                &drop_b,
30962×
                &sk.sign(crate::identity::Domain::Change, &key, &drop_b),
3097
            )
30982×
            .unwrap();
30992×
            s.index_seal(&drop_a, &ObjectRef([3; 32])).unwrap();
31002×
            let keep: HashSet<ObjectRef> = [keep_a, keep_b, key].into_iter().collect();
3101
            // Everything is too young under a grace period.
31022×
            let r = s
31032×
                .prune(&keep, std::time::Duration::from_secs(3600))
31042×
                .unwrap();
31052×
            assert_eq!((r.objects, r.deferred), (0, 2), "{which}");
31062×
            let r = s.prune(&keep, std::time::Duration::ZERO).unwrap();
31072×
            assert_eq!(r.objects, 2, "{which}: {r:?}");
31082×
            assert!(s.has(&keep_a) && s.has(&keep_b) && s.has(&key));
31092×
            assert!(!s.has(&drop_a) && !s.has(&drop_b));
31102×
            assert!(s.signatures(&drop_b).unwrap().is_empty());
31112×
            assert!(s.seals_for(&drop_a).unwrap().is_empty());
31122×
            assert_eq!(s.pack_count(), 1);
31132×
            assert_eq!(s.get(&keep_a).unwrap().1, b"keep a");
31142×
            assert_eq!(s.list().unwrap().len(), 3);
31152×
        });
31161×
    }
3117
3118
    #[test]
31191×
    fn signatures_are_stored_beside_objects() {
3120
        use crate::identity::{Domain, SecretKey};
31211×
        let dir = tempfile::tempdir().unwrap();
31222×
        each_store(dir.path(), |which, s| {
31232×
            let r = s.put(ObjectType::Chunk, b"x").unwrap();
31242×
            assert!(s.signatures(&r).unwrap().is_empty());
31252×
            let sk = SecretKey {
31262×
                ed_seed: [1; 32],
31272×
                ml_seed: [2; 32],
31282×
            };
31292×
            let id = sk.identity("a", "b", 0);
31302×
            let key = s.put_identity(&id).unwrap();
31312×
            assert_eq!(s.get_identity(&key).unwrap(), id);
31322×
            let sig = sk.sign(Domain::Change, &key, &r);
31332×
            s.put_signature(&r, &sig).unwrap();
31342×
            s.put_signature(&r, &sig).unwrap();
31352×
            assert_eq!(s.signatures(&r).unwrap(), vec![sig.clone()], "{which}");
3136
            // A second signer is kept beside the first, sorted by key.
31372×
            let sk2 = SecretKey {
31382×
                ed_seed: [3; 32],
31392×
                ml_seed: [4; 32],
31402×
            };
31412×
            let key2 = s.put_identity(&sk2.identity("c", "d", 0)).unwrap();
31422×
            let sig2 = sk2.sign(Domain::Change, &key2, &r);
31432×
            s.put_signature(&r, &sig2).unwrap();
31442×
            let mut both = vec![sig.clone(), sig2.clone()];
31452×
            both.sort_by_key(|s| s.key);
31462×
            assert_eq!(s.signatures(&r).unwrap(), both);
3147
            // The object itself is untouched by its signatures.
31482×
            assert_eq!(s.get(&r).unwrap().1, b"x");
3149
            // Seal and receipt entries, and the states with receipts.
31502×
            s.index_seal(&r, &ObjectRef([5; 32])).unwrap();
31512×
            s.index_seal(&r, &ObjectRef([4; 32])).unwrap();
31522×
            s.index_seal(&r, &ObjectRef([5; 32])).unwrap();
31532×
            assert_eq!(
31542×
                s.seals_for(&r).unwrap(),
31552×
                vec![ObjectRef([4; 32]), ObjectRef([5; 32])]
3156
            );
31572×
            let st = ObjectRef([6; 32]);
31582×
            s.index_receipt(&st, &ObjectRef([7; 32])).unwrap();
31592×
            assert_eq!(s.receipts_for(&st).unwrap(), vec![ObjectRef([7; 32])]);
31602×
            assert_eq!(s.receipt_states().unwrap(), vec![st]);
31612×
        });
31621×
    }
3163
3164
    /// A store written before signatures, seals and receipts were one
3165
    /// entry each (`sigs/`, `seals/`, `receipts/` lists) still reads, and
3166
    /// what is written now sits beside what was there.
3167
    #[test]
31681×
    fn an_older_store_s_sidecar_lists_are_still_read() {
31691×
        if !super::files_on_disk() {
31700×
            return;
31711×
        }
3172
        use crate::identity::{Domain, SecretKey};
31731×
        let dir = tempfile::tempdir().unwrap();
31741×
        let s = Store::open(dir.path()).unwrap();
31751×
        let r = s.put(ObjectType::Chunk, b"old").unwrap();
31761×
        let h = r.to_hex();
31771×
        let sk = SecretKey {
31781×
            ed_seed: [1; 32],
31791×
            ml_seed: [2; 32],
31801×
        };
31811×
        let key = s.put_identity(&sk.identity("a", "b", 0)).unwrap();
31821×
        let sig = sk.sign(Domain::Change, &key, &r);
31831×
        let legacy = dir.path().join("sigs").join(&h[..2]);
31841×
        fs::create_dir_all(&legacy).unwrap();
31851×
        fs::write(
31861×
            legacy.join(&h[2..]),
31871×
            Signature::encode_list(std::slice::from_ref(&sig)),
3188
        )
31891×
        .unwrap();
31901×
        let seals = dir.path().join("seals").join(&h[..2]);
31911×
        fs::create_dir_all(&seals).unwrap();
31921×
        fs::write(
31931×
            seals.join(&h[2..]),
31941×
            format!("{}\n", ObjectRef([9; 32]).to_hex()),
3195
        )
31961×
        .unwrap();
31971×
        let receipts = dir.path().join("receipts").join(&h[..2]);
31981×
        fs::create_dir_all(&receipts).unwrap();
31991×
        fs::write(
32001×
            receipts.join(&h[2..]),
32011×
            format!("{}\n", ObjectRef([8; 32]).to_hex()),
3202
        )
32031×
        .unwrap();
3204
32051×
        assert_eq!(s.signatures(&r).unwrap(), vec![sig.clone()]);
32061×
        assert_eq!(s.seals_for(&r).unwrap(), vec![ObjectRef([9; 32])]);
32071×
        assert_eq!(s.receipts_for(&r).unwrap(), vec![ObjectRef([8; 32])]);
32081×
        assert_eq!(s.receipt_states().unwrap(), vec![r]);
3209
        // New entries join the old ones; the same signer is not doubled.
32101×
        s.put_signature(&r, &sig).unwrap();
32111×
        s.index_seal(&r, &ObjectRef([1; 32])).unwrap();
32121×
        assert_eq!(s.signatures(&r).unwrap().len(), 1);
32131×
        assert_eq!(
32141×
            s.seals_for(&r).unwrap(),
32151×
            vec![ObjectRef([1; 32]), ObjectRef([9; 32])]
3216
        );
3217
        // Pruning the object takes both layouts with it.
32181×
        let r2 = s.put(ObjectType::Chunk, b"kept").unwrap();
32191×
        let keep = [r2, key].into_iter().collect();
32201×
        s.prune(&keep, std::time::Duration::ZERO).unwrap();
32211×
        assert!(s.signatures(&r).unwrap().is_empty());
32221×
        assert!(s.seals_for(&r).unwrap().is_empty());
32231×
        assert!(s.receipt_states().unwrap().is_empty());
32241×
    }
3225
3226
    #[test]
32271×
    fn change_round_trip_through_store() {
32281×
        let dir = tempfile::tempdir().unwrap();
32292×
        each_store(dir.path(), |which, s| {
32302×
            let c = crate::objects::tests::sample_change();
32312×
            let r = s.put_change(&c).unwrap();
32322×
            assert_eq!(r, c.hash(), "{which}");
32332×
            assert_eq!(s.get_change(&r).unwrap(), c);
32342×
            assert_eq!(s.get_change_meta(&r).unwrap().message, c.message);
32352×
        });
32361×
    }
3237
32388057203×
    fn xorshift(state: &mut u64) -> u64 {
32398057203×
        *state ^= *state << 13;
32408057203×
        *state ^= *state >> 7;
32418057203×
        *state ^= *state << 17;
32428057203×
        *state
32438057203×
    }
3244
3245
    /// Random corruption of pack and index files never panics: reads fail
3246
    /// with an error or the store refuses to open, and untouched objects
3247
    /// stay readable when the index survives.
3248
    #[test]
32491×
    fn corrupted_packs_and_indexes_fail_cleanly() {
32501×
        if !super::files_on_disk() {
32510×
            return;
32521×
        }
32531×
        let dir = tempfile::tempdir().unwrap();
32541×
        let objects = dir.path().join("objects");
32551×
        let s = Store::open(&objects).unwrap();
32561×
        let mut refs = Vec::new();
32571×
        let mut seed = 0x9e3779b97f4a7c15u64;
325840×
        for i in 0..40u32 {
325940×
            let len = (xorshift(&mut seed) % 5000) as usize;
326040×
            let mut body = vec![0u8; len];
3261104686×
            for b in body.iter_mut() {
3262104686×
                *b = if i % 2 == 0 {
326356338×
                    (xorshift(&mut seed) & 0xff) as u8
3264
                } else {
326548348×
                    b'a' + (i % 26) as u8
3266
                };
3267
            }
326840×
            refs.push(s.put(ObjectType::Chunk, &body).unwrap());
3269
        }
32701×
        s.pack_loose().unwrap();
32711×
        let pack = fs::read_dir(&objects)
32721×
            .unwrap()
32733×
            .filter_map(|e| e.ok().map(|e| e.path()))
32743×
            .find(|p| p.extension().is_some_and(|x| x == "pack"))
32751×
            .unwrap();
32761×
        let idx = pack.with_extension("idx");
32771×
        let pristine_pack = fs::read(&pack).unwrap();
32781×
        let pristine_idx = fs::read(&idx).unwrap();
3279
3280300×
        for round in 0..300u32 {
3281300×
            let (path, pristine) = if round % 2 == 0 {
3282150×
                (&pack, &pristine_pack)
3283
            } else {
3284150×
                (&idx, &pristine_idx)
3285
            };
3286300×
            let mut data = pristine.clone();
3287300×
            match xorshift(&mut seed) % 4 {
328881×
                0 => {
328981×
                    let at = (xorshift(&mut seed) as usize) % data.len();
329081×
                    data[at] ^= 1 << (xorshift(&mut seed) % 8);
329181×
                }
329273×
                1 => {
329373×
                    let at = (xorshift(&mut seed) as usize) % data.len();
329473×
                    data[at] = (xorshift(&mut seed) & 0xff) as u8;
329573×
                }
329675×
                2 => {
329775×
                    let keep = (xorshift(&mut seed) as usize) % data.len();
329875×
                    data.truncate(keep);
329975×
                }
3300
                _ => {
330171×
                    let at = (xorshift(&mut seed) as usize) % data.len();
330271×
                    let n = ((xorshift(&mut seed) % 8) + 1) as usize;
3303324×
                    for k in 0..n.min(data.len() - at) {
3304324×
                        data[at + k] = 0xff;
3305324×
                    }
3306
                }
3307
            }
3308300×
            fs::write(path, &data).unwrap();
3309
            // Opening may refuse a broken index; a store that opens must
3310
            // either return each object intact or report it corrupt/missing.
3311300×
            if let Ok(s2) = Store::open(&objects) {
33127640×
                for r in &refs {
33137640×
                    match s2.get(r) {
33147517×
                        Ok((ty, body)) => assert_eq!(
33157517×
                            hash_object(ty, &body),
3316
                            *r,
3317
                            "a returned object always matches its reference"
3318
                        ),
3319
                        Err(StoreError::Corrupt(..))
3320
                        | Err(StoreError::NotFound(_))
3321123×
                        | Err(StoreError::Io(_)) => {}
33220×
                        Err(e) => panic!("unexpected error {e}"),
3323
                    }
3324
                }
3325109×
            }
3326300×
            fs::write(path, pristine).unwrap();
3327
        }
3328
        // Restored files read back in full.
33291×
        let s3 = Store::open(&objects).unwrap();
333040×
        for r in &refs {
333140×
            let (ty, body) = s3.get(r).unwrap();
333240×
            assert_eq!(hash_object(ty, &body), *r);
3333
        }
33341×
    }
3335
3336
    /// A small batch becomes loose objects; a large one becomes a single
3337
    /// pack. Objects are readable while pending, and batches nest.
3338
    #[test]
33391×
    fn batches_write_loose_or_one_pack() {
33401×
        let dir = tempfile::tempdir().unwrap();
33412×
        each_store(dir.path(), |which, s| {
33422×
            let outer = s.begin_batch().unwrap();
33432×
            let a = s.put(ObjectType::Chunk, b"small a").unwrap();
3344
            {
33452×
                let inner = s.begin_batch().unwrap();
33462×
                let b = s.put(ObjectType::Chunk, b"small b").unwrap();
33472×
                assert!(s.has(&b));
33482×
                assert_eq!(s.get(&b).unwrap().1, b"small b");
33492×
                inner.finish().unwrap();
33502×
                assert_eq!(
33512×
                    s.loose_count().unwrap(),
3352
                    0,
3353
                    "{which}: the inner batch defers to the outer one"
3354
                );
3355
            }
33562×
            assert_eq!(s.get(&a).unwrap().1, b"small a");
33572×
            assert_eq!(
33582×
                s.put(ObjectType::Chunk, b"small a").unwrap(),
3359
                a,
3360
                "idempotent while pending"
3361
            );
33622×
            outer.finish().unwrap();
33632×
            assert_eq!(s.loose_count().unwrap(), 2);
33642×
            assert_eq!(s.pack_count(), 0);
33652×
            assert_eq!(s.get(&a).unwrap().1, b"small a");
3366
3367
            // Large: spills into one pack, readable during and after.
33682×
            let big = s.begin_batch().unwrap();
33692×
            let mut refs = Vec::new();
33702×
            let mut seed = 7u64;
3371600×
            for i in 0..300u32 {
3372600×
                let mut body = vec![b'z'; 40_000];
3373600×
                if i % 3 == 0 {
33748000000×
                    for b in body.iter_mut() {
33758000000×
                        *b = (xorshift(&mut seed) & 0xff) as u8;
33768000000×
                    }
3377400×
                }
3378600×
                body[..4].copy_from_slice(&i.to_le_bytes());
3379600×
                let r = if i % 5 == 0 {
3380120×
                    s.put_chunk_nocompress(&body).unwrap()
3381
                } else {
3382480×
                    s.put(ObjectType::Chunk, &body).unwrap()
3383
                };
3384600×
                assert_eq!(
3385600×
                    s.get(&r).unwrap().1,
3386
                    body,
3387
                    "readable while pending (object {i})"
3388
                );
3389600×
                refs.push((r, body));
3390
            }
33912×
            assert!(s.has(&refs[0].0));
33922×
            big.finish().unwrap();
33932×
            assert_eq!(s.pack_count(), 1);
33942×
            assert_eq!(
33952×
                s.loose_count().unwrap(),
3396
                2,
3397
                "nothing from the big batch is loose"
3398
            );
33992×
            assert!(fs::read_dir(s.dir())
34002×
                .unwrap()
34012×
                .flatten()
34024×
                .all(|e| !e.file_name().to_string_lossy().contains("tmp-")));
34032×
            let s2 = Store::open_on(s.backend().clone(), s.dir()).unwrap();
3404600×
            for (r, body) in &refs {
3405600×
                assert_eq!(s2.get(r).unwrap().1, *body);
3406
            }
3407
            // A dropped batch still writes its objects.
34082×
            {
34092×
                let _b = s2.begin_batch().unwrap();
34102×
                s2.put(ObjectType::Chunk, b"dropped").unwrap();
34112×
            }
34122×
            assert_eq!(
34132×
                s2.get(&hash_object(ObjectType::Chunk, b"dropped"))
34142×
                    .unwrap()
3415
                    .1,
3416
                b"dropped"
3417
            );
34182×
        });
34191×
    }
3420
3421
    #[test]
34221×
    fn get_many_reads_loose_and_packed_objects_in_order() {
34231×
        let dir = tempfile::tempdir().unwrap();
34242×
        each_store(dir.path(), |which, s| {
34252×
            let packed: Vec<ObjectRef> = (0..20)
342640×
                .map(|i| {
342740×
                    s.put(ObjectType::Chunk, format!("packed {i}").as_bytes())
342840×
                        .unwrap()
342940×
                })
34302×
                .collect();
34312×
            s.pack_loose().unwrap();
34322×
            let loose: Vec<ObjectRef> = (0..20)
343340×
                .map(|i| {
343440×
                    s.put(ObjectType::Chunk, format!("loose {i}").as_bytes())
343540×
                        .unwrap()
343640×
                })
34372×
                .collect();
34382×
            let missing = hash_object(ObjectType::Chunk, b"never stored");
34392×
            let mut refs: Vec<ObjectRef> = Vec::new();
344040×
            for i in 0..20 {
344140×
                refs.push(loose[i]);
344240×
                refs.push(packed[i]);
344340×
            }
34442×
            refs.push(missing);
34452×
            let got = s.get_many(&refs);
34462×
            assert_eq!(got.len(), refs.len());
344780×
            for (i, r) in refs.iter().enumerate().take(40) {
344880×
                let want = if i % 2 == 0 {
344940×
                    format!("loose {}", i / 2)
3450
                } else {
345140×
                    format!("packed {}", i / 2)
3452
                };
345380×
                let (ty, body) = got[i].as_ref().unwrap();
345480×
                assert_eq!(
345580×
                    (*ty, body.as_slice()),
345680×
                    (ObjectType::Chunk, want.as_bytes()),
3457
                    "{which}: {r:?}"
3458
                );
3459
            }
34602×
            assert!(matches!(got[40], Err(StoreError::NotFound(_))));
34612×
        });
34621×
    }
3463
3464
    /// A packed object is read from its pack without first asking the
3465
    /// backend for a loose copy (oxygenhub#228: on a bucket each such ask
3466
    /// was a request that found nothing, every time).
3467
    #[test]
34681×
    fn packed_objects_cost_no_loose_lookups() {
3469
        use std::sync::atomic::{AtomicUsize, Ordering};
3470
        struct Counting(MemoryStore, AtomicUsize);
3471
        impl crate::strata::Store for Counting {
34720×
            fn describe(&self) -> String {
34730×
                "counting".into()
34740×
            }
347510×
            fn has(&self, k: &str) -> crate::strata::Result<bool> {
347610×
                self.0.has(k)
347710×
            }
347815×
            fn get(&self, k: &str) -> crate::strata::Result<Vec<u8>> {
347915×
                if k.starts_with("loose/") {
348011×
                    self.1.fetch_add(1, Ordering::Relaxed);
348111×
                }
348215×
                self.0.get(k)
348315×
            }
348420×
            fn get_range(
348520×
                &self,
348620×
                k: &str,
348720×
                r: std::ops::Range<u64>,
348820×
            ) -> crate::strata::Result<Vec<u8>> {
348920×
                self.0.get_range(k, r)
349020×
            }
349112×
            fn put_new(&self, k: &str, b: &[u8]) -> crate::strata::Result<()> {
349212×
                self.0.put_new(k, b)
349312×
            }
34948×
            fn list(&self, p: &str) -> crate::strata::Result<Vec<crate::strata::Entry>> {
34958×
                self.0.list(p)
34968×
            }
34970×
            fn delete(&self, k: &str) -> crate::strata::Result<()> {
34980×
                self.0.delete(k)
34990×
            }
35003×
            fn read_ref(
35013×
                &self,
35023×
                k: &str,
35033×
            ) -> crate::strata::Result<Option<(Vec<u8>, crate::strata::Version)>> {
35043×
                self.0.read_ref(k)
35053×
            }
35061×
            fn write_ref(
35071×
                &self,
35081×
                k: &str,
35091×
                b: &[u8],
35101×
                e: Option<&crate::strata::Version>,
35111×
            ) -> crate::strata::Result<crate::strata::Version> {
35121×
                self.0.write_ref(k, b, e)
35131×
            }
3514
        }
35151×
        let dir = tempfile::tempdir().unwrap();
35161×
        let backend = Arc::new(Counting(MemoryStore::new(), AtomicUsize::new(0)));
35171×
        let s = Store::open_on(backend.clone(), dir.path().join("scratch")).unwrap();
35181×
        let refs: Vec<ObjectRef> = (0..10)
351910×
            .map(|i| {
352010×
                s.put(ObjectType::Chunk, format!("p {i}").as_bytes())
352110×
                    .unwrap()
352210×
            })
35231×
            .collect();
35241×
        s.pack_loose().unwrap();
35251×
        let s = Store::open_on(backend.clone(), dir.path().join("scratch")).unwrap();
35261×
        backend.1.store(0, Ordering::Relaxed);
352710×
        for r in &refs {
352810×
            assert!(s.get(r).is_ok());
3529
        }
353010×
        assert!(s.get_many(&refs).iter().all(|g| g.is_ok()));
35311×
        assert_eq!(backend.1.load(Ordering::Relaxed), 0, "no loose lookups");
3532
        // An object no pack holds is still looked for loose.
35331×
        let missing = hash_object(ObjectType::Chunk, b"never stored");
35341×
        assert!(matches!(s.get(&missing), Err(StoreError::NotFound(_))));
35351×
        assert!(backend.1.load(Ordering::Relaxed) > 0);
35361×
    }
3537
}
3538
3539
/// Readers never wait (oxygenhub#249): a store keeps reading while other
3540
/// handles repack, collect and push into the store under it.
3541
#[cfg(test)]
3542
pub(crate) mod reader_tests {
3543
    use super::*;
3544
35451682×
    fn chunk(i: usize) -> Vec<u8> {
35461682×
        format!("object {i} ").repeat(1 + i % 5).into_bytes()
35471682×
    }
3548
3549
    /// Two handles on one store: a writer, and a reader opened at once.
35504×
    fn each_pair(f: impl Fn(&str, Store, Store)) {
35514×
        let dir = tempfile::tempdir().unwrap();
35524×
        let s = Store::open(dir.path().join("objects")).unwrap();
35534×
        let r = Store::open(dir.path().join("objects")).unwrap();
35544×
        f("directory", s, r);
35554×
        let m: Arc<Backend> = Arc::new(MemoryStore::new());
35564×
        let s = Store::open_on(m.clone(), dir.path().join("scratch")).unwrap();
35574×
        let r = Store::open_on(m, dir.path().join("scratch-r")).unwrap();
35584×
        f("memory", s, r);
35594×
    }
3560
3561
    /// Set every retired entry's time back by `by`, as if retired then.
35624×
    pub(crate) fn age_retired(s: &Store, by: std::time::Duration) {
35634×
        s.update_retired(|m| {
356445×
            for t in m.values_mut() {
356545×
                *t = t.saturating_sub(by.as_millis() as u64);
356645×
            }
35674×
        })
35684×
        .unwrap();
35694×
    }
3570
357112×
    fn retired_keys(s: &Store) -> Vec<String> {
357212×
        s.read_retired().unwrap().0.into_keys().collect()
357312×
    }
3574
3575
    #[test]
35761×
    fn a_reader_opened_before_a_repack_reads_every_object_after_it() {
35772×
        each_pair(|which, writer, _| {
35782×
            let mut refs = Vec::new();
357980×
            for i in 0..40 {
358080×
                refs.push(writer.put(ObjectType::Chunk, &chunk(i)).unwrap());
358180×
            }
35822×
            writer.pack_loose().unwrap();
358380×
            for i in 40..80 {
358480×
                refs.push(writer.put(ObjectType::Chunk, &chunk(i)).unwrap());
358580×
            }
3586
            // Opened now: one pack and forty loose objects.
35872×
            let reader = Store::open_on(writer.backend().clone(), writer.dir()).unwrap();
35882×
            assert_eq!(reader.pack_count(), 1, "{which}");
35892×
            let (_, merged) = writer.repack_all(REPACK_ZSTD_LEVEL).unwrap();
35902×
            assert_eq!(merged, 1);
3591
            // The old pack and the loose objects are retired, not deleted.
35922×
            let retired = retired_keys(&writer);
35932×
            assert_eq!(retired.len(), 81, "{which}: {retired:?}");
35942×
            assert_eq!(writer.loose_count().unwrap(), 0, "retired is not loose");
3595
            // The reader reads on from what it loaded.
3596160×
            for (i, r) in refs.iter().enumerate() {
3597160×
                assert_eq!(reader.get(r).unwrap().1, chunk(i), "{which}: {i}");
3598
            }
35992×
            assert_eq!(reader.pack_count(), 1);
3600
            // A store opened now reads the new pack only.
36012×
            let fresh = Store::open_on(writer.backend().clone(), writer.dir()).unwrap();
36022×
            assert_eq!(fresh.pack_count(), 1);
36032×
            assert_ne!(fresh.packs.borrow()[0].key, reader.packs.borrow()[0].key);
3604
            // Once the retired files are deleted, the reader finds every
3605
            // object in the new pack, by get, get_many and has alike.
36062×
            let report = writer
36072×
                .expire_retired(std::time::Duration::ZERO, &mut silent())
36082×
                .unwrap();
36092×
            assert_eq!((report.packs, report.loose), (1, 80), "{which}");
36102×
            assert!(retired_keys(&writer).is_empty());
3611160×
            for (i, r) in refs.iter().enumerate() {
3612160×
                assert_eq!(reader.get(r).unwrap().1, chunk(i), "{which}: {i}");
3613
            }
36142×
            let other = Store::open_on(writer.backend().clone(), writer.dir()).unwrap();
36152×
            let before = other.packs.borrow()[0].key.clone();
36162×
            writer.put(ObjectType::Chunk, b"new").unwrap();
36172×
            writer.repack_all(REPACK_ZSTD_LEVEL).unwrap();
36182×
            writer
36192×
                .expire_retired(std::time::Duration::ZERO, &mut silent())
36202×
                .unwrap();
3621160×
            assert!(other.get_many(&refs).iter().all(|g| g.is_ok()), "{which}");
36222×
            assert_ne!(other.packs.borrow()[0].key, before);
36232×
            let third = Store::open_on(writer.backend().clone(), writer.dir()).unwrap();
36242×
            writer.put(ObjectType::Chunk, b"newer").unwrap();
36252×
            writer.repack_all(REPACK_ZSTD_LEVEL).unwrap();
36262×
            writer
36272×
                .expire_retired(std::time::Duration::ZERO, &mut silent())
36282×
                .unwrap();
3629160×
            assert!(refs.iter().all(|r| third.has(r)), "{which}");
36302×
            assert!(!third.has(&hash_object(ObjectType::Chunk, b"never")));
36312×
        });
36321×
    }
3633
3634
    #[test]
36351×
    fn a_reader_opened_before_a_spilled_batch_reads_its_objects() {
36362×
        each_pair(|which, writer, reader| {
36372×
            assert_eq!(reader.pack_count(), 0);
36382×
            let mut refs = Vec::new();
3639
            {
36402×
                let batch = writer.begin_batch().unwrap();
3641612×
                for i in 0..(SPILL_OBJECTS + 50) {
3642612×
                    refs.push(writer.put(ObjectType::Chunk, &chunk(i)).unwrap());
3643612×
                }
36442×
                batch.finish().unwrap();
3645
            }
36462×
            assert_eq!(writer.pack_count(), 1, "{which}: the batch spilled");
36472×
            assert_eq!(reader.get(&refs[0]).unwrap().1, chunk(0), "{which}");
36482×
            assert_eq!(reader.pack_count(), 1, "the miss loaded the new pack");
36492×
            let got = reader.get_many(&refs);
3650612×
            assert!(got.iter().all(|g| g.is_ok()), "{which}");
3651
            // has, too, from another reader.
36522×
            let late = Store::open_on(writer.backend().clone(), writer.dir()).unwrap();
3653
            {
36542×
                let batch = writer.begin_batch().unwrap();
3655514×
                for i in 1000..(1000 + SPILL_OBJECTS + 1) {
3656514×
                    writer.put(ObjectType::Chunk, &chunk(i)).unwrap();
3657514×
                }
36582×
                batch.finish().unwrap();
3659
            }
36602×
            assert!(late.has(&hash_object(ObjectType::Chunk, &chunk(1000))));
36612×
        });
36621×
    }
3663
3664
    #[test]
36651×
    fn retired_entries_expire_after_the_hour_and_are_never_loaded() {
36662×
        each_pair(|which, s, _| {
366720×
            for i in 0..10 {
366820×
                s.put(ObjectType::Chunk, &chunk(i)).unwrap();
366920×
            }
36702×
            s.pack_loose().unwrap();
36712×
            s.put(ObjectType::Chunk, &chunk(10)).unwrap();
36722×
            s.repack_all(ZSTD_LEVEL).unwrap();
3673
            // Retired over an hour ago: the first pack and 10 loose objects.
36742×
            age_retired(&s, RETIRED_FOR + std::time::Duration::from_secs(60));
36752×
            let old = retired_keys(&s);
36762×
            assert_eq!(old.len(), 12, "{which}: {old:?}");
3677
            // Retired just now: the second pack.
36782×
            s.put(ObjectType::Chunk, &chunk(11)).unwrap();
36792×
            s.repack_all(ZSTD_LEVEL).unwrap();
36802×
            assert_eq!(retired_keys(&s).len(), 14);
36812×
            let listed = s
36822×
                .backend()
36832×
                .list("pack-")
36842×
                .unwrap()
36852×
                .into_iter()
368614×
                .filter(|e| e.key.ends_with(".idx"))
36872×
                .count();
36882×
            assert_eq!(listed, 3, "{which}");
36892×
            let fresh = Store::open_on(s.backend().clone(), s.dir()).unwrap();
36902×
            assert_eq!(
36912×
                fresh.pack_count(),
3692
                1,
3693
                "{which}: retired packs are not loaded"
3694
            );
36952×
            let report = s.expire_retired(RETIRED_FOR, &mut silent()).unwrap();
36962×
            assert_eq!(
3697
                report,
3698
                RetiredReport {
3699
                    packs: 1,
3700
                    loose: 11,
3701
                    kept: 2
3702
                },
3703
                "{which}"
3704
            );
37052×
            let young = retired_keys(&s);
37062×
            assert_eq!(young.len(), 2, "{which}: {young:?}");
37074×
            assert!(young.iter().all(|k| !old.contains(k)));
370824×
            for k in old.iter().filter(|k| k.ends_with(".pack")) {
37092×
                assert!(!s.backend().has(k).unwrap(), "{which}: {k}");
3710
            }
37114×
            for k in &young {
37124×
                assert!(s.backend().has(k).unwrap(), "{which}: {k} kept");
3713
            }
371424×
            for i in 0..12 {
371524×
                let r = hash_object(ObjectType::Chunk, &chunk(i));
371624×
                assert_eq!(fresh.get(&r).unwrap().1, chunk(i), "{which}");
3717
            }
37182×
        });
37191×
    }
3720
3721
    /// A retired loose object whose pack copy was pruned since is not
3722
    /// lost when the retired files go: it is loose again.
3723
    #[test]
37241×
    fn a_retired_loose_object_no_pack_holds_is_kept() {
37252×
        each_pair(|which, s, _| {
37262×
            let a = s.put(ObjectType::Chunk, b"a").unwrap();
37272×
            let b = s.put(ObjectType::Chunk, b"b").unwrap();
37282×
            s.pack_loose().unwrap();
37292×
            assert_eq!(s.loose_count().unwrap(), 0);
3730
            // The pack rewritten without b, as pruning does.
37312×
            let first = s.packs.borrow()[0].key.clone();
37322×
            s.write_pack(&[a], &[], &[first], ZSTD_LEVEL, &mut silent())
37332×
                .unwrap();
37342×
            let report = s
37352×
                .expire_retired(std::time::Duration::ZERO, &mut silent())
37362×
                .unwrap();
37372×
            assert_eq!((report.packs, report.loose), (1, 1), "{which}: a's copy");
37382×
            assert!(retired_keys(&s).is_empty());
37392×
            assert_eq!(s.get(&b).unwrap().1, b"b", "{which}");
37402×
            assert_eq!(s.get(&a).unwrap().1, b"a");
37412×
            assert_eq!(s.loose_count().unwrap(), 1, "b is loose again");
3742
            // And pruning it, unreachable, deletes it.
37432×
            let keep = [a].into_iter().collect();
37442×
            let r = s.prune(&keep, std::time::Duration::ZERO).unwrap();
37452×
            assert_eq!(r.objects, 1);
37462×
            assert!(!s.has(&b));
37472×
        });
37481×
    }
3749
3750
    /// A backend whose first pack listing names an index that is deleted
3751
    /// before it is read, as a collection running meanwhile would.
3752
    pub(crate) struct Racing {
3753
        inner: MemoryStore,
3754
        lists: std::sync::atomic::AtomicUsize,
3755
        /// Deleted right after the first listing returns.
3756
        pub(crate) doomed: std::sync::Mutex<Option<String>>,
3757
        /// Named in every listing, never stored.
3758
        ghost: bool,
3759
    }
3760
3761
    impl crate::strata::Store for Racing {
37620×
        fn describe(&self) -> String {
37630×
            "racing".into()
37640×
        }
376511×
        fn has(&self, k: &str) -> crate::strata::Result<bool> {
376611×
            self.inner.has(k)
376711×
        }
376873×
        fn get(&self, k: &str) -> crate::strata::Result<Vec<u8>> {
376973×
            self.inner.get(k)
377073×
        }
37716×
        fn get_range(&self, k: &str, r: std::ops::Range<u64>) -> crate::strata::Result<Vec<u8>> {
37726×
            self.inner.get_range(k, r)
37736×
        }
377422×
        fn put_new(&self, k: &str, b: &[u8]) -> crate::strata::Result<()> {
377522×
            self.inner.put_new(k, b)
377622×
        }
377770×
        fn list(&self, p: &str) -> crate::strata::Result<Vec<crate::strata::Entry>> {
377870×
            let mut out = self.inner.list(p)?;
377970×
            if p == "pack-" {
378038×
                self.lists
378138×
                    .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
378238×
                if let Some(k) = self.doomed.lock().unwrap().take() {
37832×
                    self.inner.delete(&k)?;
378436×
                }
378538×
                if self.ghost {
378658×
                    for k in ["pack-0000.idx", "pack-0000.pack"] {
378758×
                        out.push(crate::strata::Entry {
378858×
                            key: k.into(),
378958×
                            size: 100,
379058×
                            modified_ms: None,
379158×
                        });
379258×
                    }
3793220×
                    out.sort_by(|a, b| a.key.cmp(&b.key));
37949×
                }
379532×
            }
379670×
            Ok(out)
379770×
        }
37980×
        fn delete(&self, k: &str) -> crate::strata::Result<()> {
37990×
            self.inner.delete(k)
38000×
        }
380129×
        fn read_ref(
380229×
            &self,
380329×
            k: &str,
380429×
        ) -> crate::strata::Result<Option<(Vec<u8>, crate::strata::Version)>> {
380529×
            self.inner.read_ref(k)
380629×
        }
38076×
        fn write_ref(
38086×
            &self,
38096×
            k: &str,
38106×
            b: &[u8],
38116×
            e: Option<&crate::strata::Version>,
38126×
        ) -> crate::strata::Result<crate::strata::Version> {
38136×
            self.inner.write_ref(k, b, e)
38146×
        }
3815
    }
3816
38173×
    pub(crate) fn racing(ghost: bool) -> Arc<Racing> {
38183×
        Arc::new(Racing {
38193×
            inner: MemoryStore::new(),
38203×
            lists: Default::default(),
38213×
            doomed: Default::default(),
38223×
            ghost,
38233×
        })
38243×
    }
3825
3826
    #[test]
38271×
    fn an_index_deleted_between_listing_and_loading_is_passed_over() {
38281×
        let dir = tempfile::tempdir().unwrap();
38291×
        let backend = racing(false);
38301×
        let s = Store::open_on(backend.clone(), dir.path()).unwrap();
38311×
        let a = s.put(ObjectType::Chunk, b"a").unwrap();
38321×
        s.pack_loose().unwrap();
38331×
        let b = s.put(ObjectType::Chunk, b"b").unwrap();
38341×
        s.pack_loose().unwrap();
38351×
        let idx = {
38361×
            let packs = s.packs.borrow();
38371×
            let (i, _) = s.find_packed(&a).unwrap();
38381×
            format!("{}.idx", packs[i].key.trim_end_matches(".pack"))
3839
        };
38401×
        *backend.doomed.lock().unwrap() = Some(idx);
38411×
        let before = backend.lists.load(std::sync::atomic::Ordering::Relaxed);
38421×
        let again = Store::open_on(backend.clone(), dir.path()).unwrap();
38431×
        assert_eq!(again.pack_count(), 1, "the deleted one is passed over");
38441×
        assert_eq!(
38451×
            backend.lists.load(std::sync::atomic::Ordering::Relaxed) - before,
3846
            2,
3847
            "and the store listed again"
3848
        );
38491×
        assert_eq!(again.get(&b).unwrap().1, b"b");
3850
        // A listing that names an index never there opens too.
38511×
        let ghosted = racing(true);
38521×
        let s = Store::open_on(ghosted.clone(), dir.path().join("g")).unwrap();
38531×
        let c = s.put(ObjectType::Chunk, b"c").unwrap();
38541×
        s.pack_loose().unwrap();
38551×
        let again = Store::open_on(ghosted, dir.path().join("g")).unwrap();
38561×
        assert_eq!(again.pack_count(), 1);
38571×
        assert_eq!(again.get(&c).unwrap().1, b"c");
38581×
    }
3859
}
3860
3861
#[cfg(test)]
3862
mod batch_tests {
3863
    use super::*;
3864
3865
    #[test]
38661×
    fn a_large_batch_packs_every_object_and_indexes_them_all() {
38671×
        let dir = tempfile::tempdir().unwrap();
38681×
        let store = Store::open(dir.path()).unwrap();
38691×
        let mut refs = Vec::new();
3870
        {
38711×
            let batch = store.begin_batch().unwrap();
38721000×
            for i in 0..1000u32 {
38731000×
                let body = format!("object {i} {}", "x".repeat(20_000 + i as usize)).into_bytes();
38741000×
                refs.push(store.put(ObjectType::Blob, &body).unwrap());
38751000×
            }
38761×
            batch.finish().unwrap();
3877
        }
38781×
        let again = Store::open(dir.path()).unwrap();
38791000×
        for r in &refs {
38801000×
            again
38811000×
                .get(r)
38821000×
                .unwrap_or_else(|e| panic!("{}: {e}", r.short(12)));
3883
        }
38841×
    }
3885
3886
    #[test]
38871×
    fn a_pack_a_stopped_batch_left_is_adopted_and_its_torn_end_cut() {
3888
        // oxygen#79: a batch never finished (a process killed part way)
3889
        // leaves its objects in an unindexed .tmp pack.
38901×
        let dir = tempfile::tempdir().unwrap();
38911×
        let mut refs = Vec::new();
3892
        {
38931×
            let store = Store::open(dir.path()).unwrap();
38941×
            let batch = store.begin_batch().unwrap();
3895300×
            for i in 0..300u32 {
3896300×
                let body = format!("object {i} {}", "y".repeat(20_000 + i as usize)).into_bytes();
3897300×
                refs.push(store.put(ObjectType::Blob, &body).unwrap());
3898300×
            }
38991×
            std::mem::forget(batch);
3900
        }
39011×
        let tmp: Vec<_> = fs::read_dir(dir.path())
39021×
            .unwrap()
39031×
            .flatten()
39042×
            .filter(|e| e.file_name().to_string_lossy().starts_with(".tmp-"))
39051×
            .map(|e| e.path())
39061×
            .collect();
39071×
        assert_eq!(tmp.len(), 1, "one unfinished pack");
3908
        // A torn last entry: a header and part of its body.
39091×
        let mut f = fs::OpenOptions::new().append(true).open(&tmp[0]).unwrap();
39101×
        f.write_all(&[0, 1, 0xff, 0, 0, 0, 9, 9, 0, 0]).unwrap();
39111×
        f.write_all(&[7; 40]).unwrap();
39121×
        drop(f);
39131×
        let store = Store::open(dir.path()).unwrap();
39141×
        assert!(store.get(&refs[0]).is_err(), "unreadable before");
39151×
        let young = store
39161×
            .adopt_unfinished_packs(std::time::Duration::from_secs(3600), &|_| {})
39171×
            .unwrap();
39181×
        assert_eq!(young, (0, 0), "a pack written to lately is left alone");
39191×
        let (packs, objects) = store
39202×
            .adopt_unfinished_packs(std::time::Duration::ZERO, &|_| {})
39211×
            .unwrap();
39221×
        assert_eq!((packs, objects), (1, 300));
3923300×
        for r in &refs {
3924300×
            store
3925300×
                .get(r)
3926300×
                .unwrap_or_else(|e| panic!("{}: {e}", r.short(12)));
3927
        }
39281×
        let again = Store::open(dir.path()).unwrap();
39291×
        assert!(again.get(&refs[299]).is_ok(), "indexed for good");
39301×
    }
3931
}
3932
3933
#[cfg(test)]
3934
mod size_tests {
3935
    use super::*;
3936
3937
    #[test]
39381×
    fn an_object_over_the_limit_is_refused_and_nothing_is_written() {
39391×
        let dir = tempfile::tempdir().unwrap();
39401×
        let store = Store::open(dir.path()).unwrap();
39411×
        let big = vec![b'x'; MAX_OBJECT_LEN as usize + 1];
3942
        {
39431×
            let batch = store.begin_batch().unwrap();
39441×
            store.put(ObjectType::Blob, b"small").unwrap();
39451×
            let e = store.put(ObjectType::Change, &big).unwrap_err();
39461×
            assert!(
39471×
                matches!(e, StoreError::TooLarge { kind: "change", .. }),
3948
                "{e}"
3949
            );
39501×
            assert!(e.to_string().contains("commit it in parts"), "{e}");
39511×
            batch.finish().unwrap();
3952
        }
3953
        // The store still opens and reads what was written.
39541×
        let again = Store::open(dir.path()).unwrap();
39551×
        again.get(&hash_object(ObjectType::Blob, b"small")).unwrap();
39561×
    }
3957
}
3958
3959
#[cfg(test)]
3960
mod text_tests {
3961
    use super::*;
3962
    use crate::objects::{AtomRef, Edit, CHANGE_VERSION};
3963
39642×
    fn large_change() -> Change {
39652×
        let mut c = crate::objects::tests::sample_change();
39662×
        assert_eq!(c.version, CHANGE_VERSION);
39672×
        c.edits[0] = Edit::Insert {
39682×
            node: crate::ids::NodeId([7; 16]),
39692×
            after: AtomRef::Start,
39702×
            before: AtomRef::End,
39712×
            lines: (0..2000)
39724000×
                .map(|i| format!("line {i}\n").into_bytes())
39732×
                .collect(),
3974
        };
39752×
        c
39762×
    }
3977
3978
    #[test]
39791×
    fn a_large_insert_is_stored_beside_its_change_and_read_back() {
39801×
        let dir = tempfile::tempdir().unwrap();
39811×
        let s = Store::open(dir.path()).unwrap();
39821×
        let c = large_change();
39831×
        let r = s.put_change(&c).unwrap();
39841×
        assert_eq!(r, c.hash());
39851×
        let (_, body) = s.get(&r).unwrap();
39861×
        assert!(
39871×
            body.len() < 1024,
3988
            "the change holds a reference, not {} bytes of lines",
39890×
            body.len()
3990
        );
39911×
        let (back, parts) = s.get_change_parts(&r).unwrap();
39921×
        assert_eq!(back, c);
3993
        // The text blob and its chunk.
39941×
        assert_eq!(parts.len(), 2);
39952×
        for p in &parts {
39962×
            assert!(s.has(p));
3997
        }
3998
        // Storing it again writes nothing new.
39991×
        let n = s.list().unwrap().len();
40001×
        s.put_change(&c).unwrap();
40011×
        assert_eq!(s.list().unwrap().len(), n);
40021×
    }
4003
4004
    #[test]
40051×
    fn a_change_whose_text_blob_is_missing_is_reported_not_misread() {
40061×
        if !super::files_on_disk() {
40070×
            return;
40081×
        }
40091×
        let dir = tempfile::tempdir().unwrap();
40101×
        let s = Store::open(dir.path()).unwrap();
40111×
        let c = large_change();
40121×
        let r = s.put_change(&c).unwrap();
40131×
        let (_, parts) = s.get_change_parts(&r).unwrap();
40141×
        fs::remove_file(s.path_for(&parts[0])).unwrap();
40151×
        let e = s.get_change(&r).unwrap_err();
40161×
        assert!(
40171×
            matches!(&e, StoreError::Corrupt(h, why) if *h == r && why.contains("missing")),
4018
            "{e}"
4019
        );
40201×
    }
4021
}
4022
4023
#[cfg(test)]
4024
mod put_many_tests {
4025
    use super::*;
4026
4027
    #[test]
40281×
    fn many_objects_in_a_batch_are_compressed_together_and_all_read_back() {
40291×
        let dir = tempfile::tempdir().unwrap();
40301×
        let s = Store::open(dir.path()).unwrap();
40311×
        let before = s.put(ObjectType::Chunk, b"already here").unwrap();
40321×
        let bodies: Vec<Vec<u8>> = (0..200u32)
4033200×
            .map(|i| format!("chunk {i} ").repeat(6000).into_bytes())
40341×
            .collect();
40351×
        let mut objects: Vec<(ObjectType, &[u8])> = bodies
40361×
            .iter()
4037200×
            .map(|b| (ObjectType::Chunk, b.as_slice()))
40381×
            .collect();
40391×
        objects.push((ObjectType::Chunk, bodies[0].as_slice()));
40401×
        objects.push((ObjectType::Chunk, b"already here"));
40411×
        let refs = {
40421×
            let batch = s.begin_batch().unwrap();
40431×
            let refs = s.put_many(&objects).unwrap();
40441×
            batch.finish().unwrap();
40451×
            refs
4046
        };
40471×
        assert_eq!(refs.len(), objects.len());
40481×
        assert_eq!(refs[200], refs[0], "a repeat is the same object");
40491×
        assert_eq!(refs[201], before);
40501×
        let again = Store::open(dir.path()).unwrap();
4051202×
        for (r, b) in refs.iter().zip(&objects) {
4052202×
            assert_eq!(again.get(r).unwrap().1, b.1);
4053
        }
40541×
    }
4055
}
4056
4057
/// Signatures and index entries on a store that is not this disk's
4058
/// directory: the journal and the entry lists (spec §13.7, #116).
4059
#[cfg(test)]
4060
mod entry_tests {
4061
    use super::*;
4062
    use crate::identity::{Domain, SecretKey};
4063
    use crate::strata::Store as _;
4064
    use std::collections::HashMap;
4065
    use std::sync::atomic::{AtomicBool, Ordering};
4066
    use std::sync::Mutex;
4067
4068
    /// A store in memory that counts its requests by kind, as a bucket
4069
    /// store's /health counters do; it can also slip a competing journal
4070
    /// line in just before a write of the journal, as another process
4071
    /// would.
4072
    #[derive(Default)]
4073
    struct Counted {
4074
        inner: MemoryStore,
4075
        counts: Mutex<HashMap<&'static str, usize>>,
4076
        interfere: AtomicBool,
4077
    }
4078
4079
    impl Counted {
40802471×
        fn count(&self, kind: &'static str) {
40812471×
            *self.counts.lock().unwrap().entry(kind).or_default() += 1;
40822471×
        }
408317×
        fn n(&self, kind: &str) -> usize {
408417×
            self.counts.lock().unwrap().get(kind).copied().unwrap_or(0)
408517×
        }
40869×
        fn reset(&self) {
40879×
            self.counts.lock().unwrap().clear();
40889×
        }
40892×
        fn total(&self) -> usize {
40902×
            self.counts.lock().unwrap().values().sum()
40912×
        }
40922×
        fn summary(&self) -> String {
40932×
            let c = self.counts.lock().unwrap();
40945×
            let mut v: Vec<String> = c.iter().map(|(k, n)| format!("{k} {n}")).collect();
40952×
            v.sort();
40962×
            v.join(", ")
40972×
        }
4098
    }
4099
4100
    impl crate::strata::Store for Counted {
41010×
        fn describe(&self) -> String {
41020×
            "counted".into()
41030×
        }
410447×
        fn has(&self, k: &str) -> crate::strata::Result<bool> {
410547×
            self.count("has");
410647×
            self.inner.has(k)
410747×
        }
4108120×
        fn get(&self, k: &str) -> crate::strata::Result<Vec<u8>> {
4109120×
            self.count(if is_legacy(k) { "get_legacy" } else { "get" });
4110120×
            self.inner.get(k)
4111120×
        }
41120×
        fn get_range(&self, k: &str, r: std::ops::Range<u64>) -> crate::strata::Result<Vec<u8>> {
41130×
            self.count("range");
41140×
            self.inner.get_range(k, r)
41150×
        }
41161300×
        fn put_new(&self, k: &str, b: &[u8]) -> crate::strata::Result<()> {
41171300×
            self.count("put");
41181300×
            self.inner.put_new(k, b)
41191300×
        }
4120271×
        fn list(&self, p: &str) -> crate::strata::Result<Vec<crate::strata::Entry>> {
4121271×
            self.count(if p == "pack-" {
412274×
                "list_packs"
4123197×
            } else if is_legacy(p) {
412455×
                "list_legacy"
4125
            } else {
4126142×
                "list"
4127
            });
4128271×
            self.inner.list(p)
4129271×
        }
413020×
        fn delete(&self, k: &str) -> crate::strata::Result<()> {
413120×
            self.count("delete");
413220×
            self.inner.delete(k)
413320×
        }
4134382×
        fn read_ref(
4135382×
            &self,
4136382×
            k: &str,
4137382×
        ) -> crate::strata::Result<Option<(Vec<u8>, crate::strata::Version)>> {
4138382×
            self.count(if k == JOURNAL_KEY {
4139366×
                "read_journal"
4140
            } else {
414116×
                "read_ref"
4142
            });
4143382×
            self.inner.read_ref(k)
4144382×
        }
4145331×
        fn write_ref(
4146331×
            &self,
4147331×
            k: &str,
4148331×
            b: &[u8],
4149331×
            e: Option<&crate::strata::Version>,
4150331×
        ) -> crate::strata::Result<crate::strata::Version> {
4151331×
            self.count("write_ref");
4152331×
            if k == JOURNAL_KEY && self.interfere.swap(false, Ordering::SeqCst) {
41531×
                let (mut m, v) = match self.inner.read_ref(k)? {
41540×
                    Some((bytes, v)) => (parse_retired(&bytes), Some(v)),
41551×
                    None => (Default::default(), None),
4156
                };
41571×
                m.insert(
41581×
                    format!("seal/{}", "c".repeat(64)) + "/" + &"d".repeat(64),
4159
                    0,
4160
                );
41611×
                self.inner.write_ref(k, &encode_retired(&m), v.as_ref())?;
4162330×
            }
4163331×
            self.inner.write_ref(k, b, e)
4164331×
        }
4165
    }
4166
4167
    /// Whether a key or prefix is under an older one-list layout.
4168317×
    fn is_legacy(k: &str) -> bool {
4169317×
        ENTRY_INDEXES
4170317×
            .iter()
41711539×
            .any(|(_, legacy)| k.starts_with(&format!("{legacy}/")))
4172317×
    }
4173
4174
    struct Fixture {
4175
        backend: Arc<Counted>,
4176
        dir: tempfile::TempDir,
4177
        handles: std::cell::Cell<usize>,
4178
    }
4179
4180
    impl Fixture {
41819×
        fn new() -> Fixture {
41829×
            Fixture {
41839×
                backend: Arc::new(Counted::default()),
41849×
                dir: tempfile::tempdir().unwrap(),
41859×
                handles: Default::default(),
41869×
            }
41879×
        }
4188
        /// Another handle on the store, as another process opens it.
418931×
        fn open(&self) -> Store {
419031×
            let n = self.handles.get() + 1;
419131×
            self.handles.set(n);
419231×
            Store::open_on(self.backend.clone(), self.dir.path().join(format!("s{n}"))).unwrap()
419331×
        }
4194
    }
4195
419616×
    fn signer(n: u8) -> SecretKey {
419716×
        SecretKey {
419816×
            ed_seed: [n; 32],
419916×
            ml_seed: [n + 100; 32],
420016×
        }
420116×
    }
4202
420314×
    fn sign(s: &Store, r: &ObjectRef, n: u8) -> Signature {
420414×
        let sk = signer(n);
420514×
        let key = s.put_identity(&sk.identity("a", "b", 0)).unwrap();
420614×
        let sig = sk.sign(Domain::Change, &key, r);
420714×
        s.put_signature(r, &sig).unwrap();
420814×
        sig
420914×
    }
4210
42118×
    fn journal(s: &Store) -> std::collections::BTreeMap<String, u64> {
42128×
        s.read_record(JOURNAL_KEY).unwrap().0
42138×
    }
4214
421514×
    fn fold(s: &Store) -> bool {
421614×
        s.fold_entries(&mut silent()).unwrap()
421714×
    }
4218
4219
    /// What 0.4.6 writes for a signature: its own key, and nothing else.
42202×
    fn old_writer_signs(f: &Fixture, r: &ObjectRef, n: u8) -> Signature {
42212×
        let sk = signer(n);
42222×
        let sig = sk.sign(Domain::Change, &ObjectRef([n; 32]), r);
42232×
        f.backend
42242×
            .put_new(
42252×
                &format!("{}{}", fanout_prefix("sig", r), sig.key.to_hex()),
42262×
                &Signature::encode_list(std::slice::from_ref(&sig)),
42272×
            )
42282×
            .unwrap();
42292×
        sig
42302×
    }
4231
4232
    /// After a fold, a signature or seal another handle writes is found
4233
    /// by a store opened later through the journal, with no listing; a
4234
    /// handle that already read the entries sees it once refreshed.
4235
    #[test]
42361×
    fn an_entry_written_after_a_fold_is_found_through_the_journal() {
42371×
        let f = Fixture::new();
42381×
        let w = f.open();
42391×
        let r1 = w.put(ObjectType::Chunk, b"one").unwrap();
42401×
        let s1 = sign(&w, &r1, 1);
42411×
        assert!(fold(&w));
42421×
        assert!(!fold(&w), "nothing new: no second list");
4243
42441×
        let early = f.open();
42451×
        assert_eq!(early.signatures(&r1).unwrap(), vec![s1.clone()]);
4246
42471×
        let w2 = f.open();
42481×
        let r2 = w2.put(ObjectType::Chunk, b"two").unwrap();
42491×
        let s2 = sign(&w2, &r2, 2);
42501×
        w2.index_seal(&r2, &ObjectRef([9; 32])).unwrap();
42511×
        w2.index_receipt(&r2, &ObjectRef([8; 32])).unwrap();
4252
        // Its own writes, at once.
42531×
        assert_eq!(w2.signatures(&r2).unwrap(), vec![s2.clone()]);
4254
42551×
        let later = f.open();
42561×
        f.backend.reset();
42571×
        assert_eq!(later.signatures(&r1).unwrap(), vec![s1]);
42581×
        assert_eq!(later.signatures(&r2).unwrap(), vec![s2.clone()]);
42591×
        assert_eq!(later.seals_for(&r2).unwrap(), vec![ObjectRef([9; 32])]);
42601×
        assert_eq!(later.receipts_for(&r2).unwrap(), vec![ObjectRef([8; 32])]);
42611×
        assert_eq!(later.receipt_states().unwrap(), vec![r2]);
42621×
        assert!(later.seals_for(&r1).unwrap().is_empty());
42631×
        assert_eq!(f.backend.n("list"), 0, "{}", f.backend.summary());
42641×
        assert_eq!(f.backend.n("read_journal"), 1, "{}", f.backend.summary());
4265
4266
        // The handle that read the entries before keeps its view until
4267
        // it is refreshed, as it keeps its operation head.
42681×
        assert!(early.signatures(&r2).unwrap().is_empty());
42691×
        early.refresh().unwrap();
42701×
        assert_eq!(early.signatures(&r2).unwrap(), vec![s2]);
42711×
    }
4272
4273
    /// A fold marks the journal's lines rather than dropping them, so a
4274
    /// reader whose listing predates it still finds every entry; an hour
4275
    /// on, the old list is deleted and the lines dropped, and readers
4276
    /// find everything in the new list.
4277
    #[test]
42781×
    fn a_fold_keeps_the_journal_for_the_hour_readers_may_need_it() {
42791×
        let f = Fixture::new();
42801×
        let w = f.open();
42811×
        let r1 = w.put(ObjectType::Chunk, b"one").unwrap();
42821×
        let s1 = sign(&w, &r1, 1);
42831×
        assert!(fold(&w));
42841×
        let r2 = w.put(ObjectType::Chunk, b"two").unwrap();
42851×
        let s2 = sign(&w, &r2, 2);
42862×
        assert!(journal(&w).values().any(|t| *t == 0));
4287
4288
        // Opened before the second fold: lists the first list only.
42891×
        let stale = f.open();
42901×
        let stale_after_expiry = f.open();
42911×
        assert!(fold(&w));
42921×
        assert!(
42932×
            journal(&w).values().all(|t| *t != 0),
4294
            "every line is marked folded"
4295
        );
42961×
        assert!(journal(&w).len() >= 2, "and kept");
42971×
        assert_eq!(stale.signatures(&r1).unwrap(), vec![s1.clone()]);
42981×
        assert_eq!(stale.signatures(&r2).unwrap(), vec![s2.clone()]);
4299
4300
        // An hour later: the first list goes with the next collection,
4301
        // and the next journal update drops the folded lines.
43021×
        reader_tests::age_retired(&w, RETIRED_FOR * 2);
43031×
        w.update_record(JOURNAL_KEY, "test", |m| {
43042×
            for t in m.values_mut() {
43052×
                *t = t.saturating_sub(2 * RETIRED_FOR.as_millis() as u64);
43062×
            }
43071×
        })
43081×
        .unwrap();
43091×
        w.expire_retired(RETIRED_FOR, &mut silent()).unwrap();
43101×
        let lists: Vec<String> = f
43111×
            .backend
43121×
            .inner
43131×
            .list("pack-")
43141×
            .unwrap()
43151×
            .into_iter()
43161×
            .map(|e| e.key)
43174×
            .filter(|k| is_entry_list(k))
43181×
            .collect();
43191×
        assert_eq!(lists.len(), 1, "the replaced list is deleted: {lists:?}");
43201×
        let r3 = w.put(ObjectType::Chunk, b"three").unwrap();
43211×
        let s3 = sign(&w, &r3, 3);
43221×
        let j = journal(&w);
43231×
        assert_eq!(j.len(), 1, "only the new line is left: {j:?}");
43241×
        assert!(j.keys().all(|k| k.starts_with(&fanout_prefix("sig", &r3))));
4325
4326
        // A handle whose listing named the deleted list lists again.
43271×
        assert_eq!(
43281×
            stale_after_expiry.signatures(&r1).unwrap(),
43291×
            vec![s1.clone()]
4330
        );
43311×
        assert_eq!(
43321×
            stale_after_expiry.signatures(&r3).unwrap(),
43331×
            vec![s3.clone()]
4334
        );
43351×
        let fresh = f.open();
43361×
        f.backend.reset();
43371×
        assert_eq!(fresh.signatures(&r1).unwrap(), vec![s1]);
43381×
        assert_eq!(fresh.signatures(&r2).unwrap(), vec![s2]);
43391×
        assert_eq!(fresh.signatures(&r3).unwrap(), vec![s3]);
43401×
        assert_eq!(f.backend.n("list"), 0);
43411×
    }
4342
4343
    /// A store no fold has touched (as 0.4.6 left it: per-object keys and
4344
    /// older one-list keys, no journal, no lists) reads by listing, and
4345
    /// reads the same once folded. An older writer's entry after the
4346
    /// fold is missed until the next fold: the trade-off of §13.7.
4347
    #[test]
43481×
    fn an_old_layout_store_reads_before_and_after_its_first_fold() {
43491×
        let f = Fixture::new();
43501×
        let w = f.open();
43511×
        let r1 = w.put(ObjectType::Chunk, b"one").unwrap();
43521×
        let r2 = w.put(ObjectType::Chunk, b"two").unwrap();
43531×
        let s1 = old_writer_signs(&f, &r1, 1);
43541×
        let s2 = SecretKey {
43551×
            ed_seed: [2; 32],
43561×
            ml_seed: [3; 32],
43571×
        }
43581×
        .sign(Domain::Change, &ObjectRef([2; 32]), &r2);
43591×
        f.backend
43601×
            .put_new(
43611×
                &legacy_key("sigs", &r2),
43621×
                &Signature::encode_list(std::slice::from_ref(&s2)),
43631×
            )
43641×
            .unwrap();
43651×
        f.backend
43661×
            .put_new(
43671×
                &format!("{}{}", fanout_prefix("seal", &r1), "ab".repeat(32)),
43681×
                b"",
43691×
            )
43701×
            .unwrap();
43711×
        f.backend
43721×
            .put_new(
43731×
                &legacy_key("receipts", &r2),
43741×
                format!("{}\n", ObjectRef([7; 32]).to_hex()).as_bytes(),
43751×
            )
43761×
            .unwrap();
43772×
        let check = |s: &Store| {
43782×
            assert_eq!(s.signatures(&r1).unwrap(), vec![s1.clone()]);
43792×
            assert_eq!(s.signatures(&r2).unwrap(), vec![s2.clone()]);
43802×
            assert_eq!(s.seals_for(&r1).unwrap(), vec![ObjectRef([0xab; 32])]);
43812×
            assert_eq!(s.receipts_for(&r2).unwrap(), vec![ObjectRef([7; 32])]);
43822×
            assert_eq!(s.receipt_states().unwrap(), vec![r2]);
43832×
        };
43841×
        let reader = f.open();
43851×
        assert!(!reader.folded.get());
43861×
        check(&reader);
43871×
        assert!(f.backend.inner.read_ref(JOURNAL_KEY).unwrap().is_none());
4388
43891×
        assert!(fold(&w));
43901×
        let reader = f.open();
43911×
        assert!(reader.folded.get());
43921×
        check(&reader);
4393
4394
        // The older lists' entries are per-object entries now, and the
4395
        // marker is there.
43961×
        assert!(f.backend.inner.has(LEGACY_FOLDED_KEY).unwrap());
43971×
        assert!(f
43981×
            .backend
43991×
            .inner
44001×
            .has(&format!("{}{}", fanout_prefix("sig", &r2), s2.key.to_hex()))
44011×
            .unwrap());
44021×
        assert!(f
44031×
            .backend
44041×
            .inner
44051×
            .has(&format!(
44061×
                "{}{}",
44071×
                fanout_prefix("receipt", &r2),
44081×
                ObjectRef([7; 32]).to_hex()
44091×
            ))
44101×
            .unwrap());
4411
4412
        // An older writer after the fold: missed until the next one.
44131×
        let r3 = w.put(ObjectType::Chunk, b"three").unwrap();
44141×
        let s3 = old_writer_signs(&f, &r3, 3);
44151×
        assert!(f.open().signatures(&r3).unwrap().is_empty());
44161×
        assert!(fold(&w));
44171×
        assert_eq!(f.open().signatures(&r3).unwrap(), vec![s3]);
44181×
    }
4419
4420
    /// Writers racing on the journal never lose a line: a lost
4421
    /// compare-and-swap reads the journal again and retries.
4422
    #[test]
44231×
    fn concurrent_journal_appends_lose_nothing() {
44241×
        let f = Fixture::new();
4425
        // One race, made to happen.
44261×
        let w = f.open();
44271×
        f.backend.interfere.store(true, Ordering::SeqCst);
44281×
        w.index_seal(&ObjectRef([1; 32]), &ObjectRef([2; 32]))
44291×
            .unwrap();
44301×
        let j = journal(&w);
44311×
        assert_eq!(j.len(), 2, "the competitor's line and ours: {j:?}");
4432
        // Many, from threads, each its own handle.
44331×
        let backend: Arc<Backend> = f.backend.clone();
44341×
        std::thread::scope(|scope| {
44358×
            for t in 0..8u8 {
44368×
                let backend = backend.clone();
44378×
                let dir = f.dir.path().join(format!("t{t}"));
44388×
                scope.spawn(move || {
44398×
                    let s = Store::open_on(backend, dir).unwrap();
4440200×
                    for i in 0..25u8 {
4441200×
                        s.index_seal(&ObjectRef([t + 10; 32]), &ObjectRef([i; 32]))
4442200×
                            .unwrap();
4443200×
                    }
44448×
                });
4445
            }
44461×
        });
44471×
        assert_eq!(journal(&w).len(), 2 + 8 * 25);
44481×
        assert!(fold(&w));
44491×
        let r = f.open();
44508×
        for t in 0..8u8 {
44518×
            assert_eq!(r.seals_for(&ObjectRef([t + 10; 32])).unwrap().len(), 25);
4452
        }
44531×
    }
4454
4455
    /// A batch adds its entries to the journal in one update when it is
4456
    /// flushed; past the limit the writer folds them into a partial list.
4457
    #[test]
44581×
    fn a_long_journal_is_folded_into_a_partial_list() {
44591×
        let f = Fixture::new();
44601×
        let w = f.open();
44611×
        assert!(fold(&w), "folded, with nothing yet");
44621×
        let key = ObjectRef([5; 32]);
44631×
        f.backend.reset();
44641×
        let batch = w.begin_batch().unwrap();
44651001×
        for i in 0..=JOURNAL_FOLD_AT as u32 {
44661001×
            let mut v = [0u8; 32];
44671001×
            v[..4].copy_from_slice(&i.to_le_bytes());
44681001×
            w.index_seal(&key, &ObjectRef(v)).unwrap();
44691001×
        }
44701×
        assert_eq!(f.backend.n("write_ref"), 0, "nothing until the flush");
44711×
        batch.finish().unwrap();
44721×
        let parts = f
44731×
            .backend
44741×
            .inner
44751×
            .list("pack-")
44761×
            .unwrap()
44771×
            .into_iter()
44784×
            .filter(|e| e.key.ends_with(PART_SUFFIX))
44791×
            .count();
44801×
        assert_eq!(parts, 1);
44811001×
        assert!(journal(&w).values().all(|t| *t != 0), "all folded");
44821×
        let r = f.open();
44831×
        assert_eq!(r.seals_for(&key).unwrap().len(), JOURNAL_FOLD_AT + 1);
4484
        // The next complete fold takes the partial list in.
44851×
        assert!(fold(&w));
44861×
        let lists: Vec<String> = f.open().entry_lists.borrow().clone();
44871×
        assert_eq!(lists.len(), 1);
44881×
        assert!(lists[0].ends_with(LIST_SUFFIX));
44891×
        assert_eq!(f.open().seals_for(&key).unwrap().len(), JOURNAL_FOLD_AT + 1);
44901×
    }
4491
4492
    /// A collection that deletes an object takes its entries out of the
4493
    /// journal too, so the next fold leaves them out.
4494
    #[test]
44951×
    fn pruned_objects_leave_the_journal() {
44961×
        let f = Fixture::new();
44971×
        let w = f.open();
44981×
        let gone = w.put(ObjectType::Chunk, b"gone").unwrap();
44991×
        let kept = w.put(ObjectType::Chunk, b"kept").unwrap();
45001×
        sign(&w, &gone, 1);
45011×
        let s = sign(&w, &kept, 2);
45021×
        assert!(fold(&w));
45031×
        let keep: std::collections::HashSet<ObjectRef> = w
45041×
            .list()
45051×
            .unwrap()
45061×
            .into_iter()
45074×
            .filter(|r| *r != gone)
45081×
            .collect();
45091×
        w.prune(&keep, std::time::Duration::ZERO).unwrap();
45101×
        assert!(journal(&w).keys().all(|k| !k.contains(&gone.to_hex()[2..])));
45111×
        fold(&w);
45121×
        let r = f.open();
45131×
        assert!(r.signatures(&gone).unwrap().is_empty());
45141×
        assert_eq!(r.signatures(&kept).unwrap(), vec![s]);
45151×
        let all = r.with_entries(|e| e.clone()).unwrap().unwrap();
45161×
        assert!(
45171×
            all.iter().all(|k| !k.contains(&gone.to_hex()[2..])),
4518
            "{all:?}"
4519
        );
45201×
    }
4521
4522
    /// Requests per lookup, before and after the fold (#116): twenty
4523
    /// objects, five of them signed, each looked up for its signatures
4524
    /// and its seals, as a fetch does.
4525
    #[test]
45261×
    fn a_folded_store_looks_up_entries_without_listing() {
45271×
        let f = Fixture::new();
45281×
        let w = f.open();
45291×
        let refs: Vec<ObjectRef> = (0..20)
453020×
            .map(|i| {
453120×
                w.put(ObjectType::Chunk, format!("o {i}").as_bytes())
453220×
                    .unwrap()
453320×
            })
45341×
            .collect();
45355×
        for r in &refs[..5] {
45365×
            sign(&w, r, 1);
45375×
        }
45382×
        let lookups = |s: &Store| {
453940×
            for r in &refs {
454040×
                s.signatures(r).unwrap();
454140×
                s.seals_for(r).unwrap();
454240×
            }
45432×
        };
45441×
        let before = f.open();
45451×
        f.backend.reset();
45461×
        lookups(&before);
45471×
        let (before_total, before_list) = (f.backend.total(), f.backend.n("list"));
45481×
        eprintln!("40 lookups before the fold: {}", f.backend.summary());
4549
45501×
        assert!(fold(&w));
45511×
        let after = f.open();
45521×
        f.backend.reset();
45531×
        lookups(&after);
45541×
        eprintln!("40 lookups after the fold: {}", f.backend.summary());
45551×
        assert_eq!(before_list, 40);
45561×
        assert_eq!(f.backend.n("list"), 0);
45571×
        assert_eq!(f.backend.n("get_legacy"), 0, "the older lists are folded");
45581×
        assert_eq!(f.backend.n("read_journal"), 1);
45591×
        assert!(
45601×
            f.backend.total() < before_total / 10,
4561
            "{}",
45620×
            f.backend.summary()
4563
        );
45641×
    }
4565
4566
    /// Once a fold has copied the older lists into per-object entries and
4567
    /// written its marker, readers stop reading them, folded or not; the
4568
    /// answers are the same.
4569
    #[test]
45701×
    fn the_legacy_marker_stops_reads_of_the_older_lists() {
45711×
        let f = Fixture::new();
45721×
        let w = f.open();
45731×
        let r = w.put(ObjectType::Chunk, b"old").unwrap();
45741×
        let sig = SecretKey {
45751×
            ed_seed: [4; 32],
45761×
            ml_seed: [5; 32],
45771×
        }
45781×
        .sign(Domain::Change, &ObjectRef([4; 32]), &r);
45791×
        f.backend
45801×
            .put_new(
45811×
                &legacy_key("sigs", &r),
45821×
                &Signature::encode_list(std::slice::from_ref(&sig)),
45831×
            )
45841×
            .unwrap();
45853×
        for (legacy, v) in [("seals", 9u8), ("certs", 8), ("accepted-lists", 6)] {
45863×
            f.backend
45873×
                .put_new(
45883×
                    &legacy_key(legacy, &r),
45893×
                    format!("{}\n", ObjectRef([v; 32]).to_hex()).as_bytes(),
45903×
                )
45913×
                .unwrap();
45923×
        }
45933×
        let check = |s: &Store| {
45943×
            assert_eq!(s.signatures(&r).unwrap(), vec![sig.clone()]);
45953×
            assert_eq!(s.seals_for(&r).unwrap(), vec![ObjectRef([9; 32])]);
45963×
            assert_eq!(
45973×
                s.index_get("cert", "certs", &r).unwrap(),
45983×
                vec![ObjectRef([8; 32])]
4599
            );
46003×
            assert_eq!(
46013×
                s.index_get_pub("accepted", &r).unwrap(),
46023×
                vec![ObjectRef([6; 32])]
4603
            );
46043×
        };
46051×
        let before = f.open();
46061×
        f.backend.reset();
46071×
        check(&before);
46081×
        assert_eq!(f.backend.n("get_legacy"), 4, "{}", f.backend.summary());
4609
46101×
        assert!(fold(&w));
46111×
        let folded = f.open();
46121×
        f.backend.reset();
46131×
        check(&folded);
46141×
        assert_eq!(f.backend.n("get_legacy") + f.backend.n("list"), 0);
4615
4616
        // Not folded (its lists gone), but marked: per-object listings,
4617
        // still no older lists.
46182×
        for e in f.backend.inner.list("pack-").unwrap() {
46192×
            if is_entry_list(&e.key) {
46201×
                f.backend.inner.delete(&e.key).unwrap();
46211×
            }
4622
        }
46231×
        let marked = f.open();
46241×
        assert!(!marked.folded.get() && marked.legacy_folded.get());
46251×
        f.backend.reset();
46261×
        check(&marked);
46271×
        assert_eq!(f.backend.n("get_legacy"), 0, "{}", f.backend.summary());
46281×
        assert_eq!(f.backend.n("list"), 4);
46291×
    }
4630
4631
    /// A handle that has read the entries finds another process's new
4632
    /// signature once the pack listing's interval has passed, with no
4633
    /// refresh, at the cost of one pack listing and one journal read.
4634
    #[test]
46351×
    fn a_long_lived_handle_sees_new_entries_within_the_interval() {
46361×
        let f = Fixture::new();
46371×
        let w = f.open();
46381×
        let r1 = w.put(ObjectType::Chunk, b"one").unwrap();
46391×
        let s1 = sign(&w, &r1, 1);
46401×
        assert!(fold(&w));
46411×
        let a = f.open();
46421×
        assert_eq!(a.signatures(&r1).unwrap(), vec![s1]);
4643
46441×
        let b = f.open();
46451×
        let r2 = b.put(ObjectType::Chunk, b"two").unwrap();
46461×
        let s2 = sign(&b, &r2, 2);
46471×
        b.index_approval(&r2, &ObjectRef([3; 32])).unwrap();
4648
        // Within the interval: answered from what was read.
46491×
        assert!(a.signatures(&r2).unwrap().is_empty());
4650
4651
        // The interval passes, as the pack listing's does.
46521×
        let t = a.entries_read.get().unwrap();
46531×
        a.entries_read.set(t.checked_sub(ENTRIES_FRESH));
46541×
        f.backend.reset();
46551×
        assert_eq!(a.signatures(&r2).unwrap(), vec![s2]);
46561×
        assert_eq!(a.approvals_for(&r2).unwrap(), vec![ObjectRef([3; 32])]);
46571×
        assert_eq!(f.backend.n("read_journal"), 1, "{}", f.backend.summary());
46581×
        assert_eq!(f.backend.n("list_packs"), 1, "{}", f.backend.summary());
46591×
        assert_eq!(f.backend.n("list"), 0, "{}", f.backend.summary());
46601×
        assert_eq!(
46611×
            f.backend.n("get"),
4662
            1,
4663
            "the signature; the list already read is not read again: {}",
46640×
            f.backend.summary()
4665
        );
46661×
    }
4667
}