OXYGENHUB

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

3389 of 3553 lines run, 568 of 719 functions · run #407 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.
1341085×
fn parse_retired(bytes: &[u8]) -> std::collections::BTreeMap<String, u64> {
1351085×
    String::from_utf8_lossy(bytes)
1361085×
        .lines()
13769056×
        .filter_map(|l| {
13869056×
            let (t, k) = l.split_once(' ')?;
13969056×
            Some((k.to_string(), t.parse().ok()?))
14069056×
        })
1411085×
        .collect()
1421085×
}
143
144429×
fn encode_retired(m: &std::collections::BTreeMap<String, u64>) -> Vec<u8> {
145429×
    m.iter()
14633459×
        .map(|(k, t)| format!("{t} {k}\n"))
147429×
        .collect::<String>()
148429×
        .into_bytes()
149429×
}
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.
18710881×
fn put_new_or_existing(backend: &Backend, key: &str, bytes: &[u8]) -> Result<()> {
18810881×
    match backend.put_new(key, bytes) {
18910881×
        Ok(()) | Err(BackendError::Exists(_)) => Ok(()),
1900×
        Err(e) => Err(backend_err(e)),
191
    }
19210881×
}
193
194
/// `get` as an option: `None` when there is no such key.
19570090×
fn get_optional(backend: &Backend, key: &str) -> Result<Option<Vec<u8>>> {
19670090×
    match backend.get(key) {
19761291×
        Ok(b) => Ok(Some(b)),
1988799×
        Err(BackendError::NotFound(_)) => Ok(None),
1990×
        Err(e) => Err(backend_err(e)),
200
    }
20170090×
}
202
2032860×
fn delete(backend: &Backend, key: &str) -> Result<()> {
2042860×
    backend.delete(key).map_err(backend_err)
2052860×
}
206
207
/// `loose/<hh>/<rest>`.
208106722×
fn loose_key(r: &ObjectRef) -> String {
209106722×
    let h = r.to_hex();
210106722×
    format!("loose/{}/{}", &h[..2], &h[2..])
211106722×
}
212
213
/// The object a loose key names, if it is one.
21426888×
fn loose_ref(key: &str) -> Option<ObjectRef> {
21526888×
    let rest = key.strip_prefix("loose/")?;
21626888×
    let (hh, tail) = rest.split_once('/')?;
21726888×
    if hh.len() != 2 || tail.contains('/') {
2180×
        return None;
21926888×
    }
22026888×
    ObjectRef::from_hex(&format!("{hh}{tail}"))
22126888×
}
222
223
/// `<index>/<hh>/<rest>/` for the entries kept under an object.
2249319×
fn fanout_prefix(index: &str, r: &ObjectRef) -> String {
2259319×
    let h = r.to_hex();
2269319×
    format!("{index}/{}/{}/", &h[..2], &h[2..])
2279319×
}
228
229
/// Whether a key names an entry list (spec §13.7), complete or partial.
2305089×
fn is_entry_list(key: &str) -> bool {
2315089×
    key.starts_with("pack-") && (key.ends_with(LIST_SUFFIX) || key.ends_with(PART_SUFFIX))
2325089×
}
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.
263298×
fn expire_folded(m: &mut std::collections::BTreeMap<String, u64>, now: u64) {
264298×
    let hour = RETIRED_FOR.as_millis() as u64;
26527026×
    m.retain(|_, t| *t == 0 || now.saturating_sub(*t) < hour);
266298×
}
267
268
/// `<index>/<hh>/<rest>`: the one-file-per-object layout of older stores.
2696910×
fn legacy_key(index: &str, r: &ObjectRef) -> String {
2706910×
    let h = r.to_hex();
2716910×
    format!("{index}/{}/{}", &h[..2], &h[2..])
2726910×
}
273
274411×
fn now_ms() -> u64 {
275411×
    std::time::SystemTime::now()
276411×
        .duration_since(std::time::UNIX_EPOCH)
277411×
        .map(|d| d.as_millis() as u64)
278411×
        .unwrap_or(0)
279411×
}
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.
31117705×
fn decode_entry(
31217705×
    r: &ObjectRef,
31317705×
    e: PackEntry,
31417705×
    stored: Vec<u8>,
31517705×
    dict: Option<&Arc<zstd::dict::DecoderDictionary<'static>>>,
31617705×
) -> Result<Object> {
31717705×
    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}")))?
32213203×
    } else if e.flags & FLAG_ZSTD != 0 {
3237504×
        zstd::bulk::decompress(&stored, e.raw_len as usize)
3247504×
            .map_err(|err| StoreError::Corrupt(*r, format!("zstd: {err}")))?
325
    } else {
3265699×
        stored
327
    };
32817700×
    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×
        ));
33317581×
    }
33417581×
    Ok((e.ty, body))
33517705×
}
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.
341503×
    fn load(
342503×
        backend: &Backend,
343503×
        idx_key: &str,
344503×
        pack_key: &str,
345503×
        pack_len: u64,
346503×
    ) -> Result<Option<PackIndex>> {
347503×
        let data = match backend.get(idx_key) {
348479×
            Ok(d) => d,
34924×
            Err(BackendError::NotFound(_)) => return Ok(None),
3500×
            Err(e) => return Err(backend_err(e)),
351
        };
352479×
        let bad = |m: &str| StoreError::Corrupt(ObjectRef::ZERO, format!("{idx_key}: {m}"));
353479×
        if data.len() < 12 || &data[..8] != IDX_MAGIC {
3542×
            return Err(bad("bad index header"));
355477×
        }
356477×
        let count = u32::from_le_bytes(data[8..12].try_into().unwrap()) as usize;
357477×
        if data.len() != 12 + count * IDX_ENTRY {
35842×
            return Err(bad("index length mismatch"));
359435×
        }
360435×
        let mut refs = Vec::with_capacity(count);
361435×
        let mut entries = Vec::with_capacity(count);
36224498×
        for i in 0..count {
36324498×
            let e = &data[12 + i * IDX_ENTRY..12 + (i + 1) * IDX_ENTRY];
36424498×
            let r = ObjectRef::from_slice(&e[..32]).unwrap();
36524498×
            if let Some(prev) = refs.last() {
36624063×
                if r <= *prev {
3673×
                    return Err(bad("index not sorted"));
36824060×
                }
369435×
            }
37024495×
            refs.push(r);
37124490×
            let entry = PackEntry {
37224495×
                offset: u64::from_le_bytes(e[32..40].try_into().unwrap()),
37324495×
                stored_len: u32::from_le_bytes(e[40..44].try_into().unwrap()),
37424495×
                raw_len: u32::from_le_bytes(e[44..48].try_into().unwrap()),
37524495×
                ty: ObjectType::ALL
37624495×
                    .get(e[48] as usize)
37724495×
                    .copied()
37824495×
                    .ok_or_else(|| bad("unknown object type"))?,
37924490×
                flags: e[49],
380
            };
38124490×
            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×
                ));
38924487×
            }
39024487×
            if entry.offset.saturating_add(entry.stored_len as u64) > pack_len {
39154×
                return Err(bad("index entry points outside the pack"));
39224433×
            }
39324433×
            entries.push(entry);
394
        }
395370×
        let dict_key = format!("{}.dict", pack_key.trim_end_matches(".pack"));
396370×
        let dict = get_optional(backend, &dict_key)?
397370×
            .map(|d| Arc::new(zstd::dict::DecoderDictionary::copy(&d)));
398370×
        Ok(Some(PackIndex {
399370×
            key: pack_key.to_string(),
400370×
            refs,
401370×
            entries,
402370×
            dict,
403370×
        }))
404503×
    }
405
40623399×
    fn find(&self, r: &ObjectRef) -> Option<PackEntry> {
40723399×
        self.refs.binary_search(r).ok().map(|i| self.entries[i])
40823399×
    }
409
410
    /// Read one entry from the pack: one ranged read, then decode.
41117385×
    fn fetch(&self, backend: &Backend, r: &ObjectRef, e: PackEntry) -> Result<Object> {
41217385×
        let stored = backend
41317385×
            .get_range(&self.key, e.offset..e.offset + e.stored_len as u64)
41417385×
            .map_err(|err| match err {
415162×
                BackendError::NotFound(_) => StoreError::NotFound(*r),
4160×
                other => backend_err(other),
417162×
            })?;
41817223×
        decode_entry(r, e, stored, self.dict.as_ref())
41917385×
    }
420
}
421
42215844×
fn type_index(ty: ObjectType) -> u8 {
42322556×
    ObjectType::ALL.iter().position(|t| *t == ty).unwrap() as u8
42415844×
}
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<'_> {
5553296×
    pub fn finish(mut self) -> Result<()> {
5563296×
        self.done = true;
5573296×
        self.store.finish_batch()
5583296×
    }
559
}
560
561
impl Drop for Batch<'_> {
5623298×
    fn drop(&mut self) {
5633298×
        if !self.done {
5642×
            let _ = self.store.finish_batch();
5653296×
        }
5663298×
    }
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.
602126×
fn fresh_id() -> String {
603126×
    let mut nonce = [0u8; 8];
604126×
    rand::rngs::OsRng.fill_bytes(&mut nonce);
605126×
    hex::encode(nonce)
606126×
}
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.
6123188×
fn memory_backend_for(dir: &Path) -> Option<Arc<Backend>> {
613
    use std::collections::HashMap;
614
    use std::sync::{Mutex, OnceLock};
6153188×
    if std::env::var_os("OXY_STORE_BACKEND").as_deref() != Some(std::ffi::OsStr::new("memory")) {
6163188×
        return None;
6170×
    }
618
    static STORES: OnceLock<Mutex<HashMap<PathBuf, Arc<MemoryStore>>>> = OnceLock::new();
6190×
    let key = crate::portable::canonical(dir).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)
6293188×
}
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.
6413197×
    pub fn open(dir: impl Into<PathBuf>) -> io::Result<Store> {
6423197×
        let dir = dir.into();
6433197×
        fs::create_dir_all(dir.join("loose"))?;
644
        #[cfg(test)]
6451504×
        if let Some(backend) = TEST_BACKEND.with(|t| t.borrow().clone()) {
6469×
            return Store::open_on(backend, dir);
6471495×
        }
6483188×
        if let Some(backend) = memory_backend_for(&dir) {
6490×
            return Store::open_on(backend, dir);
6503188×
        }
6513188×
        let backend: Arc<Backend> =
6523188×
            Arc::new(LocalStore::open(&dir).map_err(|e| io::Error::other(e.to_string()))?);
6533188×
        let mut store = Store::open_on(backend, dir)?;
6543079×
        store.local = true;
6553079×
        Ok(store)
6563197×
    }
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.
6603276×
    pub fn open_on(backend: Arc<Backend>, scratch: impl Into<PathBuf>) -> io::Result<Store> {
6613276×
        let scratch = scratch.into();
6623276×
        fs::create_dir_all(&scratch)?;
6633276×
        let store = Store {
6643276×
            backend,
6653276×
            scratch,
6663276×
            local: false,
6673276×
            packs: RefCell::new(Vec::new()),
6683276×
            packs_listed: RefCell::new(Vec::new()),
6693276×
            has_relisted: std::cell::Cell::new(None),
6703276×
            pending: RefCell::new(None),
6713276×
            batch_depth: std::cell::Cell::new(0),
6723276×
            metas: RefCell::new(None),
6733276×
            entry_lists: RefCell::new(Vec::new()),
6743276×
            folded: std::cell::Cell::new(false),
6753276×
            legacy_folded: std::cell::Cell::new(false),
6763276×
            entries: RefCell::new(None),
6773276×
            entries_read: std::cell::Cell::new(None),
6783276×
            lists_read: RefCell::new(Default::default()),
6793276×
            journal_queue: RefCell::new(Vec::new()),
6803276×
        };
6813276×
        store
6823276×
            .reload_packs()
6833276×
            .map_err(|e| io::Error::other(e.to_string()))?;
6843167×
        Ok(store)
6853276×
    }
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.
69415305×
    pub fn is_local(&self) -> bool {
69515305×
        self.local
69615305×
    }
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.
701509×
    pub fn refresh(&self) -> Result<()> {
702509×
        *self.entries.borrow_mut() = None;
703509×
        self.reload_packs()
704509×
    }
705
706
    /// The pack listing, with how `packs_listed` records it.
7076508×
    fn list_packs(&self) -> Result<(Vec<strata::Entry>, Listed)> {
7086508×
        let listed = self.backend.list("pack-").map_err(backend_err)?;
7096508×
        let sig = listed
7106508×
            .iter()
7116508×
            .map(|e| (e.key.clone(), e.size, e.modified_ms))
7126508×
            .collect();
7136508×
        Ok((listed, sig))
7146508×
    }
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.
7204058×
    fn reload_packs(&self) -> Result<()> {
7214058×
        let (listed, sig) = self.list_packs()?;
7224058×
        self.load_listed(listed, sig)
7234058×
    }
724
7254067×
    fn load_listed(&self, mut listed: Vec<strata::Entry>, mut sig: Listed) -> Result<()> {
7264067×
        let had: std::collections::HashMap<String, Arc<PackIndex>> = self
7274067×
            .packs
7284067×
            .borrow()
7294067×
            .iter()
7304067×
            .map(|p| (p.key.clone(), p.clone()))
7314067×
            .collect();
7324067×
        let mut loaded: std::collections::HashMap<String, Arc<PackIndex>> = Default::default();
7334079×
        for attempt in 0..2 {
7344079×
            let retired = self.retired_in(&listed)?;
7354079×
            let mut idx_keys: Vec<&str> = listed
7364079×
                .iter()
7374079×
                .map(|e| e.key.as_str())
7384079×
                .filter(|k| k.ends_with(".idx"))
7394079×
                .collect();
7404079×
            idx_keys.sort();
7414079×
            let mut packs = Vec::new();
7424079×
            let mut vanished = false;
7434079×
            for idx in idx_keys {
7441506×
                let pack = format!("{}.pack", idx.trim_end_matches(".idx"));
7451506×
                if retired.contains_key(&pack) {
746174×
                    continue;
7471332×
                }
74811423×
                let Some(p) = listed.iter().find(|e| e.key == pack) else {
7490×
                    continue;
750
                };
7511332×
                if let Some(kept) = had.get(&pack).or_else(|| loaded.get(&pack)) {
752829×
                    packs.push(kept.clone());
753829×
                    continue;
754503×
                }
755503×
                match PackIndex::load(self.backend.as_ref(), idx, &pack, p.size)? {
756370×
                    Some(index) => {
757370×
                        let index = Arc::new(index);
758370×
                        loaded.insert(pack, index.clone());
759370×
                        packs.push(index);
760370×
                    }
76124×
                    None => vanished = true,
762
                }
763
            }
7643970×
            if vanished && attempt == 0 {
76512×
                (listed, sig) = self.list_packs()?;
76612×
                continue;
7673958×
            }
7683958×
            let lists: Vec<String> = listed
7693958×
                .iter()
7703958×
                .map(|e| e.key.as_str())
7713958×
                .filter(|k| is_entry_list(k) && !retired.contains_key(*k))
7723958×
                .map(str::to_string)
7733958×
                .collect();
7743958×
            self.folded
7753958×
                .set(lists.iter().any(|k| k.ends_with(LIST_SUFFIX)));
7763958×
            self.legacy_folded
7773958×
                .set(listed.iter().any(|e| e.key == LEGACY_FOLDED_KEY));
7783958×
            *self.entry_lists.borrow_mut() = lists;
7793958×
            if *self.packs_listed.borrow() != sig {
780541×
                *self.entries.borrow_mut() = None;
7813417×
            }
7823958×
            *self.packs.borrow_mut() = packs;
7833958×
            *self.packs_listed.borrow_mut() = sig;
7843958×
            return Ok(());
785
        }
7860×
        unreachable!("the second round always returns")
7874067×
    }
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.
793606×
    fn reload_if_changed(&self) -> Result<bool> {
794606×
        let (listed, sig) = self.list_packs()?;
795606×
        if *self.packs_listed.borrow() == sig {
796597×
            return Ok(false);
7979×
        }
7989×
        self.load_listed(listed, sig)?;
7999×
        Ok(true)
800606×
    }
801
802
    /// The retired record, read when `listed` (a pack listing) shows it
803
    /// holds anything.
8045911×
    fn retired_in(
8055911×
        &self,
8065911×
        listed: &[strata::Entry],
8075911×
    ) -> Result<std::collections::BTreeMap<String, u64>> {
8085911×
        match listed.iter().find(|e| e.key == RETIRED_KEY) {
809697×
            Some(e) if e.size > 0 => Ok(self.read_retired()?.0),
8105252×
            _ => Ok(Default::default()),
811
        }
8125911×
    }
813
814
    /// The retired record and its version (`None`: never written).
815671×
    fn read_retired(
816671×
        &self,
817671×
    ) -> Result<(
818671×
        std::collections::BTreeMap<String, u64>,
819671×
        Option<strata::Version>,
820671×
    )> {
821671×
        self.read_record(RETIRED_KEY)
822671×
    }
823
824
    /// A record of `<wall ms> <key>` lines (the retired record, the
825
    /// journal) and its version (`None`: never written).
8261143×
    fn read_record(
8271143×
        &self,
8281143×
        key: &str,
8291143×
    ) -> Result<(
8301143×
        std::collections::BTreeMap<String, u64>,
8311143×
        Option<strata::Version>,
8321143×
    )> {
8331143×
        match self.backend.read_ref(key).map_err(backend_err)? {
8341085×
            Some((bytes, v)) => Ok((parse_retired(&bytes), Some(v))),
83558×
            None => Ok((Default::default(), None)),
836
        }
8371143×
    }
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.
842129×
    fn update_retired(
843129×
        &self,
844129×
        edit: impl FnMut(&mut std::collections::BTreeMap<String, u64>),
845129×
    ) -> Result<()> {
846129×
        self.update_record(RETIRED_KEY, "the record of retired packs", edit)
847129×
    }
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.
852368×
    fn update_record(
853368×
        &self,
854368×
        key: &str,
855368×
        what: &str,
856368×
        mut edit: impl FnMut(&mut std::collections::BTreeMap<String, u64>),
857368×
    ) -> Result<()> {
858368×
        for _ in 0..RETIRED_TRIES {
859430×
            let (mut m, version) = self.read_record(key)?;
860430×
            let before = m.clone();
861430×
            edit(&mut m);
862
            // Unchanged; a record never written stays unwritten.
863430×
            if m == before && (version.is_some() || m.is_empty()) {
8642×
                return Ok(());
865428×
            }
866428×
            match self
867428×
                .backend
868428×
                .write_ref(key, &encode_retired(&m), version.as_ref())
869
            {
870366×
                Ok(_) => return Ok(()),
87162×
                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×
        ))))
878368×
    }
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.
884116×
    fn retire(&self, keys: &[String]) -> Result<()> {
885116×
        if keys.is_empty() {
88610×
            return Ok(());
887106×
        }
888106×
        let now = now_ms();
889106×
        self.update_retired(|m| {
8902558×
            for k in keys {
8912558×
                m.entry(k.clone()).or_insert(now);
8922558×
            }
893106×
        })
894116×
    }
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.
901818×
    pub fn expire_retired(
902818×
        &self,
903818×
        older_than: std::time::Duration,
904818×
        progress: Progress,
905818×
    ) -> Result<RetiredReport> {
906818×
        let (listed, _) = self.list_packs()?;
907818×
        let retired = self.retired_in(&listed)?;
908818×
        let mut report = RetiredReport::default();
909818×
        if retired.is_empty() {
910793×
            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)
949818×
    }
950
951
    /// The loose objects retired and not yet deleted, by key.
9521014×
    fn retired_loose(&self) -> Result<std::collections::HashSet<String>> {
9531014×
        let (listed, _) = self.list_packs()?;
9541014×
        Ok(self
9551014×
            .retired_in(&listed)?
9561014×
            .into_keys()
9575365×
            .filter(|k| k.starts_with("loose/"))
9581014×
            .collect())
9591014×
    }
960
96136×
    pub fn pack_count(&self) -> usize {
96236×
        self.packs.borrow().len()
96336×
    }
964
965111073×
    fn find_packed(&self, r: &ObjectRef) -> Option<(usize, PackEntry)> {
966111073×
        self.packs
967111073×
            .borrow()
968111073×
            .iter()
969111073×
            .enumerate()
970111073×
            .find_map(|(i, p)| p.find(r).map(|e| (i, e)))
971111073×
    }
972
973
    /// Loose objects, with their keys and ages; not those retired, whose
974
    /// objects a pack holds.
975980×
    fn loose_objects(&self) -> Result<Vec<(ObjectRef, strata::Entry)>> {
976980×
        let retired = self.retired_loose()?;
977980×
        let mut out = self.loose_objects_all()?;
97824602×
        out.retain(|(_, e)| !retired.contains(&e.key));
979980×
        Ok(out)
980980×
    }
981
982
    /// Every loose object, retired or not.
9831014×
    fn loose_objects_all(&self) -> Result<Vec<(ObjectRef, strata::Entry)>> {
9841014×
        let mut out: Vec<(ObjectRef, strata::Entry)> = self
9851014×
            .backend
9861014×
            .list("loose/")
9871014×
            .map_err(backend_err)?
9881014×
            .into_iter()
98925178×
            .filter_map(|e| loose_ref(&e.key).map(|r| (r, e)))
9901014×
            .collect();
9911014×
        out.sort_by_key(|(r, _)| *r);
9921014×
        Ok(out)
9931014×
    }
994
995798×
    pub fn loose_count(&self) -> io::Result<usize> {
996798×
        self.loose_objects()
997798×
            .map(|l| l.len())
998798×
            .map_err(|e| io::Error::other(e.to_string()))
999798×
    }
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.
102239×
    pub fn repack_all_with(&self, level: i32, progress: Progress) -> Result<(PackReport, usize)> {
102339×
        let loose = self.loose_objects()?;
102439×
        let (mut refs, keys, trained): (Vec<ObjectRef>, Vec<String>, bool) = {
102539×
            let packs = self.packs.borrow();
1026
            (
102739×
                packs.iter().flat_map(|p| p.refs.iter().copied()).collect(),
102839×
                packs.iter().map(|p| p.key.clone()).collect(),
102939×
                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.
103639×
        if loose.is_empty() && keys.len() <= 1 && (level == ZSTD_LEVEL || trained) {
10371×
            return Ok((PackReport::default(), 0));
103838×
        }
103938×
        refs.extend(loose.iter().map(|(r, _)| *r));
104038×
        let loose_keys: Vec<String> = loose.into_iter().map(|(_, e)| e.key).collect();
104138×
        let n = keys.len();
104238×
        let report = self.write_pack(&refs, &loose_keys, &keys, level, progress)?;
104338×
        Ok((report, n))
104439×
    }
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.
107834×
    pub fn prune_with(
107934×
        &self,
108034×
        keep: &std::collections::HashSet<ObjectRef>,
108134×
        grace: std::time::Duration,
108234×
        progress: Progress,
108334×
    ) -> Result<PruneReport> {
108434×
        progress("pruning loose objects");
108534×
        let now = now_ms();
108634×
        let grace_ms = grace.as_millis() as u64;
1087
        // An object whose age the store does not know is old enough.
108834×
        let old_enough =
108987×
            |modified: Option<u64>| modified.is_none_or(|t| now.saturating_sub(t) >= grace_ms);
109034×
        let mut report = PruneReport::default();
109134×
        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.
109434×
        let retired = self.retired_loose()?;
109534×
        let mut unretired: Vec<String> = Vec::new();
1096576×
        for (r, entry) in self.loose_objects_all()? {
1097576×
            if keep.contains(&r) {
1098498×
                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
        }
111634×
        if !unretired.is_empty() {
11176×
            self.update_retired(|m| {
111841×
                for k in &unretired {
111941×
                    m.remove(k);
112041×
                }
11216×
            })?;
112228×
        }
1123
        // Packs: rewrite those that hold unreachable objects.
112434×
        let mut keep_refs: Vec<ObjectRef> = Vec::new();
112534×
        let mut old_keys: Vec<String> = Vec::new();
112634×
        let mut rewrite = false;
112734×
        progress("pruning packs");
112834×
        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
        }
115334×
        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
            }
116728×
        }
116834×
        self.journal_forget(&gone)?;
116934×
        Ok(report)
117034×
    }
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).
119434×
    fn journal_forget(&self, gone: &[String]) -> Result<()> {
119534×
        if self.local || gone.is_empty() {
119632×
            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×
        })
120834×
    }
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.
1213111×
    fn finish_pack_files(
1214111×
        &self,
1215111×
        id: &str,
1216111×
        tmp_pack: &Path,
1217111×
        index: &[(ObjectRef, PackEntry)],
1218111×
    ) -> Result<()> {
1219111×
        let mut idx = Vec::with_capacity(12 + index.len() * IDX_ENTRY);
1220111×
        idx.extend_from_slice(IDX_MAGIC);
1221111×
        idx.extend_from_slice(&(index.len() as u32).to_le_bytes());
12227922×
        for (r, e) in index {
12237922×
            idx.extend_from_slice(&r.0);
12247922×
            idx.extend_from_slice(&e.offset.to_le_bytes());
12257922×
            idx.extend_from_slice(&e.stored_len.to_le_bytes());
12267922×
            idx.extend_from_slice(&e.raw_len.to_le_bytes());
12277922×
            idx.extend_from_slice(&[type_index(e.ty), e.flags]);
12287922×
        }
1229111×
        let put = self
1230111×
            .backend
1231111×
            .put_new_file(&format!("pack-{id}.pack"), tmp_pack)
1232111×
            .map_err(backend_err);
1233
        // A store on another disk copied it; the one in this directory
1234
        // moved it and this finds nothing.
1235111×
        let _ = fs::remove_file(tmp_pack);
1236111×
        put?;
1237111×
        self.backend
1238111×
            .put_new(&format!("pack-{id}.idx"), &idx)
1239111×
            .map_err(backend_err)?;
1240111×
        self.reload_packs()
1241111×
    }
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).
1256102×
    fn write_pack(
1257102×
        &self,
1258102×
        refs: &[ObjectRef],
1259102×
        loose_keys: &[String],
1260102×
        old_packs: &[String],
1261102×
        level: i32,
1262102×
        progress: Progress,
1263102×
    ) -> Result<PackReport> {
1264102×
        if refs.is_empty() {
12650×
            self.retire(old_packs)?;
12660×
            self.reload_packs()?;
12670×
            return Ok(PackReport::default());
1268102×
        }
1269102×
        let mut refs: Vec<ObjectRef> = refs.to_vec();
1270102×
        refs.sort();
1271102×
        refs.dedup();
1272102×
        let id = fresh_id();
1273102×
        let tmp_pack = self.scratch.join(format!(".tmp-{id}.pack"));
1274102×
        let mut report = PackReport::default();
1275102×
        let mut index: Vec<(ObjectRef, PackEntry)> = Vec::with_capacity(refs.len());
1276
        // A full repack of enough objects trains a dictionary for them.
1277102×
        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 {
1282100×
            None
1283
        };
1284102×
        let encoder = dict
1285102×
            .as_ref()
1286102×
            .map(|d| zstd::dict::EncoderDictionary::copy(d, level));
1287102×
        let nocompress = self.nocompress_set()?;
1288
        {
1289102×
            let mut f = io::BufWriter::new(fs::File::create(&tmp_pack)?);
1290102×
            f.write_all(PACK_MAGIC)?;
1291102×
            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.
1294102×
            let mut batches = refs.chunks(1024);
1295102×
            let mut compressed_batch = Vec::new();
1296102×
            let total = refs.len();
1297102×
            let mut reported = usize::MAX;
1298
            loop {
12994798×
                if compressed_batch.is_empty() {
1300206×
                    let Some(batch) = batches.next() else { break };
1301
                    // Every tenth, and the first batch.
1302104×
                    let tenth = report.objects * 10 / total.max(1);
1303104×
                    if tenth != reported {
1304104×
                        reported = tenth;
1305104×
                        progress(&format!("packing {} of {total} objects", report.objects));
1306104×
                    }
1307104×
                    let mut read: Vec<(ObjectRef, ObjectType, Vec<u8>, bool)> =
1308104×
                        Vec::with_capacity(batch.len());
13094696×
                    for r in batch {
13104696×
                        let (ty, body) = self.get(r)?;
13114696×
                        read.push((*r, ty, body, nocompress.contains(r)));
1312
                    }
1313
                    use rayon::prelude::*;
1314104×
                    compressed_batch = read
1315104×
                        .into_par_iter()
1316104×
                        .map_init(
13171516×
                            || {
13181516×
                                encoder.as_ref().and_then(|e| {
1319398×
                                    zstd::bulk::Compressor::with_prepared_dictionary(e).ok()
1320398×
                                })
13211516×
                            },
13224696×
                            |with_dict, (r, ty, body, raw)| {
13234696×
                                let packed = if raw {
13242×
                                    None
1325
                                } else {
1326
                                    // Small objects with the dictionary;
1327
                                    // a large one has context enough.
13284694×
                                    let dicted = with_dict
13294694×
                                        .as_mut()
13304694×
                                        .filter(|_| body.len() <= DICT_OBJECT_MAX)
13314694×
                                        .and_then(|c| c.compress(&body).ok())
13324694×
                                        .map(|c| (FLAG_ZSTD | FLAG_DICT, c));
13334694×
                                    dicted
13344694×
                                        .or_else(|| {
13351693×
                                            zstd::bulk::compress(&body, level)
13361693×
                                                .ok()
13371693×
                                                .map(|c| (FLAG_ZSTD, c))
13381693×
                                        })
13394694×
                                        .filter(|(_, c)| c.len() < body.len())
1340
                                };
13414696×
                                (r, ty, body, packed)
13424696×
                            },
1343
                        )
1344104×
                        .collect::<Vec<_>>();
1345104×
                    compressed_batch.reverse();
13464592×
                }
13474696×
                let Some((r, ty, body, packed)) = compressed_batch.pop() else {
13480×
                    continue;
1349
                };
13504696×
                let r = &r;
13514696×
                let (flags, stored) = match packed {
13523811×
                    Some(c) => c,
1353885×
                    None => (0, body.clone()),
1354
                };
13554696×
                let header_len = 1 + 1 + 4 + 4 + 32;
13564696×
                f.write_all(&[type_index(ty), flags])?;
13574696×
                f.write_all(&(stored.len() as u32).to_le_bytes())?;
13584696×
                f.write_all(&(body.len() as u32).to_le_bytes())?;
13594696×
                f.write_all(&r.0)?;
13604696×
                f.write_all(&stored)?;
13614696×
                index.push((
13624696×
                    *r,
13634696×
                    PackEntry {
13644696×
                        offset: offset + header_len,
13654696×
                        stored_len: stored.len() as u32,
13664696×
                        raw_len: body.len() as u32,
13674696×
                        ty,
13684696×
                        flags,
13694696×
                    },
13704696×
                ));
13714696×
                offset += header_len + stored.len() as u64;
13724696×
                report.objects += 1;
13734696×
                report.raw_bytes += body.len() as u64;
13744696×
                report.packed_bytes += stored.len() as u64;
1375
            }
1376102×
            f.flush()?;
1377102×
            f.get_ref().sync_all()?;
1378
        }
1379
        // The dictionary before the pack that needs it becomes visible.
1380102×
        if let Some(d) = &dict {
13812×
            self.backend
13822×
                .put_new(&format!("pack-{id}.dict"), d)
13832×
                .map_err(backend_err)?;
1384100×
        }
1385102×
        progress(&format!("writing the index of {} objects", index.len()));
1386102×
        self.finish_pack_files(&id, &tmp_pack, &index)?;
1387102×
        let mut retired = old_packs.to_vec();
1388102×
        retired.extend_from_slice(loose_keys);
1389102×
        self.retire(&retired)?;
1390102×
        self.reload_packs()?;
1391102×
        Ok(report)
1392102×
    }
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.
144313885×
    pub fn has(&self, r: &ObjectRef) -> bool {
144413885×
        if self.has_now(r) {
144513151×
            return true;
1446734×
        }
1447734×
        let recent = self
1448734×
            .has_relisted
1449734×
            .get()
1450734×
            .is_some_and(|t| t.elapsed() < HAS_RELIST_AFTER);
1451734×
        if recent {
1452185×
            return false;
1453549×
        }
1454549×
        match self.reload_if_changed() {
1455
            Ok(true) => {
14565×
                self.has_relisted.set(None);
14575×
                self.has_now(r)
1458
            }
1459
            _ => {
1460544×
                self.has_relisted.set(Some(std::time::Instant::now()));
1461544×
                false
1462
            }
1463
        }
146413885×
    }
1465
1466
    /// [`Store::has`] against the packs as loaded.
146713890×
    fn has_now(&self, r: &ObjectRef) -> bool {
146813890×
        self.find_packed(r).is_some()
146913297×
            || self
147013297×
                .pending
147113297×
                .borrow()
147213297×
                .as_ref()
147313297×
                .is_some_and(|p| p.lookup.contains_key(r))
147412969×
            || self.backend.has(&loose_key(r)).unwrap_or(false)
147513890×
    }
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.
14813299×
    pub fn begin_batch(&self) -> Result<Batch<'_>> {
14823299×
        let depth = self.batch_depth.get();
14833299×
        if depth == 0 {
14842667×
            *self.pending.borrow_mut() = Some(Pending {
14852667×
                buffered: vec![],
14862667×
                buffered_bytes: 0,
14872667×
                lookup: Default::default(),
14882667×
                pack: None,
14892667×
                report: PackReport::default(),
14902667×
            });
14912667×
        }
14923299×
        self.batch_depth.set(depth + 1);
14933299×
        Ok(Batch {
14943299×
            store: self,
14953299×
            done: false,
14963299×
        })
14973299×
    }
1498
14993298×
    fn finish_batch(&self) -> Result<()> {
15003298×
        let depth = self.batch_depth.get();
15013298×
        if depth == 0 {
15020×
            return Ok(());
15033298×
        }
15043298×
        self.batch_depth.set(depth - 1);
15053298×
        if depth > 1 {
1506632×
            return Ok(());
15072666×
        }
15082666×
        let Some(pending) = self.pending.borrow_mut().take() else {
15092×
            return self.flush_journal();
1510
        };
15112664×
        self.write_pending(pending)?;
15122662×
        self.flush_journal()
15133298×
    }
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.
15184068×
    pub fn flush(&self) -> Result<()> {
15194068×
        let Some(pending) = self.pending.borrow_mut().take() else {
15202897×
            return self.flush_journal();
1521
        };
15221171×
        self.write_pending(pending)?;
15231169×
        self.flush_journal()?;
15241169×
        *self.pending.borrow_mut() = Some(Pending {
15251169×
            buffered: vec![],
15261169×
            buffered_bytes: 0,
15271169×
            lookup: Default::default(),
15281169×
            pack: None,
15291169×
            report: PackReport::default(),
15301169×
        });
15311169×
        Ok(())
15324068×
    }
1533
15343835×
    fn write_pending(&self, pending: Pending) -> Result<()> {
15353835×
        match pending.pack {
1536
            None => {
15373827×
                let objects: Vec<(ObjectType, &[u8])> = pending
15383827×
                    .buffered
15393827×
                    .iter()
15406017×
                    .map(|(ty, body)| (*ty, body.as_slice()))
15413827×
                    .collect();
15423827×
                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
        }
15533835×
    }
1554
1555
    /// Whether a batch is open and holds `r`.
1556932×
    fn pending_get(&self, r: &ObjectRef) -> Option<Result<Object>> {
1557932×
        let mut guard = self.pending.borrow_mut();
1558932×
        let pending = guard.as_mut()?;
1559876×
        let entry = *pending.lookup.get(r)?;
1560876×
        Some(match entry {
1561694×
            PendingEntry::Buffered(i) => {
1562694×
                let (ty, body) = &pending.buffered[i];
1563694×
                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
        })
1581932×
    }
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
16069207×
    fn put_pending(&self, r: ObjectRef, ty: ObjectType, body: &[u8]) -> Result<()> {
16079207×
        let mut guard = self.pending.borrow_mut();
16089207×
        let pending = guard.as_mut().expect("batch open");
16099207×
        if pending.lookup.contains_key(&r) {
1610164×
            return Ok(());
16119043×
        }
16129043×
        if pending.pack.is_none()
16137979×
            && pending.buffered.len() < SPILL_OBJECTS
16147973×
            && pending.buffered_bytes + body.len() <= SPILL_BYTES
1615
        {
16167971×
            pending.buffered.push((ty, body.to_vec()));
16177971×
            pending.buffered_bytes += body.len();
16187971×
            pending
16197971×
                .lookup
16207971×
                .insert(r, PendingEntry::Buffered(pending.buffered.len() - 1));
16217971×
            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(())
16309207×
    }
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.
1665910×
    pub fn put_many(&self, objects: &[(ObjectType, &[u8])]) -> Result<Vec<ObjectRef>> {
1666
        use rayon::prelude::*;
16671118×
        let total: usize = objects.iter().map(|(_, b)| b.len()).sum();
1668910×
        if self.pending.borrow().is_none() || total < SPILL_BYTES {
1669916×
            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)
1727910×
    }
1728
1729
    /// Store `body` as an object of type `ty`; idempotent.
173019702×
    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.
173419702×
        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×
            });
174019701×
        }
174119701×
        let r = hash_object(ty, body);
174219701×
        if self.find_packed(&r).is_some() || self.backend.has(&loose_key(&r)).unwrap_or(false) {
17432248×
            return Ok(r);
174417453×
        }
174517453×
        if self.pending.borrow().is_some() {
17469207×
            self.put_pending(r, ty, body)?;
17479207×
            return Ok(r);
17488246×
        }
17498246×
        self.put_loose(ty, body)?;
17508241×
        Ok(r)
175119702×
    }
1752
1753
    /// Write one loose object, durable when this returns.
17548477×
    fn put_loose(&self, ty: ObjectType, body: &[u8]) -> Result<ObjectRef> {
17558477×
        let r = hash_object(ty, body);
17568477×
        let mut bytes = header(ty, body.len());
17578477×
        bytes.extend_from_slice(body);
17588477×
        crate::durable::step("write loose object")?;
17598472×
        put_new_or_existing(self.backend.as_ref(), &loose_key(&r), &bytes)?;
17608472×
        Ok(r)
17618477×
    }
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).
17673827×
    fn put_loose_many(&self, objects: &[(ObjectType, &[u8])]) -> Result<()> {
17683827×
        if objects.len() <= 1 {
17692233×
            for (ty, body) in objects {
1770231×
                self.put_loose(*ty, body)?;
1771
            }
17722233×
            return Ok(());
17731594×
        }
17741594×
        let mut seen = std::collections::HashSet::new();
17751594×
        let items: Vec<(String, Vec<u8>)> = objects
17761594×
            .iter()
17775786×
            .filter_map(|(ty, body)| {
17785786×
                let r = hash_object(*ty, body);
17795786×
                seen.insert(r).then(|| {
17805786×
                    let mut bytes = header(*ty, body.len());
17815786×
                    bytes.extend_from_slice(body);
17825786×
                    (loose_key(&r), bytes)
17835786×
                })
17845786×
            })
17851594×
            .collect();
17861594×
        self.backend.put_many_new(&items).map_err(backend_err)
17873827×
    }
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.
179974002×
    pub fn get(&self, r: &ObjectRef) -> Result<Object> {
180074002×
        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
            }
181773951×
            other => other,
1818
        }
181974002×
    }
1820
1821
    /// [`Store::get`] against the packs as loaded.
182274007×
    fn get_now(&self, r: &ObjectRef) -> Result<Object> {
182374007×
        if let Some((i, e)) = self.find_packed(r) {
182416317×
            let pack = self.packs.borrow()[i].clone();
182516317×
            return pack.fetch(self.backend.as_ref(), r, e);
182657690×
        }
182757690×
        match get_optional(self.backend.as_ref(), &loose_key(r))? {
182856765×
            Some(bytes) => parse_loose(&bytes, r),
1829925×
            None => self.pending_get(r).unwrap_or(Err(StoreError::NotFound(*r))),
1830
        }
183174007×
    }
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.
18381126×
    pub fn get_many(&self, refs: &[ObjectRef]) -> Vec<Result<Object>> {
18391126×
        let mut out = self.get_many_now(refs);
18401126×
        let missed: Vec<usize> = out
18411126×
            .iter()
18421126×
            .enumerate()
18432825×
            .filter(|(_, g)| g.as_ref().is_err_and(is_miss))
18441126×
            .map(|(i, _)| i)
18451126×
            .collect();
18461126×
        if missed.is_empty() {
18471117×
            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
18741126×
    }
1875
1876
    /// [`Store::get_many`] against the packs as loaded.
18771128×
    fn get_many_now(&self, refs: &[ObjectRef]) -> Vec<Result<Object>> {
1878
        use rayon::prelude::*;
18791128×
        let backend = self.backend.as_ref();
18801128×
        let packs = self.packs.borrow();
18811128×
        let place: Vec<Option<(usize, PackEntry)>> = refs
18821128×
            .iter()
18832985×
            .map(|r| {
18842985×
                packs
18852985×
                    .iter()
18862985×
                    .enumerate()
18872985×
                    .find_map(|(p, pack)| pack.find(r).map(|e| (p, e)))
18882985×
            })
18891128×
            .collect();
18901128×
        let indexes: &[Arc<PackIndex>] = &packs;
1891
        // Read in (pack, offset) order, each thread a run of it, so a
1892
        // spinning disk sweeps rather than seeks (oxygen#137: a clone's
1893
        // checkpoint reads tens of thousands of objects across ~95 packs).
18941128×
        let order = read_order(&place);
18951128×
        let fetched: Vec<(usize, Option<Result<Object>>)> = order
18961128×
            .par_iter()
18972985×
            .map(|&i| {
18982985×
                let r = &refs[i];
18992985×
                let got = match &place[i] {
19001067×
                    Some((p, e)) => Some(indexes[*p].fetch(backend, r, *e)),
19011918×
                    None => match get_optional(backend, &loose_key(r)) {
19021911×
                        Ok(Some(bytes)) => Some(parse_loose(&bytes, r)),
19037×
                        Ok(None) => None,
19040×
                        Err(e) => Some(Err(e)),
1905
                    },
1906
                };
19072985×
                (i, got)
19082985×
            })
19091128×
            .collect();
19101128×
        drop(packs);
19111128×
        let mut found: Vec<Option<Result<Object>>> = (0..refs.len()).map(|_| None).collect();
19122985×
        for (i, got) in fetched {
19132985×
            found[i] = got;
19142985×
        }
19151128×
        found
19161128×
            .into_iter()
19171128×
            .zip(refs)
19182985×
            .map(|(f, r)| {
19192985×
                f.unwrap_or_else(|| self.pending_get(r).unwrap_or(Err(StoreError::NotFound(*r))))
19202985×
            })
19211128×
            .collect()
19221128×
    }
1923
192453354×
    pub fn get_typed(&self, r: &ObjectRef, want: ObjectType) -> Result<Vec<u8>> {
192553354×
        let (ty, body) = self.get(r)?;
192653310×
        if ty != want {
19271×
            return Err(StoreError::WrongType(*r, ty.name(), want.name()));
192853309×
        }
192953309×
        Ok(body)
193053354×
    }
1931
1932
    /// Store a change and the text blobs (and their chunks) its large
1933
    /// inserts are stored as (spec §6.1).
19343389×
    pub fn put_change(&self, c: &Change) -> Result<ObjectRef> {
19353389×
        let encoding = crate::timing::phase("store: encode the change and its texts");
19363389×
        let bytes = c.encode_streaming(|batch| self.put_texts(&batch))?;
19373389×
        drop(encoding);
19383389×
        let _t = crate::timing::phase("store: put the encoding");
19393389×
        self.put(ObjectType::Change, &bytes)
19403389×
    }
1941
1942
    /// Store text blobs and their chunks: every chunk at once (compressed
1943
    /// on every core in a large batch), then the blobs.
194462×
    pub fn put_texts(&self, texts: &[crate::objects::StoredText]) -> Result<()> {
194562×
        let mut chunks: Vec<(ObjectType, &[u8])> = Vec::new();
194663×
        for t in texts {
194763×
            if !self.has(&t.blob.hash()) {
194852×
                chunks.extend(
194952×
                    crate::chunk::chunks(&t.content)
195052×
                        .into_iter()
195193×
                        .map(|c| (ObjectType::Chunk, c)),
1952
                );
195311×
            }
1954
        }
195562×
        self.put_many(&chunks)?;
195663×
        for t in texts {
195763×
            self.put(ObjectType::Blob, &t.blob.encode())?;
1958
        }
195962×
        Ok(())
196062×
    }
1961
19625443×
    pub fn get_change(&self, r: &ObjectRef) -> Result<Change> {
19635443×
        Ok(self.get_change_parts(r)?.0)
19645443×
    }
1965
1966
    /// A change and the objects its stored inserts are made of: each text
1967
    /// blob and its chunks, which travel and are kept with it.
19685609×
    pub fn get_change_parts(&self, r: &ObjectRef) -> Result<(Change, Vec<ObjectRef>)> {
19695609×
        let bytes = self.get_typed(r, ObjectType::Change)?;
19705596×
        let mut parts = Vec::new();
19715596×
        let mut failed = None;
19725596×
        let decoded = Change::decode_with(&bytes, &mut |text, count, form| match self
197359×
            .read_text(text)
1974
        {
197558×
            Ok((blob, content)) => {
197658×
                parts.push(*text);
197758×
                parts.extend(blob.chunks);
197858×
                crate::objects::text_lines(&content, count, form)
1979
            }
19801×
            Err(e) => {
19811×
                let why = e.to_string();
19821×
                failed = Some(e);
19831×
                Err(ObjectError::Schema(format!(
19841×
                    "text blob {}: {why}",
19851×
                    text.short(12)
19861×
                )))
1987
            }
198859×
        });
19895596×
        match (decoded, failed) {
19905595×
            (Ok(c), _) => Ok((c, parts)),
19911×
            (Err(_), Some(StoreError::NotFound(t))) => Err(StoreError::Corrupt(
19921×
                *r,
19931×
                format!("its text blob {} is missing", t.short(12)),
19941×
            )),
19950×
            (Err(e), _) => Err(StoreError::Corrupt(*r, e.to_string())),
1996
        }
19975609×
    }
1998
1999
    /// A change with every insert left without its lines (empty
2000
    /// placeholders), and where each one's lines are, in order: for
2001
    /// history-only states, which read text only for the graphs they build
2002
    /// (see [`Change::decode_skeleton`]).
2003585×
    pub fn get_change_lazy(&self, r: &ObjectRef) -> Result<(Change, Vec<crate::objects::TextRef>)> {
2004585×
        let bytes = self.get_typed(r, ObjectType::Change)?;
2005583×
        Change::decode_skeleton(&bytes, *r).map_err(|e| StoreError::Corrupt(*r, e.to_string()))
2006585×
    }
2007
2008
    /// A change's metadata (id, dependencies, author, clock, message, kind)
2009
    /// without its contents: read from its skeleton, so neither its text
2010
    /// blobs nor its inline lines are read (spec §11).
2011
    ///
2012
    /// Served from the `metas` cache when this store read it before (a
2013
    /// change object can be as large as the tree it adds), as long as the
2014
    /// store still has the change.
20154930×
    pub fn get_change_meta(&self, r: &ObjectRef) -> Result<crate::state::ChangeMeta> {
20164930×
        if self.metas.borrow().is_none() {
20171055×
            *self.metas.borrow_mut() = Some(self.read_metas_cache());
20183875×
        }
20194930×
        let cached = self.metas.borrow().as_ref().and_then(|m| m.get(r).cloned());
20204930×
        if let Some(meta) = cached {
20213559×
            if self.has(r) {
2022
                #[cfg(test)]
20232018×
                assert_eq!(
2024
                    meta,
20252018×
                    Change::decode_meta(&self.get_typed(r, ObjectType::Change)?).unwrap(),
2026
                    "the metadata cache holds a change's metadata"
2027
                );
20283559×
                return Ok(meta);
20290×
            }
20301371×
        }
20311371×
        let bytes = self.get_typed(r, ObjectType::Change)?;
20321371×
        let meta =
20331371×
            Change::decode_meta(&bytes).map_err(|e| StoreError::Corrupt(*r, e.to_string()))?;
20341371×
        self.remember_meta(r, &meta);
20351371×
        Ok(meta)
20364930×
    }
2037
2038
    /// The metadata cache: records of a change hash, its metadata in the
2039
    /// flat layout, and the first 8 bytes of BLAKE3 over both, each
2040
    /// appended in one write. Derived only: a record that does not check
2041
    /// ends the reading (a torn write), and the cache can be deleted. It
2042
    /// lives in the scratch directory, on this disk.
20431055×
    fn read_metas_cache(&self) -> crate::fasthash::FastMap<ObjectRef, crate::state::ChangeMeta> {
20441055×
        let mut out = crate::fasthash::FastMap::default();
20451055×
        let Ok(bytes) = fs::read(self.scratch.join("metas")) else {
2046410×
            return out;
2047
        };
2048645×
        let mut r = crate::flat::Reader::new(&bytes);
2049645×
        let mut good = 0;
20502262×
        while !r.at_end() {
20511618×
            let mut record = || -> std::result::Result<_, ObjectError> {
20521618×
                let h = r.object()?;
20531617×
                let body = r.bytes()?;
20541617×
                let check = r.fixed::<8>()?;
20551617×
                Ok((h, body, check))
20561618×
            };
20571618×
            let Ok((h, body, check)) = record() else {
20581×
                break;
2059
            };
20601617×
            let mut sum = blake3::Hasher::new();
20611617×
            sum.update(&h.0).update(body);
20621617×
            if sum.finalize().as_bytes()[..8] != check {
20630×
                break;
20641617×
            }
20651617×
            match crate::state::ChangeMeta::read_flat(&mut crate::flat::Reader::new(body)) {
20661617×
                Ok(m) => out.insert(h, m),
20670×
                Err(_) => break,
2068
            };
20691617×
            good += 32 + 4 + body.len() + 8;
2070
        }
2071
        // Cut a bad tail, so records appended later are read.
2072645×
        if good < bytes.len() {
20731×
            let _ = fs::OpenOptions::new()
20741×
                .write(true)
20751×
                .open(self.scratch.join("metas"))
20761×
                .and_then(|f| f.set_len(good as u64));
2077644×
        }
2078645×
        out
20791055×
    }
2080
2081
    /// Add a change's metadata to the cache; quietly nothing when it
2082
    /// cannot be written (it is only a cache).
20831371×
    fn remember_meta(&self, r: &ObjectRef, meta: &crate::state::ChangeMeta) {
20841371×
        let mut body = crate::flat::Writer::default();
20851371×
        meta.write_flat(&mut body);
20861371×
        let mut w = crate::flat::Writer::default();
20871371×
        w.object(r);
20881371×
        w.bytes(&body.0);
20891371×
        let mut sum = blake3::Hasher::new();
20901371×
        sum.update(&r.0).update(&body.0);
20911371×
        w.fixed(&sum.finalize().as_bytes()[..8]);
20921371×
        let _ = fs::OpenOptions::new()
20931371×
            .create(true)
20941371×
            .append(true)
20951371×
            .open(self.scratch.join("metas"))
20961371×
            .and_then(|mut f| f.write_all(&w.0));
20971371×
        if let Some(m) = self.metas.borrow_mut().as_mut() {
20981371×
            m.insert(*r, meta.clone());
20991371×
        }
21001371×
    }
2101
210259×
    fn read_text(&self, r: &ObjectRef) -> Result<(Blob, Vec<u8>)> {
210359×
        let blob = Blob::decode(&self.get_typed(r, ObjectType::Blob)?)?;
210458×
        let mut content = Vec::with_capacity(blob.length as usize);
210598×
        for c in &blob.chunks {
210698×
            content.extend(self.get_typed(c, ObjectType::Chunk)?);
2107
        }
210858×
        Ok((blob, content))
210959×
    }
2110
2111302×
    pub fn put_chunk(&self, bytes: &[u8]) -> Result<ObjectRef> {
2112302×
        self.put(ObjectType::Chunk, bytes)
2113302×
    }
2114
2115
    /// Store a chunk that packs will keep uncompressed (`nocompress`, spec §13.3).
2116123×
    pub fn put_chunk_nocompress(&self, bytes: &[u8]) -> Result<ObjectRef> {
2117123×
        let r = hash_object(ObjectType::Chunk, bytes);
2118123×
        put_new_or_existing(
2119123×
            self.backend.as_ref(),
2120123×
            &format!("nocompress/{}", r.to_hex()),
2121123×
            b"",
21220×
        )?;
2123123×
        self.put(ObjectType::Chunk, bytes)
2124123×
    }
2125
21263026×
    fn is_nocompress(&self, r: &ObjectRef) -> bool {
21273026×
        self.backend
21283026×
            .has(&format!("nocompress/{}", r.to_hex()))
21293026×
            .unwrap_or(false)
21303026×
    }
2131
2132
    /// Every chunk marked to be kept raw.
2133103×
    fn nocompress_set(&self) -> Result<std::collections::HashSet<ObjectRef>> {
2134103×
        Ok(self
2135103×
            .backend
2136103×
            .list("nocompress/")
2137103×
            .map_err(backend_err)?
2138103×
            .into_iter()
2139103×
            .filter_map(|e| ObjectRef::from_hex(e.key.trim_start_matches("nocompress/")))
2140103×
            .collect())
2141103×
    }
2142
21433427×
    pub fn put_state(&self, st: &StateObject) -> Result<ObjectRef> {
21443427×
        self.put(ObjectType::State, &st.encode())
21453427×
    }
2146
21476163×
    pub fn get_state(&self, r: &ObjectRef) -> Result<StateObject> {
21486163×
        Ok(StateObject::decode(&self.get_typed(r, ObjectType::State)?)?)
21496163×
    }
2150
2151542×
    pub fn put_attest(&self, a: &Attest) -> Result<ObjectRef> {
2152542×
        self.put(ObjectType::Attest, &a.encode())
2153542×
    }
2154
2155959×
    pub fn get_attest(&self, r: &ObjectRef) -> Result<Attest> {
2156959×
        Ok(Attest::decode(&self.get_typed(r, ObjectType::Attest)?)?)
2157959×
    }
2158
21593×
    pub fn put_receipt(&self, r: &Receipt) -> Result<ObjectRef> {
21603×
        self.put(ObjectType::Receipt, &r.encode())
21613×
    }
2162
216312×
    pub fn get_receipt(&self, r: &ObjectRef) -> Result<Receipt> {
216412×
        Ok(Receipt::decode(&self.get_typed(r, ObjectType::Receipt)?)?)
216512×
    }
2166
2167
    /// Record `value` under `key` in `index`: one write-once entry, so two
2168
    /// writers of a shared store never lose each other's.
21691453×
    fn index_add(&self, index: &str, key: &ObjectRef, value: &ObjectRef) -> Result<()> {
21703127×
        debug_assert!(ENTRY_INDEXES.iter().any(|(i, _)| *i == index));
21711453×
        let entry = format!("{}{}", fanout_prefix(index, key), value.to_hex());
21721453×
        put_new_or_existing(self.backend.as_ref(), &entry, b"")?;
21731453×
        self.journal_add(entry)
21741453×
    }
2175
2176
    /// The values recorded under `key` in `index`, with those an older
2177
    /// store kept as one list in `legacy`, sorted.
21782882×
    fn index_get(&self, index: &str, legacy: &str, key: &ObjectRef) -> Result<Vec<ObjectRef>> {
21792882×
        let mut values: Vec<ObjectRef> = self
21802882×
            .entry_keys(&fanout_prefix(index, key))?
21812882×
            .into_iter()
21822882×
            .filter_map(|k| k.rsplit('/').next().and_then(ObjectRef::from_hex))
21832882×
            .collect();
21842882×
        if !self.reads_legacy() {
218542×
            // Older lists folded into entries (spec §13.7).
21862840×
        } else if let Some(bytes) = get_optional(self.backend.as_ref(), &legacy_key(legacy, key))? {
21877×
            values.extend(
21887×
                String::from_utf8_lossy(&bytes)
21897×
                    .lines()
21907×
                    .filter_map(ObjectRef::from_hex),
21917×
            );
21922833×
        }
21932882×
        values.sort();
21942882×
        values.dedup();
21952882×
        Ok(values)
21962882×
    }
2197
2198
    /// Record that seal attestation `attest` covers `change`.
21991379×
    pub fn index_seal(&self, change: &ObjectRef, attest: &ObjectRef) -> Result<()> {
22001379×
        self.index_add("seal", change, attest)
22011379×
    }
2202
2203
    /// Seal attestations known to cover `change`.
22041616×
    pub fn seals_for(&self, change: &ObjectRef) -> Result<Vec<ObjectRef>> {
22051616×
        self.index_get("seal", "seals", change)
22061616×
    }
2207
2208
    /// Record receipt attestation `attest` for `state` (spec §14.6).
2209
    /// Keep a short-lived key's certificate (spec §14.7), with its issuer's
2210
    /// signature, indexed by the key it certifies.
221153×
    pub fn put_certificate(&self, c: &crate::cert::SignedCertificate) -> Result<ObjectRef> {
221253×
        let r = self.put(ObjectType::Certificate, &c.cert.encode())?;
221353×
        self.put_signature(&r, &c.sig)?;
221453×
        self.index_add("cert", &c.cert.key_id(), &r)?;
221553×
        Ok(r)
221653×
    }
2217
2218
    /// A kept certificate with its issuer's signature.
2219173×
    pub fn get_certificate(&self, r: &ObjectRef) -> Result<crate::cert::SignedCertificate> {
2220173×
        let cert = crate::cert::Certificate::decode(&self.get_typed(r, ObjectType::Certificate)?)?;
2221173×
        let sig = self
2222173×
            .signatures(r)?
2223173×
            .into_iter()
2224173×
            .find(|s| s.key == cert.issuer)
2225173×
            .ok_or(StoreError::NotFound(*r))?;
2226173×
        Ok(crate::cert::SignedCertificate { cert, sig })
2227173×
    }
2228
2229
    /// The certificate kept for short-lived key `key`, if any.
2230882×
    pub fn certificate_of(&self, key: &ObjectRef) -> Result<Option<ObjectRef>> {
2231882×
        Ok(self
2232882×
            .index_get("cert", "certs", key)?
2233882×
            .into_iter()
2234882×
            .find(|r| self.has(r)))
2235882×
    }
2236
2237
    /// Record `value` under `key` in index `index` (for repository code).
223813×
    pub(crate) fn index_add_pub(
223913×
        &self,
224013×
        index: &str,
224113×
        key: &ObjectRef,
224213×
        value: &ObjectRef,
224313×
    ) -> Result<()> {
224413×
        self.index_add(index, key, value)
224513×
    }
2246
2247
    /// The values recorded under `key` in index `index` (which has no
2248
    /// older one-list form: the legacy name only has to differ from it).
2249360×
    pub(crate) fn index_get_pub(&self, index: &str, key: &ObjectRef) -> Result<Vec<ObjectRef>> {
2250360×
        self.index_get(index, &format!("{index}-lists"), key)
2251360×
    }
2252
22536×
    pub fn index_receipt(&self, state: &ObjectRef, attest: &ObjectRef) -> Result<()> {
22546×
        self.index_add("receipt", state, attest)
22556×
    }
2256
2257
    /// Receipt attestations recorded for `state`.
225817×
    pub fn receipts_for(&self, state: &ObjectRef) -> Result<Vec<ObjectRef>> {
225917×
        self.index_get("receipt", "receipts", state)
226017×
    }
2261
22622×
    pub fn index_approval(&self, state: &ObjectRef, attest: &ObjectRef) -> Result<()> {
22632×
        self.index_add("approval", state, attest)
22642×
    }
2265
2266
    /// Approval attestations recorded for `state` (OxygenFlow approval
2267
    /// gates, oxygenhub#235).
22684×
    pub fn approvals_for(&self, state: &ObjectRef) -> Result<Vec<ObjectRef>> {
22694×
        self.index_get("approval", "approvals", state)
22704×
    }
2271
2272
    /// Every state with an approval, for garbage collection.
227326×
    pub fn approval_states(&self) -> Result<Vec<ObjectRef>> {
227426×
        self.indexed_states("approval", "approvals")
227526×
    }
2276
2277
    /// Every state with a receipt, for garbage collection.
227835×
    pub fn receipt_states(&self) -> Result<Vec<ObjectRef>> {
227935×
        self.indexed_states("receipt", "receipts")
228035×
    }
2281
2282
    /// The states an index (`one/<hh>/<rest>/<attest>` entries, and the
2283
    /// packed `many/<hh>/<rest>` lists) holds attestations for.
228461×
    fn indexed_states(&self, one: &str, many: &str) -> Result<Vec<ObjectRef>> {
228561×
        let mut out: Vec<ObjectRef> = Vec::new();
228661×
        for key in self.entry_keys(&format!("{one}/"))? {
2287
            // receipt/<hh>/<rest>/<attest>
22887×
            let mut parts = key.splitn(4, '/');
22897×
            let (Some(_), Some(hh), Some(rest), Some(_)) =
22907×
                (parts.next(), parts.next(), parts.next(), parts.next())
2291
            else {
22920×
                continue;
2293
            };
22947×
            out.extend(ObjectRef::from_hex(&format!("{hh}{rest}")));
2295
        }
229661×
        let legacy = if self.reads_legacy() {
229759×
            self.backend
229859×
                .list(&format!("{many}/"))
229959×
                .map_err(backend_err)?
2300
        } else {
23012×
            Vec::new()
2302
        };
230361×
        for e in legacy {
23042×
            let mut parts = e.key.splitn(3, '/');
23052×
            let (Some(_), Some(hh), Some(rest)) = (parts.next(), parts.next(), parts.next()) else {
23060×
                continue;
2307
            };
23082×
            out.extend(ObjectRef::from_hex(&format!("{hh}{rest}")));
2309
        }
231061×
        out.sort();
231161×
        out.dedup();
231261×
        Ok(out)
231361×
    }
2314
2315441×
    pub fn put_identity(&self, id: &Identity) -> Result<ObjectRef> {
2316441×
        self.put(ObjectType::Identity, &id.encode())
2317441×
    }
2318
23191134×
    pub fn get_identity(&self, r: &ObjectRef) -> Result<Identity> {
23201134×
        Ok(Identity::decode(&self.get_typed(r, ObjectType::Identity)?)?)
23211134×
    }
2322
2323
    /// Signatures stored alongside object `r` (spec §5.4), one per signing
2324
    /// key, sorted by key. Empty if none.
23253898×
    pub fn signatures(&self, r: &ObjectRef) -> Result<Vec<Signature>> {
23263898×
        let mut sigs: Vec<Signature> = Vec::new();
23273898×
        for key in self.entry_keys(&fanout_prefix("sig", r))? {
2328
            // Gone since it was listed or journalled: a collection deleted
2329
            // it with its object, or a new signature by the same signer is
2330
            // replacing it.
23312527×
            let Some(bytes) = get_optional(self.backend.as_ref(), &key)? else {
23320×
                continue;
2333
            };
23342527×
            sigs.extend(Signature::decode_list(&bytes)?);
2335
        }
2336
        // An older store's one list per object; a signer kept there and
2337
        // here counts once, as written here.
23383898×
        let legacy = if self.reads_legacy() {
23393854×
            get_optional(self.backend.as_ref(), &legacy_key("sigs", r))?
2340
        } else {
234144×
            None
2342
        };
23433898×
        if let Some(bytes) = legacy {
23444×
            for sig in Signature::decode_list(&bytes)? {
23454×
                if !sigs.iter().any(|s| s.key == sig.key) {
23463×
                    sigs.push(sig);
23473×
                }
2348
            }
23493894×
        }
23503898×
        sigs.sort_by_key(|s| s.key);
23513898×
        sigs.dedup_by_key(|s| s.key);
23523898×
        Ok(sigs)
23533898×
    }
2354
2355
    /// Attach a signature to object `r`, replacing any earlier one by the
2356
    /// same key: one entry per signer, so the latest by a key is the one
2357
    /// kept, and two signers never touch each other's.
2358864×
    pub fn put_signature(&self, r: &ObjectRef, sig: &Signature) -> Result<()> {
2359864×
        let key = format!("{}{}", fanout_prefix("sig", r), sig.key.to_hex());
2360864×
        let bytes = Signature::encode_list(std::slice::from_ref(sig));
2361864×
        if get_optional(self.backend.as_ref(), &key)?.as_deref() != Some(bytes.as_slice()) {
2362817×
            delete(self.backend.as_ref(), &key)?;
2363817×
            put_new_or_existing(self.backend.as_ref(), &key, &bytes)?;
236447×
        }
2365
        // Journalled even when the key was there: an older writer's
2366
        // entry is not in the journal or any list.
2367864×
        self.journal_add(key)
2368864×
    }
2369
2370
    // ---- the journal and entry lists (spec §13.7, #116)
2371
2372
    /// Whether lookups still read the one-list keys of older stores: until
2373
    /// a complete fold has copied them into per-object entries and the
2374
    /// pack listing shows its marker (spec §13.7).
23756841×
    fn reads_legacy(&self) -> bool {
23766841×
        !self.legacy_folded.get()
23776841×
    }
2378
2379
    /// Copy every older one-list key's contents into per-object entries,
2380
    /// keeping an entry already there, then write the marker that tells
2381
    /// readers to stop reading those keys. The keys themselves are left
2382
    /// for older readers. A list that cannot be read is passed over, and
2383
    /// said.
238410×
    fn fold_legacy(&self, progress: Progress) -> Result<()> {
238510×
        let mut copied = 0usize;
238660×
        for (index, legacy) in ENTRY_INDEXES {
238760×
            for e in self
238860×
                .backend
238960×
                .list(&format!("{legacy}/"))
239060×
                .map_err(backend_err)?
2391
            {
2392
                // <legacy>/<hh>/<rest>
23936×
                let mut parts = e.key.splitn(3, '/');
23946×
                let (Some(_), Some(hh), Some(rest)) = (parts.next(), parts.next(), parts.next())
2395
                else {
23960×
                    continue;
2397
                };
23986×
                let Some(r) = ObjectRef::from_hex(&format!("{hh}{rest}")) else {
23990×
                    continue;
2400
                };
24016×
                let Some(bytes) = get_optional(self.backend.as_ref(), &e.key)? else {
24020×
                    continue;
2403
                };
24046×
                if index == "sig" {
24052×
                    let Ok(sigs) = Signature::decode_list(&bytes) else {
24060×
                        progress(&format!("{}: not a list of signatures, passed over", e.key));
24070×
                        continue;
2408
                    };
24092×
                    for sig in sigs {
24102×
                        let key = format!("{}{}", fanout_prefix("sig", &r), sig.key.to_hex());
24112×
                        let one = Signature::encode_list(std::slice::from_ref(&sig));
24122×
                        put_new_or_existing(self.backend.as_ref(), &key, &one)?;
24132×
                        copied += 1;
2414
                    }
2415
                } else {
24164×
                    for v in String::from_utf8_lossy(&bytes)
24174×
                        .lines()
24184×
                        .filter_map(ObjectRef::from_hex)
2419
                    {
24204×
                        let key = format!("{}{}", fanout_prefix(index, &r), v.to_hex());
24214×
                        put_new_or_existing(self.backend.as_ref(), &key, b"")?;
24224×
                        copied += 1;
2423
                    }
2424
                }
2425
            }
2426
        }
242710×
        if copied > 0 {
24282×
            progress(&format!("{copied} entries copied out of older lists"));
24298×
        }
243010×
        put_new_or_existing(
243110×
            self.backend.as_ref(),
243210×
            LEGACY_FOLDED_KEY,
243310×
            format!("{}\n", now_ms()).as_bytes(),
24340×
        )?;
243510×
        self.legacy_folded.set(true);
243610×
        Ok(())
243710×
    }
2438
2439
    /// The entry keys under `prefix` (an index's `<index>/` or one
2440
    /// object's `<index>/<hh>/<rest>/`), sorted: on a folded store from
2441
    /// its lists and journal, with no request; otherwise by listing.
24426841×
    fn entry_keys(&self, prefix: &str) -> Result<Vec<String>> {
24436841×
        if let Some(keys) = self.with_entries(|set| {
244484×
            set.range(prefix.to_string()..)
24452300×
                .take_while(|k| k.starts_with(prefix))
244684×
                .cloned()
244784×
                .collect::<Vec<String>>()
244884×
        })? {
244984×
            return Ok(keys);
24506757×
        }
24516757×
        Ok(self
24526757×
            .backend
24536757×
            .list(prefix)
24546757×
            .map_err(backend_err)?
24556757×
            .into_iter()
24566757×
            .map(|e| e.key)
24576757×
            .collect())
24586841×
    }
2459
2460
    /// `f` over a folded store's entry keys, read on first use; `None`
2461
    /// when the store is not folded (or is this disk's directory), and
2462
    /// lookups list.
24636842×
    fn with_entries<T>(
24646842×
        &self,
24656842×
        f: impl FnOnce(&std::collections::BTreeSet<String>) -> T,
24666842×
    ) -> Result<Option<T>> {
24676842×
        if self.local || !self.folded.get() {
24686757×
            return Ok(None);
246985×
        }
2470
        // Read over the interval ago: list the packs again (a cloud
2471
        // client keeps that listing as long), loading what changed, and
2472
        // read the journal again, so another writer's entry is seen as
2473
        // its pack would be.
247485×
        if self
247585×
            .entries_read
247685×
            .get()
247785×
            .is_some_and(|t| t.elapsed() >= ENTRIES_FRESH)
2478
        {
24791×
            *self.entries.borrow_mut() = None;
24801×
            self.reload_if_changed()?;
24811×
            if !self.folded.get() {
24820×
                return Ok(None);
24831×
            }
248484×
        }
248585×
        if self.entries.borrow().is_none() {
248618×
            let Some(set) = self.read_entries()? else {
24870×
                return Ok(None);
2488
            };
248918×
            *self.entries.borrow_mut() = Some(set);
249067×
        }
249185×
        Ok(self.entries.borrow().as_ref().map(f))
24926842×
    }
2493
2494
    /// Every live entry list's keys and every journal line's, folded or
2495
    /// not, with what this handle has queued. A list deleted since the
2496
    /// packs were listed (retired over an hour ago) has the packs listed
2497
    /// again; `None` when one is still missing then, or the store is no
2498
    /// longer folded: lookups list.
249918×
    fn read_entries(&self) -> Result<Option<std::collections::BTreeSet<String>>> {
250019×
        'attempt: for attempt in 0..2 {
250119×
            if !self.folded.get() {
25020×
                return Ok(None);
250319×
            }
250419×
            let mut set = std::collections::BTreeSet::new();
250519×
            let lists = self.entry_lists.borrow().clone();
250619×
            self.lists_read
250719×
                .borrow_mut()
250819×
                .retain(|k, _| lists.contains(k));
250920×
            for key in &lists {
251020×
                let known = self.lists_read.borrow().get(key).cloned();
251120×
                let keys = match known {
25122×
                    Some(keys) => keys,
251318×
                    None => match get_optional(self.backend.as_ref(), key)? {
251417×
                        Some(bytes) => {
251517×
                            let keys = Arc::new(decode_entry_list(key, &bytes)?);
251617×
                            self.lists_read
251717×
                                .borrow_mut()
251817×
                                .insert(key.clone(), keys.clone());
251917×
                            keys
2520
                        }
25211×
                        None if attempt == 0 => {
25221×
                            self.reload_packs()?;
25231×
                            continue 'attempt;
2524
                        }
25250×
                        None => return Ok(None),
2526
                    },
2527
                };
252819×
                set.extend(keys.iter().cloned());
2529
            }
253018×
            set.extend(self.read_record(JOURNAL_KEY)?.0.into_keys());
253118×
            set.extend(self.journal_queue.borrow().iter().cloned());
253218×
            self.entries_read.set(Some(std::time::Instant::now()));
253318×
            return Ok(Some(set));
2534
        }
25350×
        Ok(None)
253618×
    }
2537
2538
    /// Add an entry key this handle wrote to the journal: at once, or
2539
    /// when the open batch is flushed. One the handle already knows from
2540
    /// the lists or the journal is not added again. Nothing on this
2541
    /// disk's directory.
25422317×
    fn journal_add(&self, key: String) -> Result<()> {
25432317×
        if self.local {
25441089×
            return Ok(());
25451228×
        }
25461228×
        if let Some(set) = self.entries.borrow_mut().as_mut() {
25470×
            if !set.insert(key.clone()) {
25480×
                return Ok(());
25490×
            }
25501228×
        }
25511228×
        if self.batch_depth.get() > 0 {
25521001×
            self.journal_queue.borrow_mut().push(key);
25531001×
            return Ok(());
2554227×
        }
2555227×
        self.journal_append(std::slice::from_ref(&key))
25562317×
    }
2557
2558
    /// Add what the open batch queued to the journal.
25596730×
    fn flush_journal(&self) -> Result<()> {
25606730×
        let queued = std::mem::take(&mut *self.journal_queue.borrow_mut());
25616730×
        self.journal_append(&queued)
25626730×
    }
2563
2564
    /// Add `keys` to the journal, unfolded, in one compare-and-swap
2565
    /// update, dropping lines folded over an hour ago; past
2566
    /// [`JOURNAL_FOLD_AT`] unfolded lines, fold them into a partial list.
25676957×
    fn journal_append(&self, keys: &[String]) -> Result<()> {
25686957×
        if keys.is_empty() {
25696729×
            return Ok(());
2570228×
        }
2571228×
        let now = now_ms();
2572228×
        let mut unfolded = 0;
2573290×
        self.update_record(JOURNAL_KEY, "the journal of entries", |m| {
2574290×
            expire_folded(m, now);
25751290×
            for k in keys {
25761290×
                m.entry(k.clone()).or_insert(0);
25771290×
            }
257827097×
            unfolded = m.values().filter(|t| **t == 0).count();
2579290×
        })?;
2580228×
        if unfolded > JOURNAL_FOLD_AT {
25811×
            self.fold_journal()?;
2582227×
        }
2583228×
        Ok(())
25846957×
    }
2585
2586
    /// Fold the journal's unfolded lines into a partial entry list, and
2587
    /// mark them folded. Lines added meanwhile stay unfolded.
25881×
    fn fold_journal(&self) -> Result<()> {
25891×
        let (journal, _) = self.read_record(JOURNAL_KEY)?;
25901×
        let keys: Vec<String> = journal
25911×
            .into_iter()
25921001×
            .filter(|(_, t)| *t == 0)
25931×
            .map(|(k, _)| k)
25941×
            .collect();
25951×
        if keys.is_empty() {
25960×
            return Ok(());
25971×
        }
25981×
        self.put_entry_list(PART_SUFFIX, keys.iter())?;
25991×
        self.mark_folded(&keys)
26001×
    }
2601
2602
    /// Mark these journal lines folded now, if they are not yet; drop
2603
    /// those folded over an hour ago.
26048×
    fn mark_folded(&self, keys: &[String]) -> Result<()> {
26058×
        let now = now_ms();
26068×
        self.update_record(JOURNAL_KEY, "the journal of entries", |m| {
26078×
            expire_folded(m, now);
26081214×
            for k in keys {
26091214×
                if let Some(t) = m.get_mut(k) {
26101214×
                    if *t == 0 {
26111214×
                        *t = now;
26121214×
                    }
26130×
                }
2614
            }
26158×
        })
26168×
    }
2617
2618
    /// Write a new entry list of `keys` (sorted), complete or partial by
2619
    /// `suffix`; its key.
262015×
    fn put_entry_list<'a>(
262115×
        &self,
262215×
        suffix: &str,
262315×
        keys: impl Iterator<Item = &'a String>,
262415×
    ) -> Result<String> {
262515×
        let key = format!("pack-{}{suffix}", fresh_id());
262615×
        let bytes = encode_entry_list(keys)?;
262715×
        self.backend.put_new(&key, &bytes).map_err(backend_err)?;
262815×
        Ok(key)
262915×
    }
2630
2631
    /// Fold every signature and index entry into one complete entry list
2632
    /// (spec §13.7), what a collection does last: first the older
2633
    /// one-list keys into per-object entries, once; then the journal's lines and
2634
    /// a listing of every entry prefix, written as a new list, every list
2635
    /// the packs' listing showed retired, and the journal lines read
2636
    /// marked folded. Nothing when the store is this disk's directory, or
2637
    /// when the one live list already holds exactly these entries and the
2638
    /// journal has nothing unfolded. Whether a list was written.
263940×
    pub fn fold_entries(&self, progress: Progress) -> Result<bool> {
264040×
        if self.local {
264125×
            return Ok(false);
264215×
        }
264315×
        self.reload_packs()?;
264415×
        if !self.legacy_folded.get() {
264510×
            progress("folding older lists of signatures and index entries");
264610×
            self.fold_legacy(progress)?;
26475×
        }
264815×
        let (journal, _) = self.read_record(JOURNAL_KEY)?;
264915×
        progress("folding signatures and index entries");
265015×
        let mut all: std::collections::BTreeSet<String> = journal.keys().cloned().collect();
265190×
        for (index, _) in ENTRY_INDEXES {
26521229×
            for e in self
265390×
                .backend
265490×
                .list(&format!("{index}/"))
265590×
                .map_err(backend_err)?
26561229×
            {
26571229×
                all.insert(e.key);
26581229×
            }
2659
        }
266015×
        let lists = self.entry_lists.borrow().clone();
266115×
        let unfolded: Vec<String> = journal
266215×
            .iter()
26631217×
            .filter(|(_, t)| **t == 0)
2664213×
            .map(|(k, _)| k.clone())
266515×
            .collect();
266615×
        if unfolded.is_empty() && lists.len() == 1 && lists[0].ends_with(LIST_SUFFIX) {
26673×
            if let Some(bytes) = get_optional(self.backend.as_ref(), &lists[0])? {
26683×
                let held: std::collections::BTreeSet<String> =
26693×
                    decode_entry_list(&lists[0], &bytes)?.into_iter().collect();
26703×
                if held == all {
26711×
                    return Ok(false);
26722×
                }
26730×
            }
267412×
        }
267514×
        self.put_entry_list(LIST_SUFFIX, all.iter())?;
267614×
        self.retire(&lists)?;
267714×
        if !unfolded.is_empty() {
26787×
            self.mark_folded(&unfolded)?;
26797×
        }
268014×
        *self.entries.borrow_mut() = None;
268114×
        self.reload_packs()?;
268214×
        Ok(true)
268340×
    }
2684
2685
    /// Finish packs a writer left unfinished: a batch writes its objects to
2686
    /// `.tmp-ID.pack` and only indexes it at the end, so a process that
2687
    /// stopped part way (an import killed after hours) leaves every object
2688
    /// in a pack nothing can read. Each such pack untouched for `min_age`
2689
    /// is read through: every whole entry checked against its reference,
2690
    /// a torn last one cut off, then indexed as a pack of the store.
2691
    /// Returns the packs and objects adopted.
26922×
    pub fn adopt_unfinished_packs(
26932×
        &self,
26942×
        min_age: std::time::Duration,
26952×
        progress: &dyn Fn(&str),
26962×
    ) -> Result<(usize, usize)> {
2697
        use std::io::{BufReader, Read};
26982×
        let mut packs = 0;
26992×
        let mut objects = 0;
27002×
        let Ok(dir) = fs::read_dir(&self.scratch) else {
27010×
            return Ok((0, 0));
2702
        };
27034×
        for entry in dir.flatten() {
27044×
            let name = entry.file_name().to_string_lossy().to_string();
27054×
            let Some(id) = name
27064×
                .strip_prefix(".tmp-")
27074×
                .and_then(|n| n.strip_suffix(".pack"))
2708
            else {
27092×
                continue;
2710
            };
27112×
            let path = entry.path();
27122×
            let meta = fs::metadata(&path)?;
27132×
            let age = meta
27142×
                .modified()
27152×
                .ok()
27162×
                .and_then(|m| m.elapsed().ok())
27172×
                .unwrap_or_default();
27182×
            if age < min_age {
27191×
                progress(&format!(
27201×
                    "{name}: written to {}s ago, left alone",
27211×
                    age.as_secs()
27221×
                ));
27231×
                continue;
27241×
            }
27251×
            let mut f = BufReader::with_capacity(1 << 20, fs::File::open(&path)?);
27261×
            let mut magic = [0u8; 8];
27271×
            if f.read_exact(&mut magic).is_err() || &magic != PACK_MAGIC {
27280×
                progress(&format!("{name}: not a pack, left alone"));
27290×
                continue;
27301×
            }
27311×
            let mut offset = PACK_MAGIC.len() as u64;
27321×
            let mut index: Vec<(ObjectRef, PackEntry)> = Vec::new();
2733
            loop {
2734301×
                let mut head = [0u8; 42];
2735301×
                if f.read_exact(&mut head).is_err() {
27360×
                    break;
2737301×
                }
2738301×
                let Some(ty) = ObjectType::ALL.get(head[0] as usize).copied() else {
27390×
                    break;
2740
                };
2741301×
                let flags = head[1];
2742301×
                let stored_len = u32::from_le_bytes(head[2..6].try_into().unwrap());
2743301×
                let raw_len = u32::from_le_bytes(head[6..10].try_into().unwrap());
2744301×
                let r = ObjectRef(head[10..42].try_into().unwrap());
2745301×
                let mut stored = vec![0u8; stored_len as usize];
2746301×
                if flags & FLAG_DICT != 0 || f.read_exact(&mut stored).is_err() {
27471×
                    break;
2748300×
                }
2749300×
                let e = PackEntry {
2750300×
                    offset: offset + 42,
2751300×
                    stored_len,
2752300×
                    raw_len,
2753300×
                    ty,
2754300×
                    flags,
2755300×
                };
2756300×
                if decode_entry(&r, e, stored, None).is_err() {
27570×
                    break;
2758300×
                }
2759300×
                index.push((r, e));
2760300×
                offset += 42 + stored_len as u64;
2761300×
                if index.len().is_multiple_of(200_000) {
27620×
                    progress(&format!(
27630×
                        "{name}: {} objects checked, {} MiB",
27640×
                        index.len(),
27650×
                        offset >> 20
27660×
                    ));
2767300×
                }
2768
            }
27691×
            if index.is_empty() {
27700×
                progress(&format!("{name}: no whole object, left alone"));
27710×
                continue;
27721×
            }
27731×
            if offset < meta.len() {
27741×
                progress(&format!(
27751×
                    "{name}: {} torn bytes at the end cut off",
27761×
                    meta.len() - offset
27771×
                ));
27781×
                fs::OpenOptions::new()
27791×
                    .write(true)
27801×
                    .open(&path)?
27811×
                    .set_len(offset)?;
27820×
            }
27831×
            index.sort_by_key(|(r, _)| *r);
27841×
            index.dedup_by_key(|(r, _)| *r);
27851×
            objects += index.len();
27861×
            packs += 1;
27871×
            progress(&format!("{name}: {} objects indexed", index.len()));
27881×
            self.finish_pack_files(id, &path, &index)?;
2789
        }
27902×
        Ok((packs, objects))
27912×
    }
2792
2793
    /// Every object reference in the store, loose and packed (slow; for gc and tests).
279489×
    pub fn list(&self) -> io::Result<Vec<ObjectRef>> {
279589×
        let mut out: Vec<ObjectRef> = self
279689×
            .loose_objects()
279789×
            .map_err(|e| io::Error::other(e.to_string()))?
279889×
            .into_iter()
279989×
            .map(|(r, _)| r)
280089×
            .collect();
280189×
        for p in self.packs.borrow().iter() {
280210×
            out.extend(p.refs.iter().copied());
280310×
        }
280489×
        out.sort();
280589×
        out.dedup();
280689×
        Ok(out)
280789×
    }
2808
}
2809
2810
/// Whether a read failed in a way a newer list of packs may cure: the
2811
/// object was not found, or its pack's dictionary is gone.
2812346×
fn is_miss(e: &StoreError) -> bool {
2813346×
    match e {
2814218×
        StoreError::NotFound(_) => true,
2815128×
        StoreError::Corrupt(_, why) => why == NO_DICT,
28160×
        _ => false,
2817
    }
2818346×
}
2819
2820
/// A loose object's bytes, verified against `r`.
2821
/// The order to read objects in: packed ones by pack then offset, loose
2822
/// ones after, each group keeping the asked order otherwise.
28231129×
fn read_order(place: &[Option<(usize, PackEntry)>]) -> Vec<usize> {
28241129×
    let mut order: Vec<usize> = (0..place.len()).collect();
282510282×
    order.sort_by_key(|&i| match &place[i] {
28266562×
        Some((p, e)) => (0, *p, e.offset),
28273720×
        None => (1, 0, 0),
282810282×
    });
28291129×
    order
28301129×
}
2831
283258676×
fn parse_loose(data: &[u8], r: &ObjectRef) -> Result<Object> {
283358676×
    let nul = data
283458676×
        .iter()
2835837477×
        .position(|&b| b == 0)
283658676×
        .ok_or_else(|| StoreError::Corrupt(*r, "no header".into()))?;
283758676×
    let head = std::str::from_utf8(&data[..nul])
283858676×
        .map_err(|_| StoreError::Corrupt(*r, "header not UTF-8".into()))?;
283958676×
    let mut parts = head.split(' ');
284058676×
    let (Some("oxy1"), Some(ty), Some(len), None) =
284158676×
        (parts.next(), parts.next(), parts.next(), parts.next())
2842
    else {
28430×
        return Err(StoreError::Corrupt(*r, format!("bad header {head:?}")));
2844
    };
284558676×
    let ty = ObjectType::from_name(ty)
284658676×
        .ok_or_else(|| StoreError::Corrupt(*r, format!("unknown type {ty}")))?;
284758676×
    let len: usize = len
284858676×
        .parse()
284958676×
        .map_err(|_| StoreError::Corrupt(*r, "bad length".into()))?;
285058676×
    let body = &data[nul + 1..];
285158676×
    if body.len() != len {
28520×
        return Err(StoreError::Corrupt(
28530×
            *r,
28540×
            format!("length {} != header {len}", body.len()),
28550×
        ));
285658676×
    }
285758676×
    if hash_object(ty, body) != *r {
28584×
        return Err(StoreError::Corrupt(*r, "hash mismatch".into()));
285958672×
    }
286058672×
    Ok((ty, body.to_vec()))
286158676×
}
2862
2863
/// Whether the objects are files that a test may tamper with: not under
2864
/// `OXY_STORE_BACKEND=memory`.
2865
#[cfg(test)]
28666×
fn files_on_disk() -> bool {
28676×
    std::env::var_os("OXY_STORE_BACKEND").is_none()
28686×
}
2869
2870
#[cfg(test)]
2871
mod tests {
2872
    use super::*;
2873
2874
    /// The store under test twice: in a directory, and on a store in
2875
    /// memory with a scratch directory, as over a bucket.
28766×
    fn each_store(dir: &Path, f: impl Fn(&str, Store)) {
28776×
        f("directory", Store::open(dir.join("objects")).unwrap());
28786×
        f(
28796×
            "memory",
28806×
            Store::open_on(Arc::new(MemoryStore::new()), dir.join("scratch")).unwrap(),
28816×
        );
28826×
    }
2883
2884
    #[test]
28851×
    fn put_get_verify() {
28861×
        if !super::files_on_disk() {
28870×
            return;
28881×
        }
28891×
        let dir = tempfile::tempdir().unwrap();
28901×
        let s = Store::open(dir.path().join("objects")).unwrap();
28911×
        let r = s.put(ObjectType::Chunk, b"hello").unwrap();
28921×
        assert!(s.has(&r));
28931×
        assert_eq!(s.get(&r).unwrap(), (ObjectType::Chunk, b"hello".to_vec()));
28941×
        assert_eq!(s.put(ObjectType::Chunk, b"hello").unwrap(), r);
28951×
        assert_eq!(s.list().unwrap(), vec![r]);
28961×
        assert!(matches!(
28971×
            s.get_typed(&r, ObjectType::Blob),
2898
            Err(StoreError::WrongType(..))
2899
        ));
29001×
        assert!(matches!(
29011×
            s.get(&ObjectRef([9; 32])),
2902
            Err(StoreError::NotFound(_))
2903
        ));
2904
2905
        // Tamper: flip a byte, read must fail.
29061×
        let path = s.path_for(&r);
29071×
        let mut data = fs::read(&path).unwrap();
29081×
        let last = data.len() - 1;
29091×
        data[last] ^= 1;
29101×
        fs::write(&path, data).unwrap();
29111×
        assert!(matches!(s.get(&r), Err(StoreError::Corrupt(..))));
29121×
    }
2913
2914
    #[test]
29151×
    fn packing_keeps_every_object_readable() {
29161×
        if !super::files_on_disk() {
29170×
            return;
29181×
        }
29191×
        let dir = tempfile::tempdir().unwrap();
29201×
        let s = Store::open(dir.path()).unwrap();
29211×
        let mut refs = Vec::new();
2922300×
        for i in 0..300u32 {
2923300×
            let body = format!("object {i} ").repeat(1 + (i % 7) as usize);
2924300×
            refs.push((s.put(ObjectType::Chunk, body.as_bytes()).unwrap(), body));
2925300×
        }
29261×
        let c = crate::objects::tests::sample_change();
29271×
        let cr = s.put_change(&c).unwrap();
29281×
        assert_eq!(s.loose_count().unwrap(), 301);
29291×
        let report = s.pack_loose().unwrap();
29301×
        assert_eq!(report.objects, 301);
29311×
        assert!(report.packed_bytes < report.raw_bytes, "{report:?}");
29321×
        assert_eq!(s.loose_count().unwrap(), 0);
29331×
        assert_eq!(s.pack_count(), 1);
2934300×
        for (r, body) in &refs {
2935300×
            assert!(s.has(r));
2936300×
            assert_eq!(
2937300×
                s.get(r).unwrap(),
2938300×
                (ObjectType::Chunk, body.as_bytes().to_vec())
2939
            );
2940
        }
29411×
        assert_eq!(s.get_change(&cr).unwrap(), c);
29421×
        assert_eq!(s.list().unwrap().len(), 301);
2943
        // Re-putting a packed object is a no-op; new objects go loose and a
2944
        // second pack coexists with the first.
29451×
        assert_eq!(
29461×
            s.put(ObjectType::Chunk, refs[0].1.as_bytes()).unwrap(),
29471×
            refs[0].0
2948
        );
29491×
        assert_eq!(s.loose_count().unwrap(), 0);
29501×
        let extra = s.put(ObjectType::Chunk, b"after packing").unwrap();
29511×
        s.pack_loose().unwrap();
29521×
        assert_eq!(s.pack_count(), 2);
29531×
        assert_eq!(s.get(&extra).unwrap().1, b"after packing");
2954
        // A reopened store sees the packs.
29551×
        let s2 = Store::open(dir.path()).unwrap();
29561×
        assert_eq!(s2.pack_count(), 2);
29571×
        assert_eq!(s2.get(&refs[5].0).unwrap().1, refs[5].1.as_bytes());
29581×
        assert!(matches!(
29591×
            s2.get(&ObjectRef([9; 32])),
2960
            Err(StoreError::NotFound(_))
2961
        ));
2962
        // Corrupting an object's body inside the pack is detected on read.
29631×
        let target = refs[5].0;
29641×
        let (pack_path, offset) = {
29651×
            let packs = s2.packs.borrow();
29661×
            let (i, e) = s2.find_packed(&target).unwrap();
29671×
            (dir.path().join(&packs[i].key), e.offset as usize)
29681×
        };
29691×
        let mut data = fs::read(&pack_path).unwrap();
29701×
        data[offset] ^= 0xff;
29711×
        fs::write(&pack_path, data).unwrap();
29721×
        let s3 = Store::open(dir.path()).unwrap();
29731×
        assert!(matches!(s3.get(&target), Err(StoreError::Corrupt(..))));
29741×
        assert_eq!(
29751×
            s3.get(&refs[6].0).unwrap().1,
29761×
            refs[6].1.as_bytes(),
2977
            "other objects are unaffected"
2978
        );
29791×
    }
2980
2981
    #[test]
29821×
    fn a_full_repack_trains_a_dictionary_and_reads_back() {
29831×
        if !super::files_on_disk() {
29840×
            return;
29851×
        }
29861×
        let dir = tempfile::tempdir().unwrap();
29871×
        let s = Store::open(dir.path()).unwrap();
2988
        // Many small, alike files: what a dictionary is for.
29891×
        let bodies: Vec<Vec<u8>> = (0..1500)
29901500×
            .map(|i| {
29911500×
                format!(
2992
                    "// SPDX-License-Identifier: GPL-2.0\n#include <linux/module.h>\n\
2993
                     static int probe_{i}(struct device *dev)\n{{\n\treturn {i};\n}}\n"
2994
                )
29951500×
                .into_bytes()
29961500×
            })
29971×
            .collect();
29981×
        let refs: Vec<ObjectRef> = bodies
29991×
            .iter()
30001500×
            .map(|b| s.put(ObjectType::Chunk, b).unwrap())
30011×
            .collect();
30021×
        s.repack_all(REPACK_ZSTD_LEVEL).unwrap();
30031×
        let dicts = fs::read_dir(dir.path())
30041×
            .unwrap()
30051×
            .flatten()
30065×
            .filter(|e| e.path().extension().is_some_and(|x| x == "dict"))
30071×
            .count();
30081×
        assert_eq!(dicts, 1, "the pack has a dictionary");
30091×
        let with_dict = refs
30101×
            .iter()
30111500×
            .filter(|r| s.find_packed(r).unwrap().1.flags & FLAG_DICT != 0)
30121×
            .count();
30131×
        assert!(with_dict > 1000, "most use it: {with_dict}");
30141×
        let again = Store::open(dir.path()).unwrap();
30151500×
        for (r, b) in refs.iter().zip(&bodies) {
30161500×
            assert_eq!(&again.get(r).unwrap().1, b);
3017
        }
30184×
        let packs = |dir: &Path| {
30194×
            let mut v: Vec<String> = fs::read_dir(dir)
30204×
                .unwrap()
30214×
                .flatten()
302222×
                .map(|e| e.file_name().to_string_lossy().into_owned())
302322×
                .filter(|n| n.starts_with("pack-"))
30244×
                .collect();
30254×
            v.sort();
30264×
            v
30274×
        };
3028
        // With nothing new, a repack leaves that pack as it is: it would
3029
        // only write the same again.
30301×
        let before = packs(dir.path());
30311×
        let (report, merged) = again.repack_all(REPACK_ZSTD_LEVEL).unwrap();
30321×
        assert_eq!((report.objects, merged), (0, 0));
30331×
        assert_eq!(packs(dir.path()), before);
3034
        // One new object: a repack replaces the pack and its dictionary,
3035
        // retiring the old ones, which a collection an hour on deletes.
30361×
        let extra = again.put(ObjectType::Chunk, b"one more\n").unwrap();
30371×
        again.repack_all(REPACK_ZSTD_LEVEL).unwrap();
30381×
        let files = packs(dir.path());
30391×
        assert_eq!(
30401×
            files.len(),
3041
            7,
3042
            "two of pack, index and dictionary, and the retired record: {files:?}"
3043
        );
30441×
        assert_eq!(again.pack_count(), 1, "the old pack is not read");
30451×
        assert_eq!(&again.get(&refs[7]).unwrap().1, &bodies[7]);
30461×
        assert_eq!(again.get(&extra).unwrap().1, b"one more\n");
30471×
        again
30481×
            .expire_retired(std::time::Duration::ZERO, &mut silent())
30491×
            .unwrap();
30501×
        let files = packs(dir.path());
30511×
        assert_eq!(files.len(), 4, "one pack, index, dictionary: {files:?}");
30521×
        assert!(before
30531×
            .iter()
30544×
            .filter(|f| *f != RETIRED_KEY)
30553×
            .all(|f| !files.contains(f)));
30561×
        assert_eq!(&again.get(&refs[8]).unwrap().1, &bodies[8]);
30571×
    }
3058
3059
    #[test]
30601×
    fn nocompress_chunks_pack_raw() {
30611×
        let dir = tempfile::tempdir().unwrap();
30621×
        let s = Store::open(dir.path()).unwrap();
30631×
        let body = vec![b'a'; 10_000];
30641×
        let raw = s.put_chunk_nocompress(&body).unwrap();
30651×
        let compressible = s.put(ObjectType::Chunk, &vec![b'b'; 10_000]).unwrap();
30661×
        s.pack_loose().unwrap();
30671×
        let packs = s.packs.borrow();
30681×
        let (i, e) = s.find_packed(&raw).unwrap();
30691×
        assert_eq!((e.flags, e.stored_len as usize), (0, 10_000));
30701×
        let (_, c) = s.find_packed(&compressible).unwrap();
30711×
        assert_eq!(c.flags, FLAG_ZSTD);
30721×
        assert!(c.stored_len < 10_000);
30731×
        assert_eq!(packs[i].fetch(s.backend.as_ref(), &raw, e).unwrap().1, body);
30741×
    }
3075
3076
    #[test]
30771×
    fn many_packs_merge_into_one() {
30781×
        let dir = tempfile::tempdir().unwrap();
30792×
        each_store(dir.path(), |which, s| {
30802×
            let mut refs = Vec::new();
308136×
            for i in 0..(MAX_PACKS + 2) {
308236×
                refs.push(
308336×
                    s.put(ObjectType::Chunk, format!("pack {i}").as_bytes())
308436×
                        .unwrap(),
308536×
                );
308636×
                s.pack_loose().unwrap();
308736×
            }
30882×
            assert_eq!(s.pack_count(), MAX_PACKS + 2, "{which}");
30892×
            assert_eq!(s.merge_packs().unwrap(), MAX_PACKS + 2);
30902×
            assert_eq!(s.pack_count(), 1);
30912×
            assert_eq!(s.merge_packs().unwrap(), 0);
309236×
            for (i, r) in refs.iter().enumerate() {
309336×
                assert_eq!(s.get(r).unwrap().1, format!("pack {i}").as_bytes());
3094
            }
30952×
            let again = Store::open_on(s.backend().clone(), s.dir()).unwrap();
30962×
            assert_eq!(again.pack_count(), 1, "{which}: reopened");
30972×
        });
30981×
    }
3099
3100
    #[test]
31011×
    fn prune_removes_unreachable_loose_and_packed_objects() {
3102
        use std::collections::HashSet;
31031×
        let dir = tempfile::tempdir().unwrap();
31042×
        each_store(dir.path(), |which, s| {
31052×
            let keep_a = s.put(ObjectType::Chunk, b"keep a").unwrap();
31062×
            let drop_a = s.put(ObjectType::Chunk, b"drop a").unwrap();
31072×
            s.pack_loose().unwrap();
31082×
            let keep_b = s.put(ObjectType::Chunk, b"keep b").unwrap();
31092×
            let drop_b = s.put(ObjectType::Chunk, b"drop b").unwrap();
31102×
            let sk = crate::identity::SecretKey {
31112×
                ed_seed: [1; 32],
31122×
                ml_seed: [2; 32],
31132×
            };
31142×
            let id = sk.identity("a", "b", 0);
31152×
            let key = s.put_identity(&id).unwrap();
31162×
            s.put_signature(
31172×
                &drop_b,
31182×
                &sk.sign(crate::identity::Domain::Change, &key, &drop_b),
3119
            )
31202×
            .unwrap();
31212×
            s.index_seal(&drop_a, &ObjectRef([3; 32])).unwrap();
31222×
            let keep: HashSet<ObjectRef> = [keep_a, keep_b, key].into_iter().collect();
3123
            // Everything is too young under a grace period.
31242×
            let r = s
31252×
                .prune(&keep, std::time::Duration::from_secs(3600))
31262×
                .unwrap();
31272×
            assert_eq!((r.objects, r.deferred), (0, 2), "{which}");
31282×
            let r = s.prune(&keep, std::time::Duration::ZERO).unwrap();
31292×
            assert_eq!(r.objects, 2, "{which}: {r:?}");
31302×
            assert!(s.has(&keep_a) && s.has(&keep_b) && s.has(&key));
31312×
            assert!(!s.has(&drop_a) && !s.has(&drop_b));
31322×
            assert!(s.signatures(&drop_b).unwrap().is_empty());
31332×
            assert!(s.seals_for(&drop_a).unwrap().is_empty());
31342×
            assert_eq!(s.pack_count(), 1);
31352×
            assert_eq!(s.get(&keep_a).unwrap().1, b"keep a");
31362×
            assert_eq!(s.list().unwrap().len(), 3);
31372×
        });
31381×
    }
3139
3140
    #[test]
31411×
    fn signatures_are_stored_beside_objects() {
3142
        use crate::identity::{Domain, SecretKey};
31431×
        let dir = tempfile::tempdir().unwrap();
31442×
        each_store(dir.path(), |which, s| {
31452×
            let r = s.put(ObjectType::Chunk, b"x").unwrap();
31462×
            assert!(s.signatures(&r).unwrap().is_empty());
31472×
            let sk = SecretKey {
31482×
                ed_seed: [1; 32],
31492×
                ml_seed: [2; 32],
31502×
            };
31512×
            let id = sk.identity("a", "b", 0);
31522×
            let key = s.put_identity(&id).unwrap();
31532×
            assert_eq!(s.get_identity(&key).unwrap(), id);
31542×
            let sig = sk.sign(Domain::Change, &key, &r);
31552×
            s.put_signature(&r, &sig).unwrap();
31562×
            s.put_signature(&r, &sig).unwrap();
31572×
            assert_eq!(s.signatures(&r).unwrap(), vec![sig.clone()], "{which}");
3158
            // A second signer is kept beside the first, sorted by key.
31592×
            let sk2 = SecretKey {
31602×
                ed_seed: [3; 32],
31612×
                ml_seed: [4; 32],
31622×
            };
31632×
            let key2 = s.put_identity(&sk2.identity("c", "d", 0)).unwrap();
31642×
            let sig2 = sk2.sign(Domain::Change, &key2, &r);
31652×
            s.put_signature(&r, &sig2).unwrap();
31662×
            let mut both = vec![sig.clone(), sig2.clone()];
31672×
            both.sort_by_key(|s| s.key);
31682×
            assert_eq!(s.signatures(&r).unwrap(), both);
3169
            // The object itself is untouched by its signatures.
31702×
            assert_eq!(s.get(&r).unwrap().1, b"x");
3171
            // Seal and receipt entries, and the states with receipts.
31722×
            s.index_seal(&r, &ObjectRef([5; 32])).unwrap();
31732×
            s.index_seal(&r, &ObjectRef([4; 32])).unwrap();
31742×
            s.index_seal(&r, &ObjectRef([5; 32])).unwrap();
31752×
            assert_eq!(
31762×
                s.seals_for(&r).unwrap(),
31772×
                vec![ObjectRef([4; 32]), ObjectRef([5; 32])]
3178
            );
31792×
            let st = ObjectRef([6; 32]);
31802×
            s.index_receipt(&st, &ObjectRef([7; 32])).unwrap();
31812×
            assert_eq!(s.receipts_for(&st).unwrap(), vec![ObjectRef([7; 32])]);
31822×
            assert_eq!(s.receipt_states().unwrap(), vec![st]);
31832×
        });
31841×
    }
3185
3186
    /// A store written before signatures, seals and receipts were one
3187
    /// entry each (`sigs/`, `seals/`, `receipts/` lists) still reads, and
3188
    /// what is written now sits beside what was there.
3189
    #[test]
31901×
    fn an_older_store_s_sidecar_lists_are_still_read() {
31911×
        if !super::files_on_disk() {
31920×
            return;
31931×
        }
3194
        use crate::identity::{Domain, SecretKey};
31951×
        let dir = tempfile::tempdir().unwrap();
31961×
        let s = Store::open(dir.path()).unwrap();
31971×
        let r = s.put(ObjectType::Chunk, b"old").unwrap();
31981×
        let h = r.to_hex();
31991×
        let sk = SecretKey {
32001×
            ed_seed: [1; 32],
32011×
            ml_seed: [2; 32],
32021×
        };
32031×
        let key = s.put_identity(&sk.identity("a", "b", 0)).unwrap();
32041×
        let sig = sk.sign(Domain::Change, &key, &r);
32051×
        let legacy = dir.path().join("sigs").join(&h[..2]);
32061×
        fs::create_dir_all(&legacy).unwrap();
32071×
        fs::write(
32081×
            legacy.join(&h[2..]),
32091×
            Signature::encode_list(std::slice::from_ref(&sig)),
3210
        )
32111×
        .unwrap();
32121×
        let seals = dir.path().join("seals").join(&h[..2]);
32131×
        fs::create_dir_all(&seals).unwrap();
32141×
        fs::write(
32151×
            seals.join(&h[2..]),
32161×
            format!("{}\n", ObjectRef([9; 32]).to_hex()),
3217
        )
32181×
        .unwrap();
32191×
        let receipts = dir.path().join("receipts").join(&h[..2]);
32201×
        fs::create_dir_all(&receipts).unwrap();
32211×
        fs::write(
32221×
            receipts.join(&h[2..]),
32231×
            format!("{}\n", ObjectRef([8; 32]).to_hex()),
3224
        )
32251×
        .unwrap();
3226
32271×
        assert_eq!(s.signatures(&r).unwrap(), vec![sig.clone()]);
32281×
        assert_eq!(s.seals_for(&r).unwrap(), vec![ObjectRef([9; 32])]);
32291×
        assert_eq!(s.receipts_for(&r).unwrap(), vec![ObjectRef([8; 32])]);
32301×
        assert_eq!(s.receipt_states().unwrap(), vec![r]);
3231
        // New entries join the old ones; the same signer is not doubled.
32321×
        s.put_signature(&r, &sig).unwrap();
32331×
        s.index_seal(&r, &ObjectRef([1; 32])).unwrap();
32341×
        assert_eq!(s.signatures(&r).unwrap().len(), 1);
32351×
        assert_eq!(
32361×
            s.seals_for(&r).unwrap(),
32371×
            vec![ObjectRef([1; 32]), ObjectRef([9; 32])]
3238
        );
3239
        // Pruning the object takes both layouts with it.
32401×
        let r2 = s.put(ObjectType::Chunk, b"kept").unwrap();
32411×
        let keep = [r2, key].into_iter().collect();
32421×
        s.prune(&keep, std::time::Duration::ZERO).unwrap();
32431×
        assert!(s.signatures(&r).unwrap().is_empty());
32441×
        assert!(s.seals_for(&r).unwrap().is_empty());
32451×
        assert!(s.receipt_states().unwrap().is_empty());
32461×
    }
3247
3248
    #[test]
32491×
    fn change_round_trip_through_store() {
32501×
        let dir = tempfile::tempdir().unwrap();
32512×
        each_store(dir.path(), |which, s| {
32522×
            let c = crate::objects::tests::sample_change();
32532×
            let r = s.put_change(&c).unwrap();
32542×
            assert_eq!(r, c.hash(), "{which}");
32552×
            assert_eq!(s.get_change(&r).unwrap(), c);
32562×
            assert_eq!(s.get_change_meta(&r).unwrap().message, c.message);
32572×
        });
32581×
    }
3259
32608057203×
    fn xorshift(state: &mut u64) -> u64 {
32618057203×
        *state ^= *state << 13;
32628057203×
        *state ^= *state >> 7;
32638057203×
        *state ^= *state << 17;
32648057203×
        *state
32658057203×
    }
3266
3267
    /// Random corruption of pack and index files never panics: reads fail
3268
    /// with an error or the store refuses to open, and untouched objects
3269
    /// stay readable when the index survives.
3270
    #[test]
32711×
    fn corrupted_packs_and_indexes_fail_cleanly() {
32721×
        if !super::files_on_disk() {
32730×
            return;
32741×
        }
32751×
        let dir = tempfile::tempdir().unwrap();
32761×
        let objects = dir.path().join("objects");
32771×
        let s = Store::open(&objects).unwrap();
32781×
        let mut refs = Vec::new();
32791×
        let mut seed = 0x9e3779b97f4a7c15u64;
328040×
        for i in 0..40u32 {
328140×
            let len = (xorshift(&mut seed) % 5000) as usize;
328240×
            let mut body = vec![0u8; len];
3283104686×
            for b in body.iter_mut() {
3284104686×
                *b = if i % 2 == 0 {
328556338×
                    (xorshift(&mut seed) & 0xff) as u8
3286
                } else {
328748348×
                    b'a' + (i % 26) as u8
3288
                };
3289
            }
329040×
            refs.push(s.put(ObjectType::Chunk, &body).unwrap());
3291
        }
32921×
        s.pack_loose().unwrap();
32931×
        let pack = fs::read_dir(&objects)
32941×
            .unwrap()
32953×
            .filter_map(|e| e.ok().map(|e| e.path()))
32963×
            .find(|p| p.extension().is_some_and(|x| x == "pack"))
32971×
            .unwrap();
32981×
        let idx = pack.with_extension("idx");
32991×
        let pristine_pack = fs::read(&pack).unwrap();
33001×
        let pristine_idx = fs::read(&idx).unwrap();
3301
3302300×
        for round in 0..300u32 {
3303300×
            let (path, pristine) = if round % 2 == 0 {
3304150×
                (&pack, &pristine_pack)
3305
            } else {
3306150×
                (&idx, &pristine_idx)
3307
            };
3308300×
            let mut data = pristine.clone();
3309300×
            match xorshift(&mut seed) % 4 {
331081×
                0 => {
331181×
                    let at = (xorshift(&mut seed) as usize) % data.len();
331281×
                    data[at] ^= 1 << (xorshift(&mut seed) % 8);
331381×
                }
331473×
                1 => {
331573×
                    let at = (xorshift(&mut seed) as usize) % data.len();
331673×
                    data[at] = (xorshift(&mut seed) & 0xff) as u8;
331773×
                }
331875×
                2 => {
331975×
                    let keep = (xorshift(&mut seed) as usize) % data.len();
332075×
                    data.truncate(keep);
332175×
                }
3322
                _ => {
332371×
                    let at = (xorshift(&mut seed) as usize) % data.len();
332471×
                    let n = ((xorshift(&mut seed) % 8) + 1) as usize;
3325324×
                    for k in 0..n.min(data.len() - at) {
3326324×
                        data[at + k] = 0xff;
3327324×
                    }
3328
                }
3329
            }
3330300×
            fs::write(path, &data).unwrap();
3331
            // Opening may refuse a broken index; a store that opens must
3332
            // either return each object intact or report it corrupt/missing.
3333300×
            if let Ok(s2) = Store::open(&objects) {
33347640×
                for r in &refs {
33357640×
                    match s2.get(r) {
33367517×
                        Ok((ty, body)) => assert_eq!(
33377517×
                            hash_object(ty, &body),
3338
                            *r,
3339
                            "a returned object always matches its reference"
3340
                        ),
3341
                        Err(StoreError::Corrupt(..))
3342
                        | Err(StoreError::NotFound(_))
3343123×
                        | Err(StoreError::Io(_)) => {}
33440×
                        Err(e) => panic!("unexpected error {e}"),
3345
                    }
3346
                }
3347109×
            }
3348300×
            fs::write(path, pristine).unwrap();
3349
        }
3350
        // Restored files read back in full.
33511×
        let s3 = Store::open(&objects).unwrap();
335240×
        for r in &refs {
335340×
            let (ty, body) = s3.get(r).unwrap();
335440×
            assert_eq!(hash_object(ty, &body), *r);
3355
        }
33561×
    }
3357
3358
    /// A small batch becomes loose objects; a large one becomes a single
3359
    /// pack. Objects are readable while pending, and batches nest.
3360
    #[test]
33611×
    fn batches_write_loose_or_one_pack() {
33621×
        let dir = tempfile::tempdir().unwrap();
33632×
        each_store(dir.path(), |which, s| {
33642×
            let outer = s.begin_batch().unwrap();
33652×
            let a = s.put(ObjectType::Chunk, b"small a").unwrap();
3366
            {
33672×
                let inner = s.begin_batch().unwrap();
33682×
                let b = s.put(ObjectType::Chunk, b"small b").unwrap();
33692×
                assert!(s.has(&b));
33702×
                assert_eq!(s.get(&b).unwrap().1, b"small b");
33712×
                inner.finish().unwrap();
33722×
                assert_eq!(
33732×
                    s.loose_count().unwrap(),
3374
                    0,
3375
                    "{which}: the inner batch defers to the outer one"
3376
                );
3377
            }
33782×
            assert_eq!(s.get(&a).unwrap().1, b"small a");
33792×
            assert_eq!(
33802×
                s.put(ObjectType::Chunk, b"small a").unwrap(),
3381
                a,
3382
                "idempotent while pending"
3383
            );
33842×
            outer.finish().unwrap();
33852×
            assert_eq!(s.loose_count().unwrap(), 2);
33862×
            assert_eq!(s.pack_count(), 0);
33872×
            assert_eq!(s.get(&a).unwrap().1, b"small a");
3388
3389
            // Large: spills into one pack, readable during and after.
33902×
            let big = s.begin_batch().unwrap();
33912×
            let mut refs = Vec::new();
33922×
            let mut seed = 7u64;
3393600×
            for i in 0..300u32 {
3394600×
                let mut body = vec![b'z'; 40_000];
3395600×
                if i % 3 == 0 {
33968000000×
                    for b in body.iter_mut() {
33978000000×
                        *b = (xorshift(&mut seed) & 0xff) as u8;
33988000000×
                    }
3399400×
                }
3400600×
                body[..4].copy_from_slice(&i.to_le_bytes());
3401600×
                let r = if i % 5 == 0 {
3402120×
                    s.put_chunk_nocompress(&body).unwrap()
3403
                } else {
3404480×
                    s.put(ObjectType::Chunk, &body).unwrap()
3405
                };
3406600×
                assert_eq!(
3407600×
                    s.get(&r).unwrap().1,
3408
                    body,
3409
                    "readable while pending (object {i})"
3410
                );
3411600×
                refs.push((r, body));
3412
            }
34132×
            assert!(s.has(&refs[0].0));
34142×
            big.finish().unwrap();
34152×
            assert_eq!(s.pack_count(), 1);
34162×
            assert_eq!(
34172×
                s.loose_count().unwrap(),
3418
                2,
3419
                "nothing from the big batch is loose"
3420
            );
34212×
            assert!(fs::read_dir(s.dir())
34222×
                .unwrap()
34232×
                .flatten()
34244×
                .all(|e| !e.file_name().to_string_lossy().contains("tmp-")));
34252×
            let s2 = Store::open_on(s.backend().clone(), s.dir()).unwrap();
3426600×
            for (r, body) in &refs {
3427600×
                assert_eq!(s2.get(r).unwrap().1, *body);
3428
            }
3429
            // A dropped batch still writes its objects.
34302×
            {
34312×
                let _b = s2.begin_batch().unwrap();
34322×
                s2.put(ObjectType::Chunk, b"dropped").unwrap();
34332×
            }
34342×
            assert_eq!(
34352×
                s2.get(&hash_object(ObjectType::Chunk, b"dropped"))
34362×
                    .unwrap()
3437
                    .1,
3438
                b"dropped"
3439
            );
34402×
        });
34411×
    }
3442
3443
    #[test]
34441×
    fn packed_objects_are_read_by_pack_then_offset() {
34454×
        let at = |p: usize, offset: u64| {
34464×
            Some((
34474×
                p,
34484×
                PackEntry {
34494×
                    offset,
34504×
                    stored_len: 1,
34514×
                    raw_len: 1,
34524×
                    ty: ObjectType::Chunk,
34534×
                    flags: 0,
34544×
                },
34554×
            ))
34564×
        };
34571×
        let place = vec![at(1, 50), None, at(0, 900), at(1, 10), None, at(0, 5)];
34581×
        assert_eq!(read_order(&place), [5, 2, 3, 0, 1, 4]);
34591×
    }
3460
3461
    #[test]
34621×
    fn get_many_reads_loose_and_packed_objects_in_order() {
34631×
        let dir = tempfile::tempdir().unwrap();
34642×
        each_store(dir.path(), |which, s| {
34652×
            let packed: Vec<ObjectRef> = (0..20)
346640×
                .map(|i| {
346740×
                    s.put(ObjectType::Chunk, format!("packed {i}").as_bytes())
346840×
                        .unwrap()
346940×
                })
34702×
                .collect();
34712×
            s.pack_loose().unwrap();
34722×
            let loose: Vec<ObjectRef> = (0..20)
347340×
                .map(|i| {
347440×
                    s.put(ObjectType::Chunk, format!("loose {i}").as_bytes())
347540×
                        .unwrap()
347640×
                })
34772×
                .collect();
34782×
            let missing = hash_object(ObjectType::Chunk, b"never stored");
34792×
            let mut refs: Vec<ObjectRef> = Vec::new();
348040×
            for i in 0..20 {
348140×
                refs.push(loose[i]);
348240×
                refs.push(packed[i]);
348340×
            }
34842×
            refs.push(missing);
34852×
            let got = s.get_many(&refs);
34862×
            assert_eq!(got.len(), refs.len());
348780×
            for (i, r) in refs.iter().enumerate().take(40) {
348880×
                let want = if i % 2 == 0 {
348940×
                    format!("loose {}", i / 2)
3490
                } else {
349140×
                    format!("packed {}", i / 2)
3492
                };
349380×
                let (ty, body) = got[i].as_ref().unwrap();
349480×
                assert_eq!(
349580×
                    (*ty, body.as_slice()),
349680×
                    (ObjectType::Chunk, want.as_bytes()),
3497
                    "{which}: {r:?}"
3498
                );
3499
            }
35002×
            assert!(matches!(got[40], Err(StoreError::NotFound(_))));
35012×
        });
35021×
    }
3503
3504
    /// A packed object is read from its pack without first asking the
3505
    /// backend for a loose copy (oxygenhub#228: on a bucket each such ask
3506
    /// was a request that found nothing, every time).
3507
    #[test]
35081×
    fn packed_objects_cost_no_loose_lookups() {
3509
        use std::sync::atomic::{AtomicUsize, Ordering};
3510
        struct Counting(MemoryStore, AtomicUsize);
3511
        impl crate::strata::Store for Counting {
35120×
            fn describe(&self) -> String {
35130×
                "counting".into()
35140×
            }
351510×
            fn has(&self, k: &str) -> crate::strata::Result<bool> {
351610×
                self.0.has(k)
351710×
            }
351815×
            fn get(&self, k: &str) -> crate::strata::Result<Vec<u8>> {
351915×
                if k.starts_with("loose/") {
352011×
                    self.1.fetch_add(1, Ordering::Relaxed);
352111×
                }
352215×
                self.0.get(k)
352315×
            }
352420×
            fn get_range(
352520×
                &self,
352620×
                k: &str,
352720×
                r: std::ops::Range<u64>,
352820×
            ) -> crate::strata::Result<Vec<u8>> {
352920×
                self.0.get_range(k, r)
353020×
            }
353112×
            fn put_new(&self, k: &str, b: &[u8]) -> crate::strata::Result<()> {
353212×
                self.0.put_new(k, b)
353312×
            }
35348×
            fn list(&self, p: &str) -> crate::strata::Result<Vec<crate::strata::Entry>> {
35358×
                self.0.list(p)
35368×
            }
35370×
            fn delete(&self, k: &str) -> crate::strata::Result<()> {
35380×
                self.0.delete(k)
35390×
            }
35403×
            fn read_ref(
35413×
                &self,
35423×
                k: &str,
35433×
            ) -> crate::strata::Result<Option<(Vec<u8>, crate::strata::Version)>> {
35443×
                self.0.read_ref(k)
35453×
            }
35461×
            fn write_ref(
35471×
                &self,
35481×
                k: &str,
35491×
                b: &[u8],
35501×
                e: Option<&crate::strata::Version>,
35511×
            ) -> crate::strata::Result<crate::strata::Version> {
35521×
                self.0.write_ref(k, b, e)
35531×
            }
3554
        }
35551×
        let dir = tempfile::tempdir().unwrap();
35561×
        let backend = Arc::new(Counting(MemoryStore::new(), AtomicUsize::new(0)));
35571×
        let s = Store::open_on(backend.clone(), dir.path().join("scratch")).unwrap();
35581×
        let refs: Vec<ObjectRef> = (0..10)
355910×
            .map(|i| {
356010×
                s.put(ObjectType::Chunk, format!("p {i}").as_bytes())
356110×
                    .unwrap()
356210×
            })
35631×
            .collect();
35641×
        s.pack_loose().unwrap();
35651×
        let s = Store::open_on(backend.clone(), dir.path().join("scratch")).unwrap();
35661×
        backend.1.store(0, Ordering::Relaxed);
356710×
        for r in &refs {
356810×
            assert!(s.get(r).is_ok());
3569
        }
357010×
        assert!(s.get_many(&refs).iter().all(|g| g.is_ok()));
35711×
        assert_eq!(backend.1.load(Ordering::Relaxed), 0, "no loose lookups");
3572
        // An object no pack holds is still looked for loose.
35731×
        let missing = hash_object(ObjectType::Chunk, b"never stored");
35741×
        assert!(matches!(s.get(&missing), Err(StoreError::NotFound(_))));
35751×
        assert!(backend.1.load(Ordering::Relaxed) > 0);
35761×
    }
3577
}
3578
3579
/// Readers never wait (oxygenhub#249): a store keeps reading while other
3580
/// handles repack, collect and push into the store under it.
3581
#[cfg(test)]
3582
pub(crate) mod reader_tests {
3583
    use super::*;
3584
35851682×
    fn chunk(i: usize) -> Vec<u8> {
35861682×
        format!("object {i} ").repeat(1 + i % 5).into_bytes()
35871682×
    }
3588
3589
    /// Two handles on one store: a writer, and a reader opened at once.
35904×
    fn each_pair(f: impl Fn(&str, Store, Store)) {
35914×
        let dir = tempfile::tempdir().unwrap();
35924×
        let s = Store::open(dir.path().join("objects")).unwrap();
35934×
        let r = Store::open(dir.path().join("objects")).unwrap();
35944×
        f("directory", s, r);
35954×
        let m: Arc<Backend> = Arc::new(MemoryStore::new());
35964×
        let s = Store::open_on(m.clone(), dir.path().join("scratch")).unwrap();
35974×
        let r = Store::open_on(m, dir.path().join("scratch-r")).unwrap();
35984×
        f("memory", s, r);
35994×
    }
3600
3601
    /// Set every retired entry's time back by `by`, as if retired then.
36024×
    pub(crate) fn age_retired(s: &Store, by: std::time::Duration) {
36034×
        s.update_retired(|m| {
360445×
            for t in m.values_mut() {
360545×
                *t = t.saturating_sub(by.as_millis() as u64);
360645×
            }
36074×
        })
36084×
        .unwrap();
36094×
    }
3610
361112×
    fn retired_keys(s: &Store) -> Vec<String> {
361212×
        s.read_retired().unwrap().0.into_keys().collect()
361312×
    }
3614
3615
    #[test]
36161×
    fn a_reader_opened_before_a_repack_reads_every_object_after_it() {
36172×
        each_pair(|which, writer, _| {
36182×
            let mut refs = Vec::new();
361980×
            for i in 0..40 {
362080×
                refs.push(writer.put(ObjectType::Chunk, &chunk(i)).unwrap());
362180×
            }
36222×
            writer.pack_loose().unwrap();
362380×
            for i in 40..80 {
362480×
                refs.push(writer.put(ObjectType::Chunk, &chunk(i)).unwrap());
362580×
            }
3626
            // Opened now: one pack and forty loose objects.
36272×
            let reader = Store::open_on(writer.backend().clone(), writer.dir()).unwrap();
36282×
            assert_eq!(reader.pack_count(), 1, "{which}");
36292×
            let (_, merged) = writer.repack_all(REPACK_ZSTD_LEVEL).unwrap();
36302×
            assert_eq!(merged, 1);
3631
            // The old pack and the loose objects are retired, not deleted.
36322×
            let retired = retired_keys(&writer);
36332×
            assert_eq!(retired.len(), 81, "{which}: {retired:?}");
36342×
            assert_eq!(writer.loose_count().unwrap(), 0, "retired is not loose");
3635
            // The reader reads on from what it loaded.
3636160×
            for (i, r) in refs.iter().enumerate() {
3637160×
                assert_eq!(reader.get(r).unwrap().1, chunk(i), "{which}: {i}");
3638
            }
36392×
            assert_eq!(reader.pack_count(), 1);
3640
            // A store opened now reads the new pack only.
36412×
            let fresh = Store::open_on(writer.backend().clone(), writer.dir()).unwrap();
36422×
            assert_eq!(fresh.pack_count(), 1);
36432×
            assert_ne!(fresh.packs.borrow()[0].key, reader.packs.borrow()[0].key);
3644
            // Once the retired files are deleted, the reader finds every
3645
            // object in the new pack, by get, get_many and has alike.
36462×
            let report = writer
36472×
                .expire_retired(std::time::Duration::ZERO, &mut silent())
36482×
                .unwrap();
36492×
            assert_eq!((report.packs, report.loose), (1, 80), "{which}");
36502×
            assert!(retired_keys(&writer).is_empty());
3651160×
            for (i, r) in refs.iter().enumerate() {
3652160×
                assert_eq!(reader.get(r).unwrap().1, chunk(i), "{which}: {i}");
3653
            }
36542×
            let other = Store::open_on(writer.backend().clone(), writer.dir()).unwrap();
36552×
            let before = other.packs.borrow()[0].key.clone();
36562×
            writer.put(ObjectType::Chunk, b"new").unwrap();
36572×
            writer.repack_all(REPACK_ZSTD_LEVEL).unwrap();
36582×
            writer
36592×
                .expire_retired(std::time::Duration::ZERO, &mut silent())
36602×
                .unwrap();
3661160×
            assert!(other.get_many(&refs).iter().all(|g| g.is_ok()), "{which}");
36622×
            assert_ne!(other.packs.borrow()[0].key, before);
36632×
            let third = Store::open_on(writer.backend().clone(), writer.dir()).unwrap();
36642×
            writer.put(ObjectType::Chunk, b"newer").unwrap();
36652×
            writer.repack_all(REPACK_ZSTD_LEVEL).unwrap();
36662×
            writer
36672×
                .expire_retired(std::time::Duration::ZERO, &mut silent())
36682×
                .unwrap();
3669160×
            assert!(refs.iter().all(|r| third.has(r)), "{which}");
36702×
            assert!(!third.has(&hash_object(ObjectType::Chunk, b"never")));
36712×
        });
36721×
    }
3673
3674
    #[test]
36751×
    fn a_reader_opened_before_a_spilled_batch_reads_its_objects() {
36762×
        each_pair(|which, writer, reader| {
36772×
            assert_eq!(reader.pack_count(), 0);
36782×
            let mut refs = Vec::new();
3679
            {
36802×
                let batch = writer.begin_batch().unwrap();
3681612×
                for i in 0..(SPILL_OBJECTS + 50) {
3682612×
                    refs.push(writer.put(ObjectType::Chunk, &chunk(i)).unwrap());
3683612×
                }
36842×
                batch.finish().unwrap();
3685
            }
36862×
            assert_eq!(writer.pack_count(), 1, "{which}: the batch spilled");
36872×
            assert_eq!(reader.get(&refs[0]).unwrap().1, chunk(0), "{which}");
36882×
            assert_eq!(reader.pack_count(), 1, "the miss loaded the new pack");
36892×
            let got = reader.get_many(&refs);
3690612×
            assert!(got.iter().all(|g| g.is_ok()), "{which}");
3691
            // has, too, from another reader.
36922×
            let late = Store::open_on(writer.backend().clone(), writer.dir()).unwrap();
3693
            {
36942×
                let batch = writer.begin_batch().unwrap();
3695514×
                for i in 1000..(1000 + SPILL_OBJECTS + 1) {
3696514×
                    writer.put(ObjectType::Chunk, &chunk(i)).unwrap();
3697514×
                }
36982×
                batch.finish().unwrap();
3699
            }
37002×
            assert!(late.has(&hash_object(ObjectType::Chunk, &chunk(1000))));
37012×
        });
37021×
    }
3703
3704
    #[test]
37051×
    fn retired_entries_expire_after_the_hour_and_are_never_loaded() {
37062×
        each_pair(|which, s, _| {
370720×
            for i in 0..10 {
370820×
                s.put(ObjectType::Chunk, &chunk(i)).unwrap();
370920×
            }
37102×
            s.pack_loose().unwrap();
37112×
            s.put(ObjectType::Chunk, &chunk(10)).unwrap();
37122×
            s.repack_all(ZSTD_LEVEL).unwrap();
3713
            // Retired over an hour ago: the first pack and 10 loose objects.
37142×
            age_retired(&s, RETIRED_FOR + std::time::Duration::from_secs(60));
37152×
            let old = retired_keys(&s);
37162×
            assert_eq!(old.len(), 12, "{which}: {old:?}");
3717
            // Retired just now: the second pack.
37182×
            s.put(ObjectType::Chunk, &chunk(11)).unwrap();
37192×
            s.repack_all(ZSTD_LEVEL).unwrap();
37202×
            assert_eq!(retired_keys(&s).len(), 14);
37212×
            let listed = s
37222×
                .backend()
37232×
                .list("pack-")
37242×
                .unwrap()
37252×
                .into_iter()
372614×
                .filter(|e| e.key.ends_with(".idx"))
37272×
                .count();
37282×
            assert_eq!(listed, 3, "{which}");
37292×
            let fresh = Store::open_on(s.backend().clone(), s.dir()).unwrap();
37302×
            assert_eq!(
37312×
                fresh.pack_count(),
3732
                1,
3733
                "{which}: retired packs are not loaded"
3734
            );
37352×
            let report = s.expire_retired(RETIRED_FOR, &mut silent()).unwrap();
37362×
            assert_eq!(
3737
                report,
3738
                RetiredReport {
3739
                    packs: 1,
3740
                    loose: 11,
3741
                    kept: 2
3742
                },
3743
                "{which}"
3744
            );
37452×
            let young = retired_keys(&s);
37462×
            assert_eq!(young.len(), 2, "{which}: {young:?}");
37474×
            assert!(young.iter().all(|k| !old.contains(k)));
374824×
            for k in old.iter().filter(|k| k.ends_with(".pack")) {
37492×
                assert!(!s.backend().has(k).unwrap(), "{which}: {k}");
3750
            }
37514×
            for k in &young {
37524×
                assert!(s.backend().has(k).unwrap(), "{which}: {k} kept");
3753
            }
375424×
            for i in 0..12 {
375524×
                let r = hash_object(ObjectType::Chunk, &chunk(i));
375624×
                assert_eq!(fresh.get(&r).unwrap().1, chunk(i), "{which}");
3757
            }
37582×
        });
37591×
    }
3760
3761
    /// A retired loose object whose pack copy was pruned since is not
3762
    /// lost when the retired files go: it is loose again.
3763
    #[test]
37641×
    fn a_retired_loose_object_no_pack_holds_is_kept() {
37652×
        each_pair(|which, s, _| {
37662×
            let a = s.put(ObjectType::Chunk, b"a").unwrap();
37672×
            let b = s.put(ObjectType::Chunk, b"b").unwrap();
37682×
            s.pack_loose().unwrap();
37692×
            assert_eq!(s.loose_count().unwrap(), 0);
3770
            // The pack rewritten without b, as pruning does.
37712×
            let first = s.packs.borrow()[0].key.clone();
37722×
            s.write_pack(&[a], &[], &[first], ZSTD_LEVEL, &mut silent())
37732×
                .unwrap();
37742×
            let report = s
37752×
                .expire_retired(std::time::Duration::ZERO, &mut silent())
37762×
                .unwrap();
37772×
            assert_eq!((report.packs, report.loose), (1, 1), "{which}: a's copy");
37782×
            assert!(retired_keys(&s).is_empty());
37792×
            assert_eq!(s.get(&b).unwrap().1, b"b", "{which}");
37802×
            assert_eq!(s.get(&a).unwrap().1, b"a");
37812×
            assert_eq!(s.loose_count().unwrap(), 1, "b is loose again");
3782
            // And pruning it, unreachable, deletes it.
37832×
            let keep = [a].into_iter().collect();
37842×
            let r = s.prune(&keep, std::time::Duration::ZERO).unwrap();
37852×
            assert_eq!(r.objects, 1);
37862×
            assert!(!s.has(&b));
37872×
        });
37881×
    }
3789
3790
    /// A backend whose first pack listing names an index that is deleted
3791
    /// before it is read, as a collection running meanwhile would.
3792
    pub(crate) struct Racing {
3793
        inner: MemoryStore,
3794
        lists: std::sync::atomic::AtomicUsize,
3795
        /// Deleted right after the first listing returns.
3796
        pub(crate) doomed: std::sync::Mutex<Option<String>>,
3797
        /// Named in every listing, never stored.
3798
        ghost: bool,
3799
    }
3800
3801
    impl crate::strata::Store for Racing {
38020×
        fn describe(&self) -> String {
38030×
            "racing".into()
38040×
        }
380511×
        fn has(&self, k: &str) -> crate::strata::Result<bool> {
380611×
            self.inner.has(k)
380711×
        }
380873×
        fn get(&self, k: &str) -> crate::strata::Result<Vec<u8>> {
380973×
            self.inner.get(k)
381073×
        }
38116×
        fn get_range(&self, k: &str, r: std::ops::Range<u64>) -> crate::strata::Result<Vec<u8>> {
38126×
            self.inner.get_range(k, r)
38136×
        }
381422×
        fn put_new(&self, k: &str, b: &[u8]) -> crate::strata::Result<()> {
381522×
            self.inner.put_new(k, b)
381622×
        }
381770×
        fn list(&self, p: &str) -> crate::strata::Result<Vec<crate::strata::Entry>> {
381870×
            let mut out = self.inner.list(p)?;
381970×
            if p == "pack-" {
382038×
                self.lists
382138×
                    .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
382238×
                if let Some(k) = self.doomed.lock().unwrap().take() {
38232×
                    self.inner.delete(&k)?;
382436×
                }
382538×
                if self.ghost {
382658×
                    for k in ["pack-0000.idx", "pack-0000.pack"] {
382758×
                        out.push(crate::strata::Entry {
382858×
                            key: k.into(),
382958×
                            size: 100,
383058×
                            modified_ms: None,
383158×
                        });
383258×
                    }
3833220×
                    out.sort_by(|a, b| a.key.cmp(&b.key));
38349×
                }
383532×
            }
383670×
            Ok(out)
383770×
        }
38380×
        fn delete(&self, k: &str) -> crate::strata::Result<()> {
38390×
            self.inner.delete(k)
38400×
        }
384129×
        fn read_ref(
384229×
            &self,
384329×
            k: &str,
384429×
        ) -> crate::strata::Result<Option<(Vec<u8>, crate::strata::Version)>> {
384529×
            self.inner.read_ref(k)
384629×
        }
38476×
        fn write_ref(
38486×
            &self,
38496×
            k: &str,
38506×
            b: &[u8],
38516×
            e: Option<&crate::strata::Version>,
38526×
        ) -> crate::strata::Result<crate::strata::Version> {
38536×
            self.inner.write_ref(k, b, e)
38546×
        }
3855
    }
3856
38573×
    pub(crate) fn racing(ghost: bool) -> Arc<Racing> {
38583×
        Arc::new(Racing {
38593×
            inner: MemoryStore::new(),
38603×
            lists: Default::default(),
38613×
            doomed: Default::default(),
38623×
            ghost,
38633×
        })
38643×
    }
3865
3866
    #[test]
38671×
    fn an_index_deleted_between_listing_and_loading_is_passed_over() {
38681×
        let dir = tempfile::tempdir().unwrap();
38691×
        let backend = racing(false);
38701×
        let s = Store::open_on(backend.clone(), dir.path()).unwrap();
38711×
        let a = s.put(ObjectType::Chunk, b"a").unwrap();
38721×
        s.pack_loose().unwrap();
38731×
        let b = s.put(ObjectType::Chunk, b"b").unwrap();
38741×
        s.pack_loose().unwrap();
38751×
        let idx = {
38761×
            let packs = s.packs.borrow();
38771×
            let (i, _) = s.find_packed(&a).unwrap();
38781×
            format!("{}.idx", packs[i].key.trim_end_matches(".pack"))
3879
        };
38801×
        *backend.doomed.lock().unwrap() = Some(idx);
38811×
        let before = backend.lists.load(std::sync::atomic::Ordering::Relaxed);
38821×
        let again = Store::open_on(backend.clone(), dir.path()).unwrap();
38831×
        assert_eq!(again.pack_count(), 1, "the deleted one is passed over");
38841×
        assert_eq!(
38851×
            backend.lists.load(std::sync::atomic::Ordering::Relaxed) - before,
3886
            2,
3887
            "and the store listed again"
3888
        );
38891×
        assert_eq!(again.get(&b).unwrap().1, b"b");
3890
        // A listing that names an index never there opens too.
38911×
        let ghosted = racing(true);
38921×
        let s = Store::open_on(ghosted.clone(), dir.path().join("g")).unwrap();
38931×
        let c = s.put(ObjectType::Chunk, b"c").unwrap();
38941×
        s.pack_loose().unwrap();
38951×
        let again = Store::open_on(ghosted, dir.path().join("g")).unwrap();
38961×
        assert_eq!(again.pack_count(), 1);
38971×
        assert_eq!(again.get(&c).unwrap().1, b"c");
38981×
    }
3899
}
3900
3901
#[cfg(test)]
3902
mod batch_tests {
3903
    use super::*;
3904
3905
    #[test]
39061×
    fn a_large_batch_packs_every_object_and_indexes_them_all() {
39071×
        let dir = tempfile::tempdir().unwrap();
39081×
        let store = Store::open(dir.path()).unwrap();
39091×
        let mut refs = Vec::new();
3910
        {
39111×
            let batch = store.begin_batch().unwrap();
39121000×
            for i in 0..1000u32 {
39131000×
                let body = format!("object {i} {}", "x".repeat(20_000 + i as usize)).into_bytes();
39141000×
                refs.push(store.put(ObjectType::Blob, &body).unwrap());
39151000×
            }
39161×
            batch.finish().unwrap();
3917
        }
39181×
        let again = Store::open(dir.path()).unwrap();
39191000×
        for r in &refs {
39201000×
            again
39211000×
                .get(r)
39221000×
                .unwrap_or_else(|e| panic!("{}: {e}", r.short(12)));
3923
        }
39241×
    }
3925
3926
    #[test]
39271×
    fn a_pack_a_stopped_batch_left_is_adopted_and_its_torn_end_cut() {
3928
        // oxygen#79: a batch never finished (a process killed part way)
3929
        // leaves its objects in an unindexed .tmp pack.
39301×
        let dir = tempfile::tempdir().unwrap();
39311×
        let mut refs = Vec::new();
3932
        {
39331×
            let store = Store::open(dir.path()).unwrap();
39341×
            let batch = store.begin_batch().unwrap();
3935300×
            for i in 0..300u32 {
3936300×
                let body = format!("object {i} {}", "y".repeat(20_000 + i as usize)).into_bytes();
3937300×
                refs.push(store.put(ObjectType::Blob, &body).unwrap());
3938300×
            }
39391×
            std::mem::forget(batch);
3940
        }
39411×
        let tmp: Vec<_> = fs::read_dir(dir.path())
39421×
            .unwrap()
39431×
            .flatten()
39442×
            .filter(|e| e.file_name().to_string_lossy().starts_with(".tmp-"))
39451×
            .map(|e| e.path())
39461×
            .collect();
39471×
        assert_eq!(tmp.len(), 1, "one unfinished pack");
3948
        // A torn last entry: a header and part of its body.
39491×
        let mut f = fs::OpenOptions::new().append(true).open(&tmp[0]).unwrap();
39501×
        f.write_all(&[0, 1, 0xff, 0, 0, 0, 9, 9, 0, 0]).unwrap();
39511×
        f.write_all(&[7; 40]).unwrap();
39521×
        drop(f);
39531×
        let store = Store::open(dir.path()).unwrap();
39541×
        assert!(store.get(&refs[0]).is_err(), "unreadable before");
39551×
        let young = store
39561×
            .adopt_unfinished_packs(std::time::Duration::from_secs(3600), &|_| {})
39571×
            .unwrap();
39581×
        assert_eq!(young, (0, 0), "a pack written to lately is left alone");
39591×
        let (packs, objects) = store
39602×
            .adopt_unfinished_packs(std::time::Duration::ZERO, &|_| {})
39611×
            .unwrap();
39621×
        assert_eq!((packs, objects), (1, 300));
3963300×
        for r in &refs {
3964300×
            store
3965300×
                .get(r)
3966300×
                .unwrap_or_else(|e| panic!("{}: {e}", r.short(12)));
3967
        }
39681×
        let again = Store::open(dir.path()).unwrap();
39691×
        assert!(again.get(&refs[299]).is_ok(), "indexed for good");
39701×
    }
3971
}
3972
3973
#[cfg(test)]
3974
mod size_tests {
3975
    use super::*;
3976
3977
    #[test]
39781×
    fn an_object_over_the_limit_is_refused_and_nothing_is_written() {
39791×
        let dir = tempfile::tempdir().unwrap();
39801×
        let store = Store::open(dir.path()).unwrap();
39811×
        let big = vec![b'x'; MAX_OBJECT_LEN as usize + 1];
3982
        {
39831×
            let batch = store.begin_batch().unwrap();
39841×
            store.put(ObjectType::Blob, b"small").unwrap();
39851×
            let e = store.put(ObjectType::Change, &big).unwrap_err();
39861×
            assert!(
39871×
                matches!(e, StoreError::TooLarge { kind: "change", .. }),
3988
                "{e}"
3989
            );
39901×
            assert!(e.to_string().contains("commit it in parts"), "{e}");
39911×
            batch.finish().unwrap();
3992
        }
3993
        // The store still opens and reads what was written.
39941×
        let again = Store::open(dir.path()).unwrap();
39951×
        again.get(&hash_object(ObjectType::Blob, b"small")).unwrap();
39961×
    }
3997
}
3998
3999
#[cfg(test)]
4000
mod text_tests {
4001
    use super::*;
4002
    use crate::objects::{AtomRef, Edit, CHANGE_VERSION};
4003
40042×
    fn large_change() -> Change {
40052×
        let mut c = crate::objects::tests::sample_change();
40062×
        assert_eq!(c.version, CHANGE_VERSION);
40072×
        c.edits[0] = Edit::Insert {
40082×
            node: crate::ids::NodeId([7; 16]),
40092×
            after: AtomRef::Start,
40102×
            before: AtomRef::End,
40112×
            lines: (0..2000)
40124000×
                .map(|i| format!("line {i}\n").into_bytes())
40132×
                .collect(),
4014
        };
40152×
        c
40162×
    }
4017
4018
    #[test]
40191×
    fn a_large_insert_is_stored_beside_its_change_and_read_back() {
40201×
        let dir = tempfile::tempdir().unwrap();
40211×
        let s = Store::open(dir.path()).unwrap();
40221×
        let c = large_change();
40231×
        let r = s.put_change(&c).unwrap();
40241×
        assert_eq!(r, c.hash());
40251×
        let (_, body) = s.get(&r).unwrap();
40261×
        assert!(
40271×
            body.len() < 1024,
4028
            "the change holds a reference, not {} bytes of lines",
40290×
            body.len()
4030
        );
40311×
        let (back, parts) = s.get_change_parts(&r).unwrap();
40321×
        assert_eq!(back, c);
4033
        // The text blob and its chunk.
40341×
        assert_eq!(parts.len(), 2);
40352×
        for p in &parts {
40362×
            assert!(s.has(p));
4037
        }
4038
        // Storing it again writes nothing new.
40391×
        let n = s.list().unwrap().len();
40401×
        s.put_change(&c).unwrap();
40411×
        assert_eq!(s.list().unwrap().len(), n);
40421×
    }
4043
4044
    #[test]
40451×
    fn a_change_whose_text_blob_is_missing_is_reported_not_misread() {
40461×
        if !super::files_on_disk() {
40470×
            return;
40481×
        }
40491×
        let dir = tempfile::tempdir().unwrap();
40501×
        let s = Store::open(dir.path()).unwrap();
40511×
        let c = large_change();
40521×
        let r = s.put_change(&c).unwrap();
40531×
        let (_, parts) = s.get_change_parts(&r).unwrap();
40541×
        fs::remove_file(s.path_for(&parts[0])).unwrap();
40551×
        let e = s.get_change(&r).unwrap_err();
40561×
        assert!(
40571×
            matches!(&e, StoreError::Corrupt(h, why) if *h == r && why.contains("missing")),
4058
            "{e}"
4059
        );
40601×
    }
4061
}
4062
4063
#[cfg(test)]
4064
mod put_many_tests {
4065
    use super::*;
4066
4067
    #[test]
40681×
    fn many_objects_in_a_batch_are_compressed_together_and_all_read_back() {
40691×
        let dir = tempfile::tempdir().unwrap();
40701×
        let s = Store::open(dir.path()).unwrap();
40711×
        let before = s.put(ObjectType::Chunk, b"already here").unwrap();
40721×
        let bodies: Vec<Vec<u8>> = (0..200u32)
4073200×
            .map(|i| format!("chunk {i} ").repeat(6000).into_bytes())
40741×
            .collect();
40751×
        let mut objects: Vec<(ObjectType, &[u8])> = bodies
40761×
            .iter()
4077200×
            .map(|b| (ObjectType::Chunk, b.as_slice()))
40781×
            .collect();
40791×
        objects.push((ObjectType::Chunk, bodies[0].as_slice()));
40801×
        objects.push((ObjectType::Chunk, b"already here"));
40811×
        let refs = {
40821×
            let batch = s.begin_batch().unwrap();
40831×
            let refs = s.put_many(&objects).unwrap();
40841×
            batch.finish().unwrap();
40851×
            refs
4086
        };
40871×
        assert_eq!(refs.len(), objects.len());
40881×
        assert_eq!(refs[200], refs[0], "a repeat is the same object");
40891×
        assert_eq!(refs[201], before);
40901×
        let again = Store::open(dir.path()).unwrap();
4091202×
        for (r, b) in refs.iter().zip(&objects) {
4092202×
            assert_eq!(again.get(r).unwrap().1, b.1);
4093
        }
40941×
    }
4095
}
4096
4097
/// Signatures and index entries on a store that is not this disk's
4098
/// directory: the journal and the entry lists (spec §13.7, #116).
4099
#[cfg(test)]
4100
mod entry_tests {
4101
    use super::*;
4102
    use crate::identity::{Domain, SecretKey};
4103
    use crate::strata::Store as _;
4104
    use std::collections::HashMap;
4105
    use std::sync::atomic::{AtomicBool, Ordering};
4106
    use std::sync::Mutex;
4107
4108
    /// A store in memory that counts its requests by kind, as a bucket
4109
    /// store's /health counters do; it can also slip a competing journal
4110
    /// line in just before a write of the journal, as another process
4111
    /// would.
4112
    #[derive(Default)]
4113
    struct Counted {
4114
        inner: MemoryStore,
4115
        counts: Mutex<HashMap<&'static str, usize>>,
4116
        interfere: AtomicBool,
4117
    }
4118
4119
    impl Counted {
41202403×
        fn count(&self, kind: &'static str) {
41212403×
            *self.counts.lock().unwrap().entry(kind).or_default() += 1;
41222403×
        }
412317×
        fn n(&self, kind: &str) -> usize {
412417×
            self.counts.lock().unwrap().get(kind).copied().unwrap_or(0)
412517×
        }
41269×
        fn reset(&self) {
41279×
            self.counts.lock().unwrap().clear();
41289×
        }
41292×
        fn total(&self) -> usize {
41302×
            self.counts.lock().unwrap().values().sum()
41312×
        }
41322×
        fn summary(&self) -> String {
41332×
            let c = self.counts.lock().unwrap();
41345×
            let mut v: Vec<String> = c.iter().map(|(k, n)| format!("{k} {n}")).collect();
41352×
            v.sort();
41362×
            v.join(", ")
41372×
        }
4138
    }
4139
4140
    impl crate::strata::Store for Counted {
41410×
        fn describe(&self) -> String {
41420×
            "counted".into()
41430×
        }
414447×
        fn has(&self, k: &str) -> crate::strata::Result<bool> {
414547×
            self.count("has");
414647×
            self.inner.has(k)
414747×
        }
4148120×
        fn get(&self, k: &str) -> crate::strata::Result<Vec<u8>> {
4149120×
            self.count(if is_legacy(k) { "get_legacy" } else { "get" });
4150120×
            self.inner.get(k)
4151120×
        }
41520×
        fn get_range(&self, k: &str, r: std::ops::Range<u64>) -> crate::strata::Result<Vec<u8>> {
41530×
            self.count("range");
41540×
            self.inner.get_range(k, r)
41550×
        }
41561300×
        fn put_new(&self, k: &str, b: &[u8]) -> crate::strata::Result<()> {
41571300×
            self.count("put");
41581300×
            self.inner.put_new(k, b)
41591300×
        }
4160271×
        fn list(&self, p: &str) -> crate::strata::Result<Vec<crate::strata::Entry>> {
4161271×
            self.count(if p == "pack-" {
416274×
                "list_packs"
4163197×
            } else if is_legacy(p) {
416455×
                "list_legacy"
4165
            } else {
4166142×
                "list"
4167
            });
4168271×
            self.inner.list(p)
4169271×
        }
417020×
        fn delete(&self, k: &str) -> crate::strata::Result<()> {
417120×
            self.count("delete");
417220×
            self.inner.delete(k)
417320×
        }
4174348×
        fn read_ref(
4175348×
            &self,
4176348×
            k: &str,
4177348×
        ) -> crate::strata::Result<Option<(Vec<u8>, crate::strata::Version)>> {
4178348×
            self.count(if k == JOURNAL_KEY {
4179332×
                "read_journal"
4180
            } else {
418116×
                "read_ref"
4182
            });
4183348×
            self.inner.read_ref(k)
4184348×
        }
4185297×
        fn write_ref(
4186297×
            &self,
4187297×
            k: &str,
4188297×
            b: &[u8],
4189297×
            e: Option<&crate::strata::Version>,
4190297×
        ) -> crate::strata::Result<crate::strata::Version> {
4191297×
            self.count("write_ref");
4192297×
            if k == JOURNAL_KEY && self.interfere.swap(false, Ordering::SeqCst) {
41931×
                let (mut m, v) = match self.inner.read_ref(k)? {
41940×
                    Some((bytes, v)) => (parse_retired(&bytes), Some(v)),
41951×
                    None => (Default::default(), None),
4196
                };
41971×
                m.insert(
41981×
                    format!("seal/{}", "c".repeat(64)) + "/" + &"d".repeat(64),
4199
                    0,
4200
                );
42011×
                self.inner.write_ref(k, &encode_retired(&m), v.as_ref())?;
4202296×
            }
4203297×
            self.inner.write_ref(k, b, e)
4204297×
        }
4205
    }
4206
4207
    /// Whether a key or prefix is under an older one-list layout.
4208317×
    fn is_legacy(k: &str) -> bool {
4209317×
        ENTRY_INDEXES
4210317×
            .iter()
42111539×
            .any(|(_, legacy)| k.starts_with(&format!("{legacy}/")))
4212317×
    }
4213
4214
    struct Fixture {
4215
        backend: Arc<Counted>,
4216
        dir: tempfile::TempDir,
4217
        handles: std::cell::Cell<usize>,
4218
    }
4219
4220
    impl Fixture {
42219×
        fn new() -> Fixture {
42229×
            Fixture {
42239×
                backend: Arc::new(Counted::default()),
42249×
                dir: tempfile::tempdir().unwrap(),
42259×
                handles: Default::default(),
42269×
            }
42279×
        }
4228
        /// Another handle on the store, as another process opens it.
422931×
        fn open(&self) -> Store {
423031×
            let n = self.handles.get() + 1;
423131×
            self.handles.set(n);
423231×
            Store::open_on(self.backend.clone(), self.dir.path().join(format!("s{n}"))).unwrap()
423331×
        }
4234
    }
4235
423616×
    fn signer(n: u8) -> SecretKey {
423716×
        SecretKey {
423816×
            ed_seed: [n; 32],
423916×
            ml_seed: [n + 100; 32],
424016×
        }
424116×
    }
4242
424314×
    fn sign(s: &Store, r: &ObjectRef, n: u8) -> Signature {
424414×
        let sk = signer(n);
424514×
        let key = s.put_identity(&sk.identity("a", "b", 0)).unwrap();
424614×
        let sig = sk.sign(Domain::Change, &key, r);
424714×
        s.put_signature(r, &sig).unwrap();
424814×
        sig
424914×
    }
4250
42518×
    fn journal(s: &Store) -> std::collections::BTreeMap<String, u64> {
42528×
        s.read_record(JOURNAL_KEY).unwrap().0
42538×
    }
4254
425514×
    fn fold(s: &Store) -> bool {
425614×
        s.fold_entries(&mut silent()).unwrap()
425714×
    }
4258
4259
    /// What 0.4.6 writes for a signature: its own key, and nothing else.
42602×
    fn old_writer_signs(f: &Fixture, r: &ObjectRef, n: u8) -> Signature {
42612×
        let sk = signer(n);
42622×
        let sig = sk.sign(Domain::Change, &ObjectRef([n; 32]), r);
42632×
        f.backend
42642×
            .put_new(
42652×
                &format!("{}{}", fanout_prefix("sig", r), sig.key.to_hex()),
42662×
                &Signature::encode_list(std::slice::from_ref(&sig)),
42672×
            )
42682×
            .unwrap();
42692×
        sig
42702×
    }
4271
4272
    /// After a fold, a signature or seal another handle writes is found
4273
    /// by a store opened later through the journal, with no listing; a
4274
    /// handle that already read the entries sees it once refreshed.
4275
    #[test]
42761×
    fn an_entry_written_after_a_fold_is_found_through_the_journal() {
42771×
        let f = Fixture::new();
42781×
        let w = f.open();
42791×
        let r1 = w.put(ObjectType::Chunk, b"one").unwrap();
42801×
        let s1 = sign(&w, &r1, 1);
42811×
        assert!(fold(&w));
42821×
        assert!(!fold(&w), "nothing new: no second list");
4283
42841×
        let early = f.open();
42851×
        assert_eq!(early.signatures(&r1).unwrap(), vec![s1.clone()]);
4286
42871×
        let w2 = f.open();
42881×
        let r2 = w2.put(ObjectType::Chunk, b"two").unwrap();
42891×
        let s2 = sign(&w2, &r2, 2);
42901×
        w2.index_seal(&r2, &ObjectRef([9; 32])).unwrap();
42911×
        w2.index_receipt(&r2, &ObjectRef([8; 32])).unwrap();
4292
        // Its own writes, at once.
42931×
        assert_eq!(w2.signatures(&r2).unwrap(), vec![s2.clone()]);
4294
42951×
        let later = f.open();
42961×
        f.backend.reset();
42971×
        assert_eq!(later.signatures(&r1).unwrap(), vec![s1]);
42981×
        assert_eq!(later.signatures(&r2).unwrap(), vec![s2.clone()]);
42991×
        assert_eq!(later.seals_for(&r2).unwrap(), vec![ObjectRef([9; 32])]);
43001×
        assert_eq!(later.receipts_for(&r2).unwrap(), vec![ObjectRef([8; 32])]);
43011×
        assert_eq!(later.receipt_states().unwrap(), vec![r2]);
43021×
        assert!(later.seals_for(&r1).unwrap().is_empty());
43031×
        assert_eq!(f.backend.n("list"), 0, "{}", f.backend.summary());
43041×
        assert_eq!(f.backend.n("read_journal"), 1, "{}", f.backend.summary());
4305
4306
        // The handle that read the entries before keeps its view until
4307
        // it is refreshed, as it keeps its operation head.
43081×
        assert!(early.signatures(&r2).unwrap().is_empty());
43091×
        early.refresh().unwrap();
43101×
        assert_eq!(early.signatures(&r2).unwrap(), vec![s2]);
43111×
    }
4312
4313
    /// A fold marks the journal's lines rather than dropping them, so a
4314
    /// reader whose listing predates it still finds every entry; an hour
4315
    /// on, the old list is deleted and the lines dropped, and readers
4316
    /// find everything in the new list.
4317
    #[test]
43181×
    fn a_fold_keeps_the_journal_for_the_hour_readers_may_need_it() {
43191×
        let f = Fixture::new();
43201×
        let w = f.open();
43211×
        let r1 = w.put(ObjectType::Chunk, b"one").unwrap();
43221×
        let s1 = sign(&w, &r1, 1);
43231×
        assert!(fold(&w));
43241×
        let r2 = w.put(ObjectType::Chunk, b"two").unwrap();
43251×
        let s2 = sign(&w, &r2, 2);
43262×
        assert!(journal(&w).values().any(|t| *t == 0));
4327
4328
        // Opened before the second fold: lists the first list only.
43291×
        let stale = f.open();
43301×
        let stale_after_expiry = f.open();
43311×
        assert!(fold(&w));
43321×
        assert!(
43332×
            journal(&w).values().all(|t| *t != 0),
4334
            "every line is marked folded"
4335
        );
43361×
        assert!(journal(&w).len() >= 2, "and kept");
43371×
        assert_eq!(stale.signatures(&r1).unwrap(), vec![s1.clone()]);
43381×
        assert_eq!(stale.signatures(&r2).unwrap(), vec![s2.clone()]);
4339
4340
        // An hour later: the first list goes with the next collection,
4341
        // and the next journal update drops the folded lines.
43421×
        reader_tests::age_retired(&w, RETIRED_FOR * 2);
43431×
        w.update_record(JOURNAL_KEY, "test", |m| {
43442×
            for t in m.values_mut() {
43452×
                *t = t.saturating_sub(2 * RETIRED_FOR.as_millis() as u64);
43462×
            }
43471×
        })
43481×
        .unwrap();
43491×
        w.expire_retired(RETIRED_FOR, &mut silent()).unwrap();
43501×
        let lists: Vec<String> = f
43511×
            .backend
43521×
            .inner
43531×
            .list("pack-")
43541×
            .unwrap()
43551×
            .into_iter()
43561×
            .map(|e| e.key)
43574×
            .filter(|k| is_entry_list(k))
43581×
            .collect();
43591×
        assert_eq!(lists.len(), 1, "the replaced list is deleted: {lists:?}");
43601×
        let r3 = w.put(ObjectType::Chunk, b"three").unwrap();
43611×
        let s3 = sign(&w, &r3, 3);
43621×
        let j = journal(&w);
43631×
        assert_eq!(j.len(), 1, "only the new line is left: {j:?}");
43641×
        assert!(j.keys().all(|k| k.starts_with(&fanout_prefix("sig", &r3))));
4365
4366
        // A handle whose listing named the deleted list lists again.
43671×
        assert_eq!(
43681×
            stale_after_expiry.signatures(&r1).unwrap(),
43691×
            vec![s1.clone()]
4370
        );
43711×
        assert_eq!(
43721×
            stale_after_expiry.signatures(&r3).unwrap(),
43731×
            vec![s3.clone()]
4374
        );
43751×
        let fresh = f.open();
43761×
        f.backend.reset();
43771×
        assert_eq!(fresh.signatures(&r1).unwrap(), vec![s1]);
43781×
        assert_eq!(fresh.signatures(&r2).unwrap(), vec![s2]);
43791×
        assert_eq!(fresh.signatures(&r3).unwrap(), vec![s3]);
43801×
        assert_eq!(f.backend.n("list"), 0);
43811×
    }
4382
4383
    /// A store no fold has touched (as 0.4.6 left it: per-object keys and
4384
    /// older one-list keys, no journal, no lists) reads by listing, and
4385
    /// reads the same once folded. An older writer's entry after the
4386
    /// fold is missed until the next fold: the trade-off of §13.7.
4387
    #[test]
43881×
    fn an_old_layout_store_reads_before_and_after_its_first_fold() {
43891×
        let f = Fixture::new();
43901×
        let w = f.open();
43911×
        let r1 = w.put(ObjectType::Chunk, b"one").unwrap();
43921×
        let r2 = w.put(ObjectType::Chunk, b"two").unwrap();
43931×
        let s1 = old_writer_signs(&f, &r1, 1);
43941×
        let s2 = SecretKey {
43951×
            ed_seed: [2; 32],
43961×
            ml_seed: [3; 32],
43971×
        }
43981×
        .sign(Domain::Change, &ObjectRef([2; 32]), &r2);
43991×
        f.backend
44001×
            .put_new(
44011×
                &legacy_key("sigs", &r2),
44021×
                &Signature::encode_list(std::slice::from_ref(&s2)),
44031×
            )
44041×
            .unwrap();
44051×
        f.backend
44061×
            .put_new(
44071×
                &format!("{}{}", fanout_prefix("seal", &r1), "ab".repeat(32)),
44081×
                b"",
44091×
            )
44101×
            .unwrap();
44111×
        f.backend
44121×
            .put_new(
44131×
                &legacy_key("receipts", &r2),
44141×
                format!("{}\n", ObjectRef([7; 32]).to_hex()).as_bytes(),
44151×
            )
44161×
            .unwrap();
44172×
        let check = |s: &Store| {
44182×
            assert_eq!(s.signatures(&r1).unwrap(), vec![s1.clone()]);
44192×
            assert_eq!(s.signatures(&r2).unwrap(), vec![s2.clone()]);
44202×
            assert_eq!(s.seals_for(&r1).unwrap(), vec![ObjectRef([0xab; 32])]);
44212×
            assert_eq!(s.receipts_for(&r2).unwrap(), vec![ObjectRef([7; 32])]);
44222×
            assert_eq!(s.receipt_states().unwrap(), vec![r2]);
44232×
        };
44241×
        let reader = f.open();
44251×
        assert!(!reader.folded.get());
44261×
        check(&reader);
44271×
        assert!(f.backend.inner.read_ref(JOURNAL_KEY).unwrap().is_none());
4428
44291×
        assert!(fold(&w));
44301×
        let reader = f.open();
44311×
        assert!(reader.folded.get());
44321×
        check(&reader);
4433
4434
        // The older lists' entries are per-object entries now, and the
4435
        // marker is there.
44361×
        assert!(f.backend.inner.has(LEGACY_FOLDED_KEY).unwrap());
44371×
        assert!(f
44381×
            .backend
44391×
            .inner
44401×
            .has(&format!("{}{}", fanout_prefix("sig", &r2), s2.key.to_hex()))
44411×
            .unwrap());
44421×
        assert!(f
44431×
            .backend
44441×
            .inner
44451×
            .has(&format!(
44461×
                "{}{}",
44471×
                fanout_prefix("receipt", &r2),
44481×
                ObjectRef([7; 32]).to_hex()
44491×
            ))
44501×
            .unwrap());
4451
4452
        // An older writer after the fold: missed until the next one.
44531×
        let r3 = w.put(ObjectType::Chunk, b"three").unwrap();
44541×
        let s3 = old_writer_signs(&f, &r3, 3);
44551×
        assert!(f.open().signatures(&r3).unwrap().is_empty());
44561×
        assert!(fold(&w));
44571×
        assert_eq!(f.open().signatures(&r3).unwrap(), vec![s3]);
44581×
    }
4459
4460
    /// Writers racing on the journal never lose a line: a lost
4461
    /// compare-and-swap reads the journal again and retries.
4462
    #[test]
44631×
    fn concurrent_journal_appends_lose_nothing() {
44641×
        let f = Fixture::new();
4465
        // One race, made to happen.
44661×
        let w = f.open();
44671×
        f.backend.interfere.store(true, Ordering::SeqCst);
44681×
        w.index_seal(&ObjectRef([1; 32]), &ObjectRef([2; 32]))
44691×
            .unwrap();
44701×
        let j = journal(&w);
44711×
        assert_eq!(j.len(), 2, "the competitor's line and ours: {j:?}");
4472
        // Many, from threads, each its own handle.
44731×
        let backend: Arc<Backend> = f.backend.clone();
44741×
        std::thread::scope(|scope| {
44758×
            for t in 0..8u8 {
44768×
                let backend = backend.clone();
44778×
                let dir = f.dir.path().join(format!("t{t}"));
44788×
                scope.spawn(move || {
44798×
                    let s = Store::open_on(backend, dir).unwrap();
4480200×
                    for i in 0..25u8 {
4481200×
                        s.index_seal(&ObjectRef([t + 10; 32]), &ObjectRef([i; 32]))
4482200×
                            .unwrap();
4483200×
                    }
44848×
                });
4485
            }
44861×
        });
44871×
        assert_eq!(journal(&w).len(), 2 + 8 * 25);
44881×
        assert!(fold(&w));
44891×
        let r = f.open();
44908×
        for t in 0..8u8 {
44918×
            assert_eq!(r.seals_for(&ObjectRef([t + 10; 32])).unwrap().len(), 25);
4492
        }
44931×
    }
4494
4495
    /// A batch adds its entries to the journal in one update when it is
4496
    /// flushed; past the limit the writer folds them into a partial list.
4497
    #[test]
44981×
    fn a_long_journal_is_folded_into_a_partial_list() {
44991×
        let f = Fixture::new();
45001×
        let w = f.open();
45011×
        assert!(fold(&w), "folded, with nothing yet");
45021×
        let key = ObjectRef([5; 32]);
45031×
        f.backend.reset();
45041×
        let batch = w.begin_batch().unwrap();
45051001×
        for i in 0..=JOURNAL_FOLD_AT as u32 {
45061001×
            let mut v = [0u8; 32];
45071001×
            v[..4].copy_from_slice(&i.to_le_bytes());
45081001×
            w.index_seal(&key, &ObjectRef(v)).unwrap();
45091001×
        }
45101×
        assert_eq!(f.backend.n("write_ref"), 0, "nothing until the flush");
45111×
        batch.finish().unwrap();
45121×
        let parts = f
45131×
            .backend
45141×
            .inner
45151×
            .list("pack-")
45161×
            .unwrap()
45171×
            .into_iter()
45184×
            .filter(|e| e.key.ends_with(PART_SUFFIX))
45191×
            .count();
45201×
        assert_eq!(parts, 1);
45211001×
        assert!(journal(&w).values().all(|t| *t != 0), "all folded");
45221×
        let r = f.open();
45231×
        assert_eq!(r.seals_for(&key).unwrap().len(), JOURNAL_FOLD_AT + 1);
4524
        // The next complete fold takes the partial list in.
45251×
        assert!(fold(&w));
45261×
        let lists: Vec<String> = f.open().entry_lists.borrow().clone();
45271×
        assert_eq!(lists.len(), 1);
45281×
        assert!(lists[0].ends_with(LIST_SUFFIX));
45291×
        assert_eq!(f.open().seals_for(&key).unwrap().len(), JOURNAL_FOLD_AT + 1);
45301×
    }
4531
4532
    /// A collection that deletes an object takes its entries out of the
4533
    /// journal too, so the next fold leaves them out.
4534
    #[test]
45351×
    fn pruned_objects_leave_the_journal() {
45361×
        let f = Fixture::new();
45371×
        let w = f.open();
45381×
        let gone = w.put(ObjectType::Chunk, b"gone").unwrap();
45391×
        let kept = w.put(ObjectType::Chunk, b"kept").unwrap();
45401×
        sign(&w, &gone, 1);
45411×
        let s = sign(&w, &kept, 2);
45421×
        assert!(fold(&w));
45431×
        let keep: std::collections::HashSet<ObjectRef> = w
45441×
            .list()
45451×
            .unwrap()
45461×
            .into_iter()
45474×
            .filter(|r| *r != gone)
45481×
            .collect();
45491×
        w.prune(&keep, std::time::Duration::ZERO).unwrap();
45501×
        assert!(journal(&w).keys().all(|k| !k.contains(&gone.to_hex()[2..])));
45511×
        fold(&w);
45521×
        let r = f.open();
45531×
        assert!(r.signatures(&gone).unwrap().is_empty());
45541×
        assert_eq!(r.signatures(&kept).unwrap(), vec![s]);
45551×
        let all = r.with_entries(|e| e.clone()).unwrap().unwrap();
45561×
        assert!(
45571×
            all.iter().all(|k| !k.contains(&gone.to_hex()[2..])),
4558
            "{all:?}"
4559
        );
45601×
    }
4561
4562
    /// Requests per lookup, before and after the fold (#116): twenty
4563
    /// objects, five of them signed, each looked up for its signatures
4564
    /// and its seals, as a fetch does.
4565
    #[test]
45661×
    fn a_folded_store_looks_up_entries_without_listing() {
45671×
        let f = Fixture::new();
45681×
        let w = f.open();
45691×
        let refs: Vec<ObjectRef> = (0..20)
457020×
            .map(|i| {
457120×
                w.put(ObjectType::Chunk, format!("o {i}").as_bytes())
457220×
                    .unwrap()
457320×
            })
45741×
            .collect();
45755×
        for r in &refs[..5] {
45765×
            sign(&w, r, 1);
45775×
        }
45782×
        let lookups = |s: &Store| {
457940×
            for r in &refs {
458040×
                s.signatures(r).unwrap();
458140×
                s.seals_for(r).unwrap();
458240×
            }
45832×
        };
45841×
        let before = f.open();
45851×
        f.backend.reset();
45861×
        lookups(&before);
45871×
        let (before_total, before_list) = (f.backend.total(), f.backend.n("list"));
45881×
        eprintln!("40 lookups before the fold: {}", f.backend.summary());
4589
45901×
        assert!(fold(&w));
45911×
        let after = f.open();
45921×
        f.backend.reset();
45931×
        lookups(&after);
45941×
        eprintln!("40 lookups after the fold: {}", f.backend.summary());
45951×
        assert_eq!(before_list, 40);
45961×
        assert_eq!(f.backend.n("list"), 0);
45971×
        assert_eq!(f.backend.n("get_legacy"), 0, "the older lists are folded");
45981×
        assert_eq!(f.backend.n("read_journal"), 1);
45991×
        assert!(
46001×
            f.backend.total() < before_total / 10,
4601
            "{}",
46020×
            f.backend.summary()
4603
        );
46041×
    }
4605
4606
    /// Once a fold has copied the older lists into per-object entries and
4607
    /// written its marker, readers stop reading them, folded or not; the
4608
    /// answers are the same.
4609
    #[test]
46101×
    fn the_legacy_marker_stops_reads_of_the_older_lists() {
46111×
        let f = Fixture::new();
46121×
        let w = f.open();
46131×
        let r = w.put(ObjectType::Chunk, b"old").unwrap();
46141×
        let sig = SecretKey {
46151×
            ed_seed: [4; 32],
46161×
            ml_seed: [5; 32],
46171×
        }
46181×
        .sign(Domain::Change, &ObjectRef([4; 32]), &r);
46191×
        f.backend
46201×
            .put_new(
46211×
                &legacy_key("sigs", &r),
46221×
                &Signature::encode_list(std::slice::from_ref(&sig)),
46231×
            )
46241×
            .unwrap();
46253×
        for (legacy, v) in [("seals", 9u8), ("certs", 8), ("accepted-lists", 6)] {
46263×
            f.backend
46273×
                .put_new(
46283×
                    &legacy_key(legacy, &r),
46293×
                    format!("{}\n", ObjectRef([v; 32]).to_hex()).as_bytes(),
46303×
                )
46313×
                .unwrap();
46323×
        }
46333×
        let check = |s: &Store| {
46343×
            assert_eq!(s.signatures(&r).unwrap(), vec![sig.clone()]);
46353×
            assert_eq!(s.seals_for(&r).unwrap(), vec![ObjectRef([9; 32])]);
46363×
            assert_eq!(
46373×
                s.index_get("cert", "certs", &r).unwrap(),
46383×
                vec![ObjectRef([8; 32])]
4639
            );
46403×
            assert_eq!(
46413×
                s.index_get_pub("accepted", &r).unwrap(),
46423×
                vec![ObjectRef([6; 32])]
4643
            );
46443×
        };
46451×
        let before = f.open();
46461×
        f.backend.reset();
46471×
        check(&before);
46481×
        assert_eq!(f.backend.n("get_legacy"), 4, "{}", f.backend.summary());
4649
46501×
        assert!(fold(&w));
46511×
        let folded = f.open();
46521×
        f.backend.reset();
46531×
        check(&folded);
46541×
        assert_eq!(f.backend.n("get_legacy") + f.backend.n("list"), 0);
4655
4656
        // Not folded (its lists gone), but marked: per-object listings,
4657
        // still no older lists.
46582×
        for e in f.backend.inner.list("pack-").unwrap() {
46592×
            if is_entry_list(&e.key) {
46601×
                f.backend.inner.delete(&e.key).unwrap();
46611×
            }
4662
        }
46631×
        let marked = f.open();
46641×
        assert!(!marked.folded.get() && marked.legacy_folded.get());
46651×
        f.backend.reset();
46661×
        check(&marked);
46671×
        assert_eq!(f.backend.n("get_legacy"), 0, "{}", f.backend.summary());
46681×
        assert_eq!(f.backend.n("list"), 4);
46691×
    }
4670
4671
    /// A handle that has read the entries finds another process's new
4672
    /// signature once the pack listing's interval has passed, with no
4673
    /// refresh, at the cost of one pack listing and one journal read.
4674
    #[test]
46751×
    fn a_long_lived_handle_sees_new_entries_within_the_interval() {
46761×
        let f = Fixture::new();
46771×
        let w = f.open();
46781×
        let r1 = w.put(ObjectType::Chunk, b"one").unwrap();
46791×
        let s1 = sign(&w, &r1, 1);
46801×
        assert!(fold(&w));
46811×
        let a = f.open();
46821×
        assert_eq!(a.signatures(&r1).unwrap(), vec![s1]);
4683
46841×
        let b = f.open();
46851×
        let r2 = b.put(ObjectType::Chunk, b"two").unwrap();
46861×
        let s2 = sign(&b, &r2, 2);
46871×
        b.index_approval(&r2, &ObjectRef([3; 32])).unwrap();
4688
        // Within the interval: answered from what was read.
46891×
        assert!(a.signatures(&r2).unwrap().is_empty());
4690
4691
        // The interval passes, as the pack listing's does.
46921×
        let t = a.entries_read.get().unwrap();
46931×
        a.entries_read.set(t.checked_sub(ENTRIES_FRESH));
46941×
        f.backend.reset();
46951×
        assert_eq!(a.signatures(&r2).unwrap(), vec![s2]);
46961×
        assert_eq!(a.approvals_for(&r2).unwrap(), vec![ObjectRef([3; 32])]);
46971×
        assert_eq!(f.backend.n("read_journal"), 1, "{}", f.backend.summary());
46981×
        assert_eq!(f.backend.n("list_packs"), 1, "{}", f.backend.summary());
46991×
        assert_eq!(f.backend.n("list"), 0, "{}", f.backend.summary());
47001×
        assert_eq!(
47011×
            f.backend.n("get"),
4702
            1,
4703
            "the signature; the list already read is not read again: {}",
47040×
            f.backend.summary()
4705
        );
47061×
    }
4707
}