OXYGENHUB

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

2007 of 2132 lines run, 317 of 456 functions · run #328 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
28
use crate::attest::Attest;
29
use crate::hash::{hash_object, header, ObjectRef, ObjectType};
30
use crate::identity::{Identity, Signature};
31
use crate::objects::{Blob, Change, ObjectError, StateObject};
32
use crate::receipt::Receipt;
33
use crate::strata::{self, Error as BackendError, LocalStore, MemoryStore};
34
use rand::RngCore;
35
use std::cell::RefCell;
36
use std::fs;
37
use std::io::{self, Write};
38
use std::path::{Path, PathBuf};
39
use std::sync::Arc;
40
41
const PACK_MAGIC: &[u8; 8] = b"OXYPACK1";
42
const IDX_MAGIC: &[u8; 8] = b"OXYIDX01";
43
const IDX_ENTRY: usize = 32 + 8 + 4 + 4 + 1 + 1;
44
const FLAG_ZSTD: u8 = 1;
45
/// Compressed with the pack's trained dictionary (`pack-ID.dict`), which a
46
/// full repack makes from a sample of the objects it packs: small files
47
/// share most of their vocabulary, which each one alone cannot use.
48
const FLAG_DICT: u8 = 2;
49
/// The size of a pack's dictionary, and the least a repack must pack for
50
/// one to pay.
51
const DICT_BYTES: usize = 110 << 10;
52
const DICT_MIN_OBJECTS: usize = 1000;
53
/// Objects up to this size are compressed with the dictionary.
54
const DICT_OBJECT_MAX: usize = 64 << 10;
55
const ZSTD_LEVEL: i32 = 3;
56
/// The level of a full repack (`oxy gc`): slower, once, for a smaller store.
57
pub const REPACK_ZSTD_LEVEL: i32 = 9;
58
/// Largest object a pack entry may claim; anything bigger is corruption,
59
/// and refusing it keeps a bad index from driving huge allocations.
60
const MAX_OBJECT_LEN: u32 = 1 << 30;
61
62
/// An object read from the store: its type and body.
63
pub type Object = (ObjectType, Vec<u8>);
64
65
/// Where a long operation reports what it is doing (oxygen#66): a phase,
66
/// or a count within one, as a line for a spinner or a log.
67
pub type Progress<'a> = &'a mut dyn FnMut(&str);
68
69
/// The storage under a store: objects by key, see [`crate::strata`].
70
pub type Backend = dyn strata::Store;
71
7255×
fn silent() -> impl FnMut(&str) {
73116×
    |_| {}
7455×
}
75
76
/// A backend error as a store error: what is not found or already there
77
/// is handled where it matters; the rest is an I/O failure.
784×
fn backend_err(e: BackendError) -> StoreError {
794×
    match e {
804×
        BackendError::Io(e) => StoreError::Io(e),
810×
        other => StoreError::Io(io::Error::other(other.to_string())),
82
    }
834×
}
84
85
/// `put_new` where an object already there is what we wanted.
868069×
fn put_new_or_existing(backend: &Backend, key: &str, bytes: &[u8]) -> Result<()> {
878069×
    match backend.put_new(key, bytes) {
888069×
        Ok(()) | Err(BackendError::Exists(_)) => Ok(()),
890×
        Err(e) => Err(backend_err(e)),
90
    }
918069×
}
92
93
/// `get` as an option: `None` when there is no such key.
9456453×
fn get_optional(backend: &Backend, key: &str) -> Result<Option<Vec<u8>>> {
9556453×
    match backend.get(key) {
9648961×
        Ok(b) => Ok(Some(b)),
977492×
        Err(BackendError::NotFound(_)) => Ok(None),
980×
        Err(e) => Err(backend_err(e)),
99
    }
10056453×
}
101
1023176×
fn delete(backend: &Backend, key: &str) -> Result<()> {
1033176×
    backend.delete(key).map_err(backend_err)
1043176×
}
105
106
/// `loose/<hh>/<rest>`.
10790214×
fn loose_key(r: &ObjectRef) -> String {
10890214×
    let h = r.to_hex();
10990214×
    format!("loose/{}/{}", &h[..2], &h[2..])
11090214×
}
111
112
/// The object a loose key names, if it is one.
11317155×
fn loose_ref(key: &str) -> Option<ObjectRef> {
11417155×
    let rest = key.strip_prefix("loose/")?;
11517155×
    let (hh, tail) = rest.split_once('/')?;
11617155×
    if hh.len() != 2 || tail.contains('/') {
1170×
        return None;
11817155×
    }
11917155×
    ObjectRef::from_hex(&format!("{hh}{tail}"))
12017155×
}
121
122
/// `<index>/<hh>/<rest>/` for the entries kept under an object.
1235910×
fn fanout_prefix(index: &str, r: &ObjectRef) -> String {
1245910×
    let h = r.to_hex();
1255910×
    format!("{index}/{}/{}/", &h[..2], &h[2..])
1265910×
}
127
128
/// `<index>/<hh>/<rest>`: the one-file-per-object layout of older stores.
1295131×
fn legacy_key(index: &str, r: &ObjectRef) -> String {
1305131×
    let h = r.to_hex();
1315131×
    format!("{index}/{}/{}", &h[..2], &h[2..])
1325131×
}
133
13423×
fn now_ms() -> u64 {
13523×
    std::time::SystemTime::now()
13623×
        .duration_since(std::time::UNIX_EPOCH)
13723×
        .map(|d| d.as_millis() as u64)
13823×
        .unwrap_or(0)
13923×
}
140
141
#[derive(Clone, Copy, Debug)]
142
struct PackEntry {
143
    offset: u64,
144
    stored_len: u32,
145
    raw_len: u32,
146
    ty: ObjectType,
147
    flags: u8,
148
}
149
150
/// One pack's index, fully in memory and sorted by ObjectRef.
151
struct PackIndex {
152
    /// The pack's key, `pack-<id>.pack`.
153
    key: String,
154
    refs: Vec<ObjectRef>,
155
    entries: Vec<PackEntry>,
156
    /// The pack's dictionary, prepared once, when it has one.
157
    dict: Option<Arc<zstd::dict::DecoderDictionary<'static>>>,
158
}
159
160
impl std::fmt::Debug for PackIndex {
1610×
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
1620×
        f.debug_struct("PackIndex")
1630×
            .field("key", &self.key)
1640×
            .field("objects", &self.refs.len())
1650×
            .field("dict", &self.dict.is_some())
1660×
            .finish()
1670×
    }
168
}
169
170
/// Decode a stored pack entry and check it against its reference.
17115860×
fn decode_entry(
17215860×
    r: &ObjectRef,
17315860×
    e: PackEntry,
17415860×
    stored: Vec<u8>,
17515860×
    dict: Option<&Arc<zstd::dict::DecoderDictionary<'static>>>,
17615860×
) -> Result<Object> {
17715860×
    let body = if e.flags & FLAG_DICT != 0 {
1784501×
        let dict = dict.ok_or_else(|| {
1790×
            StoreError::Corrupt(*r, "packed with a dictionary the pack lacks".into())
1800×
        })?;
1814501×
        zstd::bulk::Decompressor::with_prepared_dictionary(dict)
1824501×
            .and_then(|mut d| d.decompress(&stored, e.raw_len as usize))
1834501×
            .map_err(|err| StoreError::Corrupt(*r, format!("zstd: {err}")))?
18411359×
    } else if e.flags & FLAG_ZSTD != 0 {
1856491×
        zstd::bulk::decompress(&stored, e.raw_len as usize)
1866491×
            .map_err(|err| StoreError::Corrupt(*r, format!("zstd: {err}")))?
187
    } else {
1884868×
        stored
189
    };
19015855×
    if body.len() != e.raw_len as usize || hash_object(e.ty, &body) != *r {
191119×
        return Err(StoreError::Corrupt(
192119×
            *r,
193119×
            "packed object does not match its reference".into(),
194119×
        ));
19515736×
    }
19615736×
    Ok((e.ty, body))
19715860×
}
198
199
impl PackIndex {
200
    /// Load a pack's index; `pack_len` is the pack's size, from the listing
201
    /// that found it (a stat of it would be a request of its own).
2021142×
    fn load(backend: &Backend, idx_key: &str, pack_key: &str, pack_len: u64) -> Result<PackIndex> {
2031142×
        let data = backend.get(idx_key).map_err(backend_err)?;
2041142×
        let bad = |m: &str| StoreError::Corrupt(ObjectRef::ZERO, format!("{idx_key}: {m}"));
2051142×
        if data.len() < 12 || &data[..8] != IDX_MAGIC {
2062×
            return Err(bad("bad index header"));
2071140×
        }
2081140×
        let count = u32::from_le_bytes(data[8..12].try_into().unwrap()) as usize;
2091140×
        if data.len() != 12 + count * IDX_ENTRY {
21042×
            return Err(bad("index length mismatch"));
2111098×
        }
2121098×
        let mut refs = Vec::with_capacity(count);
2131098×
        let mut entries = Vec::with_capacity(count);
21426375×
        for i in 0..count {
21526375×
            let e = &data[12 + i * IDX_ENTRY..12 + (i + 1) * IDX_ENTRY];
21626375×
            let r = ObjectRef::from_slice(&e[..32]).unwrap();
21726375×
            if let Some(prev) = refs.last() {
21825277×
                if r <= *prev {
2193×
                    return Err(bad("index not sorted"));
22025274×
                }
2211098×
            }
22226372×
            refs.push(r);
22326367×
            let entry = PackEntry {
22426372×
                offset: u64::from_le_bytes(e[32..40].try_into().unwrap()),
22526372×
                stored_len: u32::from_le_bytes(e[40..44].try_into().unwrap()),
22626372×
                raw_len: u32::from_le_bytes(e[44..48].try_into().unwrap()),
22726372×
                ty: ObjectType::ALL
22826372×
                    .get(e[48] as usize)
22926372×
                    .copied()
23026372×
                    .ok_or_else(|| bad("unknown object type"))?,
23126367×
                flags: e[49],
232
            };
23326367×
            if entry.stored_len > MAX_OBJECT_LEN || entry.raw_len > MAX_OBJECT_LEN {
2343×
                return Err(StoreError::Corrupt(
2353×
                    r,
2363×
                    format!(
2373×
                        "{idx_key}: the index claims an object of {} bytes ({} stored), more than the {MAX_OBJECT_LEN} an object may be",
2383×
                        entry.raw_len, entry.stored_len
2393×
                    ),
2403×
                ));
24126364×
            }
24226364×
            if entry.offset.saturating_add(entry.stored_len as u64) > pack_len {
24354×
                return Err(bad("index entry points outside the pack"));
24426310×
            }
24526310×
            entries.push(entry);
246
        }
2471033×
        let dict_key = format!("{}.dict", pack_key.trim_end_matches(".pack"));
2481033×
        let dict = get_optional(backend, &dict_key)?
2491033×
            .map(|d| Arc::new(zstd::dict::DecoderDictionary::copy(&d)));
2501033×
        Ok(PackIndex {
2511033×
            key: pack_key.to_string(),
2521033×
            refs,
2531033×
            entries,
2541033×
            dict,
2551033×
        })
2561142×
    }
257
25818543×
    fn find(&self, r: &ObjectRef) -> Option<PackEntry> {
25918543×
        self.refs.binary_search(r).ok().map(|i| self.entries[i])
26018543×
    }
261
262
    /// Read one entry from the pack: one ranged read, then decode.
26315378×
    fn fetch(&self, backend: &Backend, r: &ObjectRef, e: PackEntry) -> Result<Object> {
26415378×
        let stored = backend
26515378×
            .get_range(&self.key, e.offset..e.offset + e.stored_len as u64)
26615378×
            .map_err(|err| match err {
2670×
                BackendError::NotFound(_) => StoreError::NotFound(*r),
2680×
                other => backend_err(other),
2690×
            })?;
27015378×
        decode_entry(r, e, stored, self.dict.as_ref())
27115378×
    }
272
}
273
27412038×
fn type_index(ty: ObjectType) -> u8 {
27517716×
    ObjectType::ALL.iter().position(|t| *t == ty).unwrap() as u8
27612038×
}
277
278
/// What `pack_loose` did.
279
#[derive(Debug, Clone, Default, PartialEq, Eq)]
280
pub struct PackReport {
281
    pub objects: usize,
282
    pub raw_bytes: u64,
283
    pub packed_bytes: u64,
284
}
285
286
/// Packs tolerated before `merge_packs` folds them into one (spec §13.4).
287
pub const MAX_PACKS: usize = 16;
288
289
/// What `prune` did.
290
#[derive(Debug, Clone, Default, PartialEq, Eq)]
291
pub struct PruneReport {
292
    pub objects: usize,
293
    pub bytes: u64,
294
    /// Unreachable objects left alone because they are younger than the grace period.
295
    pub deferred: usize,
296
}
297
298
#[derive(Debug, thiserror::Error)]
299
pub enum StoreError {
300
    #[error("io: {0}")]
301
    Io(#[from] io::Error),
302
    #[error("corrupt object {0}: {1}")]
303
    Corrupt(ObjectRef, String),
304
    #[error("object {0}")]
305
    Object(#[from] ObjectError),
306
    #[error("object {0} is a {1}, expected {2}")]
307
    WrongType(ObjectRef, &'static str, &'static str),
308
    #[error("object {0} not found")]
309
    NotFound(ObjectRef),
310
    #[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)")]
311
    TooLarge {
312
        kind: &'static str,
313
        size: usize,
314
        limit: u32,
315
    },
316
}
317
318
pub type Result<T> = std::result::Result<T, StoreError>;
319
320
pub struct Store {
321
    backend: Arc<Backend>,
322
    /// Temporary files (a pack being streamed) and derived caches (`metas`).
323
    /// For a store in a directory it is that directory.
324
    scratch: PathBuf,
325
    /// The backend is that directory, so the old file paths still hold.
326
    local: bool,
327
    packs: RefCell<Vec<PackIndex>>,
328
    /// An open write batch (spec §13.4): objects go here instead of to loose
329
    /// files, and are written out with one `fsync` when the batch finishes.
330
    pending: RefCell<Option<Pending>>,
331
    batch_depth: std::cell::Cell<u32>,
332
    /// Change metadata already read, from the `metas` cache file.
333
    metas: RefCell<Option<crate::fasthash::FastMap<ObjectRef, crate::state::ChangeMeta>>>,
334
}
335
336
impl std::fmt::Debug for Store {
3370×
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
3380×
        f.debug_struct("Store")
3390×
            .field("backend", &self.backend.describe())
3400×
            .field("scratch", &self.scratch)
3410×
            .field("packs", &self.packs.borrow().len())
3420×
            .finish()
3430×
    }
344
}
345
346
/// Objects written since `begin_batch`. Small batches stay in memory and
347
/// become loose objects; once `SPILL_OBJECTS` or `SPILL_BYTES` is exceeded
348
/// the batch streams into a pack file instead.
349
#[derive(Debug)]
350
struct Pending {
351
    buffered: Vec<(ObjectType, Vec<u8>)>,
352
    buffered_bytes: usize,
353
    lookup: std::collections::HashMap<ObjectRef, PendingEntry>,
354
    pack: Option<PendingPack>,
355
    report: PackReport,
356
}
357
358
#[derive(Debug, Clone, Copy)]
359
enum PendingEntry {
360
    Buffered(usize),
361
    Packed(PackEntry),
362
}
363
364
#[derive(Debug)]
365
struct PendingPack {
366
    id: String,
367
    tmp_path: PathBuf,
368
    writer: io::BufWriter<fs::File>,
369
    offset: u64,
370
    index: Vec<(ObjectRef, PackEntry)>,
371
}
372
373
const SPILL_OBJECTS: usize = 256;
374
const SPILL_BYTES: usize = 8 << 20;
375
376
/// Finishes the enclosing batch when dropped; call `finish` to see errors.
377
pub struct Batch<'a> {
378
    store: &'a Store,
379
    done: bool,
380
}
381
382
impl Batch<'_> {
3832742×
    pub fn finish(mut self) -> Result<()> {
3842742×
        self.done = true;
3852742×
        self.store.finish_batch()
3862742×
    }
387
}
388
389
impl Drop for Batch<'_> {
3902744×
    fn drop(&mut self) {
3912744×
        if !self.done {
3922×
            let _ = self.store.finish_batch();
3932742×
        }
3942744×
    }
395
}
396
397
/// Append one stored object to a pending pack and index it.
3982100×
fn write_pack_entry(
3992100×
    pack: &mut PendingPack,
4002100×
    r: &ObjectRef,
4012100×
    ty: ObjectType,
4022100×
    raw_len: usize,
4032100×
    flags: u8,
4042100×
    stored: &[u8],
4052100×
    report: &mut PackReport,
4062100×
) -> Result<PackEntry> {
4072100×
    let header_len = 1 + 1 + 4 + 4 + 32;
4082100×
    pack.writer.write_all(&[type_index(ty), flags])?;
4092100×
    pack.writer
4102100×
        .write_all(&(stored.len() as u32).to_le_bytes())?;
4112100×
    pack.writer.write_all(&(raw_len as u32).to_le_bytes())?;
4122100×
    pack.writer.write_all(&r.0)?;
4132100×
    pack.writer.write_all(stored)?;
4142100×
    let entry = PackEntry {
4152100×
        offset: pack.offset + header_len,
4162100×
        stored_len: stored.len() as u32,
4172100×
        raw_len: raw_len as u32,
4182100×
        ty,
4192100×
        flags,
4202100×
    };
4212100×
    pack.offset += header_len + stored.len() as u64;
4222100×
    pack.index.push((*r, entry));
4232100×
    report.objects += 1;
4242100×
    report.raw_bytes += raw_len as u64;
4252100×
    report.packed_bytes += stored.len() as u64;
4262100×
    Ok(entry)
4272100×
}
428
429
/// A random name for a temporary file or a pack.
43078×
fn fresh_id() -> String {
43178×
    let mut nonce = [0u8; 8];
43278×
    rand::rngs::OsRng.fill_bytes(&mut nonce);
43378×
    hex::encode(nonce)
43478×
}
435
436
/// Under `OXY_STORE_BACKEND=memory` every store opens on a store in
437
/// memory shared by directory, so the whole test suite can run without a
438
/// directory store, as it would over a bucket. Objects live for the
439
/// process.
4402813×
fn memory_backend_for(dir: &Path) -> Option<Arc<Backend>> {
441
    use std::collections::HashMap;
442
    use std::sync::{Mutex, OnceLock};
4432813×
    if std::env::var_os("OXY_STORE_BACKEND").as_deref() != Some(std::ffi::OsStr::new("memory")) {
4442813×
        return None;
4450×
    }
446
    static STORES: OnceLock<Mutex<HashMap<PathBuf, Arc<MemoryStore>>>> = OnceLock::new();
4470×
    let key = dir.canonicalize().unwrap_or_else(|_| dir.to_path_buf());
4480×
    let mut map = STORES
4490×
        .get_or_init(Default::default)
4500×
        .lock()
4510×
        .unwrap_or_else(|p| p.into_inner());
4520×
    let store = map
4530×
        .entry(key)
4540×
        .or_insert_with(|| Arc::new(MemoryStore::new()))
4550×
        .clone();
4560×
    Some(store)
4572813×
}
458
459
#[cfg(test)]
460
thread_local! {
461
    /// A backend every store opened on this thread uses instead of its
462
    /// directory: tests of a repository kept on a store that is not this
463
    /// disk's directory, with several handles on one store.
464
    pub(crate) static TEST_BACKEND: RefCell<Option<Arc<Backend>>> = const { RefCell::new(None) };
465
}
466
467
impl Store {
468
    /// Open the store in `dir`, which is also its scratch directory.
4692817×
    pub fn open(dir: impl Into<PathBuf>) -> io::Result<Store> {
4702817×
        let dir = dir.into();
4712817×
        fs::create_dir_all(dir.join("loose"))?;
472
        #[cfg(test)]
4731227×
        if let Some(backend) = TEST_BACKEND.with(|t| t.borrow().clone()) {
4744×
            return Store::open_on(backend, dir);
4751223×
        }
4762813×
        if let Some(backend) = memory_backend_for(&dir) {
4770×
            return Store::open_on(backend, dir);
4782813×
        }
4792813×
        let backend: Arc<Backend> =
4802813×
            Arc::new(LocalStore::open(&dir).map_err(|e| io::Error::other(e.to_string()))?);
4812813×
        let mut store = Store::open_on(backend, dir)?;
4822704×
        store.local = true;
4832704×
        Ok(store)
4842817×
    }
485
486
    /// Open the store kept on `backend`, with `scratch` for what must be
487
    /// on this disk: a pack while it is being written, and caches.
4882833×
    pub fn open_on(backend: Arc<Backend>, scratch: impl Into<PathBuf>) -> io::Result<Store> {
4892833×
        let scratch = scratch.into();
4902833×
        fs::create_dir_all(&scratch)?;
4912833×
        let store = Store {
4922833×
            backend,
4932833×
            scratch,
4942833×
            local: false,
4952833×
            packs: RefCell::new(Vec::new()),
4962833×
            pending: RefCell::new(None),
4972833×
            batch_depth: std::cell::Cell::new(0),
4982833×
            metas: RefCell::new(None),
4992833×
        };
5002833×
        store
5012833×
            .reload_packs()
5022833×
            .map_err(|e| io::Error::other(e.to_string()))?;
5032724×
        Ok(store)
5042833×
    }
505
506
    /// The storage under this store.
50735×
    pub fn backend(&self) -> &Arc<Backend> {
50835×
        &self.backend
50935×
    }
510
511
    /// Whether the store is the directory `dir()` names, where every key
512
    /// is a file.
51324043×
    pub fn is_local(&self) -> bool {
51424043×
        self.local
51524043×
    }
516
5172984×
    fn reload_packs(&self) -> Result<()> {
5182984×
        let listed = self.backend.list("pack-").map_err(backend_err)?;
5192984×
        let mut idx_keys: Vec<&str> = listed
5202984×
            .iter()
5212984×
            .map(|e| e.key.as_str())
5222984×
            .filter(|k| k.ends_with(".idx"))
5232984×
            .collect();
5242984×
        idx_keys.sort();
5252984×
        let mut packs = Vec::new();
5262984×
        for idx in idx_keys {
5271142×
            let pack = format!("{}.pack", idx.trim_end_matches(".idx"));
52810761×
            if let Some(p) = listed.iter().find(|e| e.key == pack) {
5291142×
                packs.push(PackIndex::load(self.backend.as_ref(), idx, &pack, p.size)?);
5300×
            }
531
        }
5322875×
        *self.packs.borrow_mut() = packs;
5332875×
        Ok(())
5342984×
    }
535
53615×
    pub fn pack_count(&self) -> usize {
53715×
        self.packs.borrow().len()
53815×
    }
539
54082203×
    fn find_packed(&self, r: &ObjectRef) -> Option<(usize, PackEntry)> {
54182203×
        self.packs
54282203×
            .borrow()
54382203×
            .iter()
54482203×
            .enumerate()
54582203×
            .find_map(|(i, p)| p.find(r).map(|e| (i, e)))
54682203×
    }
547
548
    /// Loose objects, with their keys and ages.
549842×
    fn loose_objects(&self) -> Result<Vec<(ObjectRef, strata::Entry)>> {
550842×
        let mut out: Vec<(ObjectRef, strata::Entry)> = self
551842×
            .backend
552842×
            .list("loose/")
553842×
            .map_err(backend_err)?
554842×
            .into_iter()
55517155×
            .filter_map(|e| loose_ref(&e.key).map(|r| (r, e)))
556842×
            .collect();
557842×
        out.sort_by_key(|(r, _)| *r);
558842×
        Ok(out)
559842×
    }
560
561667×
    pub fn loose_count(&self) -> io::Result<usize> {
562667×
        self.loose_objects()
563667×
            .map(|l| l.len())
564667×
            .map_err(|e| io::Error::other(e.to_string()))
565667×
    }
566
567
    /// Move every loose object into a new pack (spec §13.4). Loose objects
568
    /// are deleted only after the pack and its index are durable.
56945×
    pub fn pack_loose(&self) -> Result<PackReport> {
57045×
        let loose = self.loose_objects()?;
57145×
        self.write_pack(
57245×
            &loose.iter().map(|(r, _)| *r).collect::<Vec<_>>(),
573434×
            &loose.iter().map(|(_, e)| e.key.clone()).collect::<Vec<_>>(),
57445×
            &[],
575
            ZSTD_LEVEL,
57645×
            &mut silent(),
577
        )
57845×
    }
579
580
    /// Fold every pack and loose object into one pack compressed at
581
    /// `level` (spec §13.4): what an explicit `oxy gc` does, as `git gc`
582
    /// repacks. Returns the report and how many packs were folded in.
5833×
    pub fn repack_all(&self, level: i32) -> Result<(PackReport, usize)> {
5843×
        self.repack_all_with(level, &mut silent())
5853×
    }
586
587
    /// [`Store::repack_all`], reporting progress.
58821×
    pub fn repack_all_with(&self, level: i32, progress: Progress) -> Result<(PackReport, usize)> {
58921×
        let loose = self.loose_objects()?;
59021×
        let (mut refs, keys, trained): (Vec<ObjectRef>, Vec<String>, bool) = {
59121×
            let packs = self.packs.borrow();
592
            (
59321×
                packs.iter().flat_map(|p| p.refs.iter().copied()).collect(),
59421×
                packs.iter().map(|p| p.key.clone()).collect(),
59521×
                packs.first().is_some_and(|p| p.dict.is_some()),
596
            )
597
        };
598
        // Nothing loose and one pack already made by a full repack (only
599
        // such a pack has a trained dictionary): rewriting it would give
600
        // the same pack. A hub's nightly gc rewrote andymccall/oxygen's
601
        // 48 000 objects (37 s under its write lock) every time.
60221×
        if loose.is_empty() && keys.len() <= 1 && (level == ZSTD_LEVEL || trained) {
6031×
            return Ok((PackReport::default(), 0));
60420×
        }
60520×
        refs.extend(loose.iter().map(|(r, _)| *r));
60620×
        let loose_keys: Vec<String> = loose.into_iter().map(|(_, e)| e.key).collect();
60720×
        let n = keys.len();
60820×
        let report = self.write_pack(&refs, &loose_keys, &keys, level, progress)?;
60920×
        Ok((report, n))
61021×
    }
611
612
    /// Fold every pack into one when there are more than `MAX_PACKS`
613
    /// (spec §13.4). Returns how many packs were merged.
6144×
    pub fn merge_packs(&self) -> Result<usize> {
6152×
        let (refs, keys): (Vec<ObjectRef>, Vec<String>) = {
6164×
            let packs = self.packs.borrow();
6174×
            if packs.len() <= MAX_PACKS {
6182×
                return Ok(0);
6192×
            }
620
            (
62136×
                packs.iter().flat_map(|p| p.refs.iter().copied()).collect(),
62236×
                packs.iter().map(|p| p.key.clone()).collect(),
623
            )
624
        };
6252×
        let n = keys.len();
6262×
        self.write_pack(&refs, &[], &keys, ZSTD_LEVEL, &mut silent())?;
6272×
        Ok(n)
6284×
    }
629
630
    /// Delete every object not in `keep` (spec §21). Loose objects and packs
631
    /// younger than `grace` are left alone, since a concurrent writer may not
632
    /// have recorded its references yet. Packs holding unreachable objects
633
    /// are rewritten without them. Signatures, seal and receipt entries of
634
    /// deleted objects go too.
6355×
    pub fn prune(
6365×
        &self,
6375×
        keep: &std::collections::HashSet<ObjectRef>,
6385×
        grace: std::time::Duration,
6395×
    ) -> Result<PruneReport> {
6405×
        self.prune_with(keep, grace, &mut silent())
6415×
    }
642
643
    /// [`Store::prune`], reporting progress.
64423×
    pub fn prune_with(
64523×
        &self,
64623×
        keep: &std::collections::HashSet<ObjectRef>,
64723×
        grace: std::time::Duration,
64823×
        progress: Progress,
64923×
    ) -> Result<PruneReport> {
65023×
        progress("pruning loose objects");
65123×
        let now = now_ms();
65223×
        let grace_ms = grace.as_millis() as u64;
653
        // An object whose age the store does not know is old enough.
65423×
        let old_enough =
65537×
            |modified: Option<u64>| modified.is_none_or(|t| now.saturating_sub(t) >= grace_ms);
65623×
        let mut report = PruneReport::default();
657
        // Loose objects.
658320×
        for (r, entry) in self.loose_objects()? {
659320×
            if keep.contains(&r) {
660292×
                continue;
66128×
            }
66228×
            if !old_enough(entry.modified_ms) {
6634×
                report.deferred += 1;
6644×
                continue;
66524×
            }
66624×
            report.bytes += entry.size;
66724×
            report.objects += 1;
66824×
            delete(self.backend.as_ref(), &entry.key)?;
66924×
            self.remove_sidecars(&r)?;
670
        }
671
        // Packs: rewrite those that hold unreachable objects.
67223×
        let mut keep_refs: Vec<ObjectRef> = Vec::new();
67323×
        let mut old_keys: Vec<String> = Vec::new();
67423×
        let mut rewrite = false;
67523×
        progress("pruning packs");
67623×
        for p in self.packs.borrow().iter() {
677150×
            let unreachable = p.refs.iter().filter(|r| !keep.contains(r)).count();
67810×
            if unreachable == 0 {
6791×
                continue;
6809×
            }
6819×
            let modified = self
6829×
                .backend
6839×
                .stat(&p.key)
6849×
                .map_err(backend_err)?
6859×
                .and_then(|e| e.modified_ms);
6869×
            if !old_enough(modified) {
6873×
                report.deferred += unreachable;
6883×
                continue;
6896×
            }
6906×
            rewrite = true;
6916×
            old_keys.push(p.key.clone());
692102×
            for (r, e) in p.refs.iter().zip(p.entries.iter()) {
693102×
                if keep.contains(r) {
69461×
                    keep_refs.push(*r);
69561×
                } else {
69641×
                    report.objects += 1;
69741×
                    report.bytes += e.stored_len as u64;
69841×
                }
699
            }
700
        }
70123×
        if rewrite {
7026×
            let dropped: Vec<ObjectRef> = {
7036×
                let packs = self.packs.borrow();
7046×
                packs
7056×
                    .iter()
7066×
                    .filter(|p| old_keys.contains(&p.key))
7076×
                    .flat_map(|p| p.refs.iter().copied())
708102×
                    .filter(|r| !keep.contains(r))
7096×
                    .collect()
710
            };
7116×
            self.write_pack(&keep_refs, &[], &old_keys, ZSTD_LEVEL, progress)?;
71241×
            for r in dropped {
71341×
                self.remove_sidecars(&r)?;
714
            }
71517×
        }
71623×
        Ok(report)
71723×
    }
718
719
    /// Everything kept beside an object: its signatures and the seal and
720
    /// receipt entries under it, in both layouts.
72165×
    fn remove_sidecars(&self, r: &ObjectRef) -> Result<()> {
722195×
        for index in ["sigs", "seals", "receipts"] {
723195×
            delete(self.backend.as_ref(), &legacy_key(index, r))?;
724
        }
725195×
        for index in ["sig", "seal", "receipt"] {
726195×
            for e in self
727195×
                .backend
728195×
                .list(&fanout_prefix(index, r))
729195×
                .map_err(backend_err)?
730
            {
73112×
                delete(self.backend.as_ref(), &e.key)?;
732
            }
733
        }
73465×
        Ok(())
73565×
    }
736
737
    /// Put a complete, synced temporary pack file and its index into the
738
    /// store; `index` must be sorted by ObjectRef. The pack goes first: an
739
    /// index never names a pack that is not there.
74078×
    fn finish_pack_files(
74178×
        &self,
74278×
        id: &str,
74378×
        tmp_pack: &Path,
74478×
        index: &[(ObjectRef, PackEntry)],
74578×
    ) -> Result<()> {
74678×
        let mut idx = Vec::with_capacity(12 + index.len() * IDX_ENTRY);
74778×
        idx.extend_from_slice(IDX_MAGIC);
74878×
        idx.extend_from_slice(&(index.len() as u32).to_le_bytes());
7496019×
        for (r, e) in index {
7506019×
            idx.extend_from_slice(&r.0);
7516019×
            idx.extend_from_slice(&e.offset.to_le_bytes());
7526019×
            idx.extend_from_slice(&e.stored_len.to_le_bytes());
7536019×
            idx.extend_from_slice(&e.raw_len.to_le_bytes());
7546019×
            idx.extend_from_slice(&[type_index(e.ty), e.flags]);
7556019×
        }
75678×
        let put = self
75778×
            .backend
75878×
            .put_new_file(&format!("pack-{id}.pack"), tmp_pack)
75978×
            .map_err(backend_err);
760
        // A store on another disk copied it; the one in this directory
761
        // moved it and this finds nothing.
76278×
        let _ = fs::remove_file(tmp_pack);
76378×
        put?;
76478×
        self.backend
76578×
            .put_new(&format!("pack-{id}.idx"), &idx)
76678×
            .map_err(backend_err)?;
76778×
        self.reload_packs()
76878×
    }
769
770
    /// Remove a pack with its index and dictionary.
77149×
    fn remove_pack(&self, pack_key: &str) -> Result<()> {
77249×
        let stem = pack_key.trim_end_matches(".pack");
77349×
        delete(self.backend.as_ref(), pack_key)?;
77449×
        delete(self.backend.as_ref(), &format!("{stem}.idx"))?;
77549×
        delete(self.backend.as_ref(), &format!("{stem}.dict"))
77649×
    }
777
778
    /// Write `refs` into a new pack, then delete `loose_keys` and `old_packs`.
77973×
    fn write_pack(
78073×
        &self,
78173×
        refs: &[ObjectRef],
78273×
        loose_keys: &[String],
78373×
        old_packs: &[String],
78473×
        level: i32,
78573×
        progress: Progress,
78673×
    ) -> Result<PackReport> {
78773×
        if refs.is_empty() {
7880×
            for p in old_packs {
7890×
                self.remove_pack(p)?;
790
            }
7910×
            self.reload_packs()?;
7920×
            return Ok(PackReport::default());
79373×
        }
79473×
        let mut refs: Vec<ObjectRef> = refs.to_vec();
79573×
        refs.sort();
79673×
        refs.dedup();
79773×
        let id = fresh_id();
79873×
        let tmp_pack = self.scratch.join(format!(".tmp-{id}.pack"));
79973×
        let mut report = PackReport::default();
80073×
        let mut index: Vec<(ObjectRef, PackEntry)> = Vec::with_capacity(refs.len());
801
        // A full repack of enough objects trains a dictionary for them.
80273×
        let dict = if level == REPACK_ZSTD_LEVEL && refs.len() >= DICT_MIN_OBJECTS {
8032×
            let _t = crate::timing::phase("pack: train a dictionary");
8042×
            progress(&format!("training a dictionary on {} objects", refs.len()));
8052×
            self.train_dictionary(&refs)
806
        } else {
80771×
            None
808
        };
80973×
        let encoder = dict
81073×
            .as_ref()
81173×
            .map(|d| zstd::dict::EncoderDictionary::copy(d, level));
81273×
        let nocompress = self.nocompress_set()?;
813
        {
81473×
            let mut f = io::BufWriter::new(fs::File::create(&tmp_pack)?);
81573×
            f.write_all(PACK_MAGIC)?;
81673×
            let mut offset = PACK_MAGIC.len() as u64;
817
            // Read a batch (the store is not shared between threads),
818
            // compress it on every core, write it in order.
81973×
            let mut batches = refs.chunks(1024);
82073×
            let mut compressed_batch = Vec::new();
82173×
            let total = refs.len();
82273×
            let mut reported = usize::MAX;
823
            loop {
8243992×
                if compressed_batch.is_empty() {
825148×
                    let Some(batch) = batches.next() else { break };
826
                    // Every tenth, and the first batch.
82775×
                    let tenth = report.objects * 10 / total.max(1);
82875×
                    if tenth != reported {
82975×
                        reported = tenth;
83075×
                        progress(&format!("packing {} of {total} objects", report.objects));
83175×
                    }
83275×
                    let mut read: Vec<(ObjectRef, ObjectType, Vec<u8>, bool)> =
83375×
                        Vec::with_capacity(batch.len());
8343919×
                    for r in batch {
8353919×
                        let (ty, body) = self.get(r)?;
8363919×
                        read.push((*r, ty, body, nocompress.contains(r)));
837
                    }
838
                    use rayon::prelude::*;
83975×
                    compressed_batch = read
84075×
                        .into_par_iter()
84175×
                        .map_init(
8421030×
                            || {
8431030×
                                encoder.as_ref().and_then(|e| {
844395×
                                    zstd::bulk::Compressor::with_prepared_dictionary(e).ok()
845395×
                                })
8461030×
                            },
8473919×
                            |with_dict, (r, ty, body, raw)| {
8483919×
                                let packed = if raw {
8492×
                                    None
850
                                } else {
851
                                    // Small objects with the dictionary;
852
                                    // a large one has context enough.
8533917×
                                    let dicted = with_dict
8543917×
                                        .as_mut()
8553917×
                                        .filter(|_| body.len() <= DICT_OBJECT_MAX)
8563917×
                                        .and_then(|c| c.compress(&body).ok())
8573917×
                                        .map(|c| (FLAG_ZSTD | FLAG_DICT, c));
8583917×
                                    dicted
8593917×
                                        .or_else(|| {
860916×
                                            zstd::bulk::compress(&body, level)
861916×
                                                .ok()
862916×
                                                .map(|c| (FLAG_ZSTD, c))
863916×
                                        })
8643917×
                                        .filter(|(_, c)| c.len() < body.len())
865
                                };
8663919×
                                (r, ty, body, packed)
8673919×
                            },
868
                        )
86975×
                        .collect::<Vec<_>>();
87075×
                    compressed_batch.reverse();
8713844×
                }
8723919×
                let Some((r, ty, body, packed)) = compressed_batch.pop() else {
8730×
                    continue;
874
                };
8753919×
                let r = &r;
8763919×
                let (flags, stored) = match packed {
8773386×
                    Some(c) => c,
878533×
                    None => (0, body.clone()),
879
                };
8803919×
                let header_len = 1 + 1 + 4 + 4 + 32;
8813919×
                f.write_all(&[type_index(ty), flags])?;
8823919×
                f.write_all(&(stored.len() as u32).to_le_bytes())?;
8833919×
                f.write_all(&(body.len() as u32).to_le_bytes())?;
8843919×
                f.write_all(&r.0)?;
8853919×
                f.write_all(&stored)?;
8863919×
                index.push((
8873919×
                    *r,
8883919×
                    PackEntry {
8893919×
                        offset: offset + header_len,
8903919×
                        stored_len: stored.len() as u32,
8913919×
                        raw_len: body.len() as u32,
8923919×
                        ty,
8933919×
                        flags,
8943919×
                    },
8953919×
                ));
8963919×
                offset += header_len + stored.len() as u64;
8973919×
                report.objects += 1;
8983919×
                report.raw_bytes += body.len() as u64;
8993919×
                report.packed_bytes += stored.len() as u64;
900
            }
90173×
            f.flush()?;
90273×
            f.get_ref().sync_all()?;
903
        }
904
        // The dictionary before the pack that needs it becomes visible.
90573×
        if let Some(d) = &dict {
9062×
            self.backend
9072×
                .put_new(&format!("pack-{id}.dict"), d)
9082×
                .map_err(backend_err)?;
90971×
        }
91073×
        progress(&format!("writing the index of {} objects", index.len()));
91173×
        self.finish_pack_files(&id, &tmp_pack, &index)?;
91273×
        for p in old_packs {
91349×
            self.remove_pack(p)?;
914
        }
91573×
        self.reload_packs()?;
9162219×
        for key in loose_keys {
9172219×
            delete(self.backend.as_ref(), key)?;
918
        }
91973×
        Ok(report)
92073×
    }
921
922
    /// A dictionary for these objects, trained on a sample of them (at
923
    /// most 10 000, 32 MB); `None` when training fails.
9242×
    fn train_dictionary(&self, refs: &[ObjectRef]) -> Option<Vec<u8>> {
9252×
        let step = (refs.len() / 10_000).max(1);
9262×
        let mut samples: Vec<Vec<u8>> = Vec::new();
9272×
        let mut total = 0usize;
9283001×
        for r in refs.iter().step_by(step) {
9293001×
            if total > 32 << 20 {
9300×
                break;
9313001×
            }
9323001×
            let Ok((ty, body)) = self.get(r) else {
9330×
                continue;
934
            };
9350×
            if matches!(
9363001×
                ty,
937
                ObjectType::Chunk | ObjectType::Blob | ObjectType::Change
9383001×
            ) && !body.is_empty()
9393001×
                && body.len() < 1 << 20
9403001×
            {
9413001×
                total += body.len();
9423001×
                samples.push(body);
9433001×
            }
944
        }
9452×
        zstd::dict::from_samples(&samples, DICT_BYTES).ok()
9462×
    }
947
948
    /// The scratch directory: for a store in a directory, that directory.
9497×
    pub fn dir(&self) -> &Path {
9507×
        &self.scratch
9517×
    }
952
953
    /// Where a loose object is kept, for a store in a directory (tests
954
    /// tamper with and remove objects there).
955
    #[cfg(test)]
9564×
    pub(crate) fn path_for(&self, r: &ObjectRef) -> PathBuf {
9574×
        self.scratch.join(loose_key(r))
9584×
    }
959
960
    /// Remove a loose object, whatever the store is on (tests make an
961
    /// object go missing).
962
    #[cfg(test)]
9633×
    pub(crate) fn remove_loose_for_test(&self, r: &ObjectRef) -> Result<()> {
9643×
        delete(self.backend.as_ref(), &loose_key(r))
9653×
    }
966
96712133×
    pub fn has(&self, r: &ObjectRef) -> bool {
96812133×
        self.backend.has(&loose_key(r)).unwrap_or(false)
9691305×
            || self.find_packed(r).is_some()
970930×
            || self
971930×
                .pending
972930×
                .borrow()
973930×
                .as_ref()
974930×
                .is_some_and(|p| p.lookup.contains_key(r))
97512133×
    }
976
977
    /// Start a write batch: until the matching `finish`, `put` collects
978
    /// objects and writes them out together, with one `fsync` for a large
979
    /// batch (a pack) instead of one per object. Batches nest; only the
980
    /// outermost does the work. Objects are readable while pending.
9812745×
    pub fn begin_batch(&self) -> Result<Batch<'_>> {
9822745×
        let depth = self.batch_depth.get();
9832745×
        if depth == 0 {
9842234×
            *self.pending.borrow_mut() = Some(Pending {
9852234×
                buffered: vec![],
9862234×
                buffered_bytes: 0,
9872234×
                lookup: Default::default(),
9882234×
                pack: None,
9892234×
                report: PackReport::default(),
9902234×
            });
9912234×
        }
9922745×
        self.batch_depth.set(depth + 1);
9932745×
        Ok(Batch {
9942745×
            store: self,
9952745×
            done: false,
9962745×
        })
9972745×
    }
998
9992744×
    fn finish_batch(&self) -> Result<()> {
10002744×
        let depth = self.batch_depth.get();
10012744×
        if depth == 0 {
10020×
            return Ok(());
10032744×
        }
10042744×
        self.batch_depth.set(depth - 1);
10052744×
        if depth > 1 {
1006511×
            return Ok(());
10072233×
        }
10082233×
        let Some(pending) = self.pending.borrow_mut().take() else {
10092×
            return Ok(());
1010
        };
10112231×
        self.write_pending(pending)
10122744×
    }
1013
1014
    /// Make every object written so far durable, even inside an open
1015
    /// batch, which stays open. Call before a reference that names them
1016
    /// is moved (spec §13.6): the operation head, a bookmark pointer.
10173466×
    pub fn flush(&self) -> Result<()> {
10183466×
        let Some(pending) = self.pending.borrow_mut().take() else {
10192467×
            return Ok(());
1020
        };
1021999×
        self.write_pending(pending)?;
1022997×
        *self.pending.borrow_mut() = Some(Pending {
1023997×
            buffered: vec![],
1024997×
            buffered_bytes: 0,
1025997×
            lookup: Default::default(),
1026997×
            pack: None,
1027997×
            report: PackReport::default(),
1028997×
        });
1029997×
        Ok(())
10303466×
    }
1031
10323230×
    fn write_pending(&self, pending: Pending) -> Result<()> {
10333230×
        match pending.pack {
1034
            None => {
10353226×
                let objects: Vec<(ObjectType, &[u8])> = pending
10363226×
                    .buffered
10373226×
                    .iter()
10385098×
                    .map(|(ty, body)| (*ty, body.as_slice()))
10393226×
                    .collect();
10403226×
                self.put_loose_many(&objects)
1041
            }
10424×
            Some(mut pack) => {
10434×
                pack.writer.flush()?;
10444×
                pack.writer.get_ref().sync_all()?;
10454×
                drop(pack.writer);
10464×
                pack.index.sort_by_key(|(r, _)| *r);
10474×
                self.finish_pack_files(&pack.id, &pack.tmp_path, &pack.index)?;
10484×
                Ok(())
1049
            }
1050
        }
10513230×
    }
1052
1053
    /// Whether a batch is open and holds `r`.
1054959×
    fn pending_get(&self, r: &ObjectRef) -> Option<Result<Object>> {
1055959×
        let mut guard = self.pending.borrow_mut();
1056959×
        let pending = guard.as_mut()?;
1057852×
        let entry = *pending.lookup.get(r)?;
1058852×
        Some(match entry {
1059670×
            PendingEntry::Buffered(i) => {
1060670×
                let (ty, body) = &pending.buffered[i];
1061670×
                Ok((*ty, body.clone()))
1062
            }
1063182×
            PendingEntry::Packed(e) => {
1064182×
                let Some(pack) = pending.pack.as_mut() else {
10650×
                    return Some(Err(StoreError::NotFound(*r)));
1066
                };
1067
                // Read from the pack file being written, on this disk.
1068182×
                (|| {
1069182×
                    pack.writer.flush()?;
1070
                    use std::io::{Read, Seek, SeekFrom};
1071182×
                    let mut f = fs::File::open(&pack.tmp_path)?;
1072182×
                    f.seek(SeekFrom::Start(e.offset))?;
1073182×
                    let mut stored = vec![0u8; e.stored_len as usize];
1074182×
                    f.read_exact(&mut stored)?;
1075182×
                    decode_entry(r, e, stored, None)
1076
                })()
1077
            }
1078
        })
1079959×
    }
1080
1081
    /// Append one object to the pending pack file.
10821900×
    fn pending_pack_write(
10831900×
        &self,
10841900×
        pack: &mut PendingPack,
10851900×
        r: &ObjectRef,
10861900×
        ty: ObjectType,
10871900×
        body: &[u8],
10881900×
        report: &mut PackReport,
10891900×
    ) -> Result<PackEntry> {
10901900×
        let (flags, stored) = if self.is_nocompress(r) {
1091120×
            (0, body.to_vec())
1092
        } else {
10931780×
            let compressed = zstd::bulk::compress(body, ZSTD_LEVEL)
10941780×
                .map_err(|e| StoreError::Corrupt(*r, format!("zstd: {e}")))?;
10951780×
            if compressed.len() < body.len() {
10961620×
                (FLAG_ZSTD, compressed)
1097
            } else {
1098160×
                (0, body.to_vec())
1099
            }
1100
        };
11011900×
        write_pack_entry(pack, r, ty, body.len(), flags, &stored, report)
11021900×
    }
1103
11047113×
    fn put_pending(&self, r: ObjectRef, ty: ObjectType, body: &[u8]) -> Result<()> {
11057113×
        let mut guard = self.pending.borrow_mut();
11067113×
        let pending = guard.as_mut().expect("batch open");
11077113×
        if pending.lookup.contains_key(&r) {
1108115×
            return Ok(());
11096998×
        }
11106998×
        if pending.pack.is_none()
11116032×
            && pending.buffered.len() < SPILL_OBJECTS
11126030×
            && pending.buffered_bytes + body.len() <= SPILL_BYTES
1113
        {
11146028×
            pending.buffered.push((ty, body.to_vec()));
11156028×
            pending.buffered_bytes += body.len();
11166028×
            pending
11176028×
                .lookup
11186028×
                .insert(r, PendingEntry::Buffered(pending.buffered.len() - 1));
11196028×
            return Ok(());
1120970×
        }
1121970×
        self.ensure_pack(pending)?;
1122970×
        let mut report = std::mem::take(&mut pending.report);
1123970×
        let pack = pending.pack.as_mut().unwrap();
1124970×
        let e = self.pending_pack_write(pack, &r, ty, body, &mut report)?;
1125970×
        pending.report = report;
1126970×
        pending.lookup.insert(r, PendingEntry::Packed(e));
1127970×
        Ok(())
11287113×
    }
1129
1130
    /// Open the pending batch's pack, moving the objects buffered so far
1131
    /// into it (a spill).
1132971×
    fn ensure_pack(&self, pending: &mut Pending) -> Result<()> {
1133971×
        if pending.pack.is_none() {
1134
            // Spill: open the pack and move the buffered objects into it.
11355×
            let id = fresh_id();
11365×
            let tmp_path = self.scratch.join(format!(".tmp-{id}.pack"));
11375×
            let mut writer = io::BufWriter::new(fs::File::create(&tmp_path)?);
11385×
            writer.write_all(PACK_MAGIC)?;
11395×
            let mut pack = PendingPack {
11405×
                id,
11415×
                tmp_path,
11425×
                writer,
11435×
                offset: PACK_MAGIC.len() as u64,
11445×
                index: vec![],
11455×
            };
11465×
            let buffered = std::mem::take(&mut pending.buffered);
11475×
            pending.buffered_bytes = 0;
11485×
            let mut report = std::mem::take(&mut pending.report);
1149930×
            for (bty, bbody) in &buffered {
1150930×
                let br = hash_object(*bty, bbody);
1151930×
                let e = self.pending_pack_write(&mut pack, &br, *bty, bbody, &mut report)?;
1152930×
                pending.lookup.insert(br, PendingEntry::Packed(e));
1153
            }
11545×
            pending.report = report;
11555×
            pending.pack = Some(pack);
1156966×
        }
1157971×
        Ok(())
1158971×
    }
1159
1160
    /// Store many objects at once. Inside a large write batch they are
1161
    /// compressed on every core and appended in order; otherwise each is
1162
    /// stored as [`Store::put`] does.
1163792×
    pub fn put_many(&self, objects: &[(ObjectType, &[u8])]) -> Result<Vec<ObjectRef>> {
1164
        use rayon::prelude::*;
1165823×
        let total: usize = objects.iter().map(|(_, b)| b.len()).sum();
1166792×
        if self.pending.borrow().is_none() || total < SPILL_BYTES {
1167791×
            return objects.iter().map(|(ty, b)| self.put(*ty, b)).collect();
11681×
        }
1169202×
        for (ty, b) in objects {
1170202×
            if b.len() > MAX_OBJECT_LEN as usize {
11710×
                return Err(StoreError::TooLarge {
11720×
                    kind: ty.name(),
11730×
                    size: b.len(),
11740×
                    limit: MAX_OBJECT_LEN,
11750×
                });
1176202×
            }
1177
        }
11781×
        let refs: Vec<ObjectRef> = objects
11791×
            .par_iter()
1180202×
            .map(|(ty, b)| hash_object(*ty, b))
11811×
            .collect();
1182
        // What is new: not stored already, not earlier in this call.
11831×
        let mut seen: std::collections::HashSet<ObjectRef> = std::collections::HashSet::new();
11841×
        let mut todo: Vec<usize> = Vec::new();
1185202×
        for (i, r) in refs.iter().enumerate() {
1186202×
            let known = self
1187202×
                .pending
1188202×
                .borrow()
1189202×
                .as_ref()
1190202×
                .is_some_and(|p| p.lookup.contains_key(r));
1191202×
            if seen.insert(*r)
1192201×
                && !known
1193201×
                && self.find_packed(r).is_none()
1194201×
                && !self.backend.has(&loose_key(r)).unwrap_or(false)
1195200×
            {
1196200×
                todo.push(i);
1197200×
            }
1198
        }
11991×
        let raw = self.nocompress_set()?;
12001×
        let packed: Vec<(u8, Vec<u8>)> = todo
12011×
            .par_iter()
1202200×
            .map(|i| {
1203200×
                let body = objects[*i].1;
1204200×
                if raw.contains(&refs[*i]) {
12050×
                    return (0, body.to_vec());
1206200×
                }
1207200×
                match zstd::bulk::compress(body, ZSTD_LEVEL) {
1208200×
                    Ok(c) if c.len() < body.len() => (FLAG_ZSTD, c),
12090×
                    _ => (0, body.to_vec()),
1210
                }
1211200×
            })
12121×
            .collect();
12131×
        let mut guard = self.pending.borrow_mut();
12141×
        let pending = guard.as_mut().expect("batch open");
12151×
        self.ensure_pack(pending)?;
12161×
        let mut report = std::mem::take(&mut pending.report);
12171×
        let pack = pending.pack.as_mut().unwrap();
1218200×
        for (i, (flags, stored)) in todo.iter().zip(packed) {
1219200×
            let (ty, body) = objects[*i];
1220200×
            let e = write_pack_entry(pack, &refs[*i], ty, body.len(), flags, &stored, &mut report)?;
1221200×
            pending.lookup.insert(refs[*i], PendingEntry::Packed(e));
1222
        }
12231×
        pending.report = report;
12241×
        Ok(refs)
1225792×
    }
1226
1227
    /// Store `body` as an object of type `ty`; idempotent.
122815911×
    pub fn put(&self, ty: ObjectType, body: &[u8]) -> Result<ObjectRef> {
1229
        // Refuse what the store could not read back (spec §13.2), before
1230
        // anything is written: a pack must never hold an entry its index
1231
        // would reject.
123215911×
        if body.len() > MAX_OBJECT_LEN as usize {
12331×
            return Err(StoreError::TooLarge {
12341×
                kind: ty.name(),
12351×
                size: body.len(),
12361×
                limit: MAX_OBJECT_LEN,
12371×
            });
123815910×
        }
123915910×
        let r = hash_object(ty, body);
124015910×
        if self.find_packed(&r).is_some() || self.backend.has(&loose_key(&r)).unwrap_or(false) {
12411802×
            return Ok(r);
124214108×
        }
124314108×
        if self.pending.borrow().is_some() {
12447113×
            self.put_pending(r, ty, body)?;
12457113×
            return Ok(r);
12466995×
        }
12476995×
        self.put_loose(ty, body)?;
12486990×
        Ok(r)
124915911×
    }
1250
1251
    /// Write one loose object, durable when this returns.
12527195×
    fn put_loose(&self, ty: ObjectType, body: &[u8]) -> Result<ObjectRef> {
12537195×
        let r = hash_object(ty, body);
12547195×
        let mut bytes = header(ty, body.len());
12557195×
        bytes.extend_from_slice(body);
12567195×
        crate::durable::step("write loose object")?;
12577190×
        put_new_or_existing(self.backend.as_ref(), &loose_key(&r), &bytes)?;
12587190×
        Ok(r)
12597195×
    }
1260
1261
    /// Write several loose objects with as few waits on the disk as the
1262
    /// guarantees allow (a directory store writes them all, then syncs
1263
    /// each directory once). Durable when this returns, as `put_loose`
1264
    /// one at a time would be (spec §13.6).
12653226×
    fn put_loose_many(&self, objects: &[(ObjectType, &[u8])]) -> Result<()> {
12663226×
        if objects.len() <= 1 {
12671839×
            for (ty, body) in objects {
1268200×
                self.put_loose(*ty, body)?;
1269
            }
12701839×
            return Ok(());
12711387×
        }
12721387×
        let mut seen = std::collections::HashSet::new();
12731387×
        let items: Vec<(String, Vec<u8>)> = objects
12741387×
            .iter()
12754898×
            .filter_map(|(ty, body)| {
12764898×
                let r = hash_object(*ty, body);
12774898×
                seen.insert(r).then(|| {
12784898×
                    let mut bytes = header(*ty, body.len());
12794898×
                    bytes.extend_from_slice(body);
12804898×
                    (loose_key(&r), bytes)
12814898×
                })
12824898×
            })
12831387×
            .collect();
12841387×
        self.backend.put_many_new(&items).map_err(backend_err)
12853226×
    }
1286
1287
    /// Read and verify an object, packed, loose, or pending in a batch.
1288
    /// The packs' indexes are in memory, so they are asked first: asking
1289
    /// for a loose copy of a packed object costs a request that finds
1290
    /// nothing on a store in a bucket, every time (oxygenhub#228).
129163284×
    pub fn get(&self, r: &ObjectRef) -> Result<Object> {
129263284×
        if let Some((i, e)) = self.find_packed(r) {
129315244×
            return self.packs.borrow()[i].fetch(self.backend.as_ref(), r, e);
129448040×
        }
129548040×
        match get_optional(self.backend.as_ref(), &loose_key(r))? {
129647088×
            Some(bytes) => parse_loose(&bytes, r),
1297952×
            None => self.pending_get(r).unwrap_or(Err(StoreError::NotFound(*r))),
1298
        }
129963284×
    }
1300
1301
    /// Many objects at once, read and verified in parallel: packed ones
1302
    /// from their packs, the others loose (as [`Store::get`], packs first),
1303
    /// then what is pending in a batch. Results are in the order of `refs`.
1304973×
    pub fn get_many(&self, refs: &[ObjectRef]) -> Vec<Result<Object>> {
1305
        use rayon::prelude::*;
1306973×
        let backend = self.backend.as_ref();
1307973×
        let packs = self.packs.borrow();
1308973×
        let place: Vec<Option<(usize, PackEntry)>> = refs
1309973×
            .iter()
13101978×
            .map(|r| {
13111978×
                packs
13121978×
                    .iter()
13131978×
                    .enumerate()
13141978×
                    .find_map(|(p, pack)| pack.find(r).map(|e| (p, e)))
13151978×
            })
1316973×
            .collect();
1317973×
        let indexes: &[PackIndex] = &packs;
1318973×
        let found: Vec<Option<Result<Object>>> = refs
1319973×
            .par_iter()
1320973×
            .zip(place.par_iter())
13211978×
            .map(|(r, at)| match at {
1322133×
                Some((p, e)) => Some(indexes[*p].fetch(backend, r, *e)),
13231845×
                None => match get_optional(backend, &loose_key(r)) {
13241838×
                    Ok(Some(bytes)) => Some(parse_loose(&bytes, r)),
13257×
                    Ok(None) => None,
13260×
                    Err(e) => Some(Err(e)),
1327
                },
13281978×
            })
1329973×
            .collect();
1330973×
        drop(packs);
1331973×
        found
1332973×
            .into_iter()
1333973×
            .zip(refs)
13341978×
            .map(|(f, r)| {
13351978×
                f.unwrap_or_else(|| self.pending_get(r).unwrap_or(Err(StoreError::NotFound(*r))))
13361978×
            })
1337973×
            .collect()
1338973×
    }
1339
134043955×
    pub fn get_typed(&self, r: &ObjectRef, want: ObjectType) -> Result<Vec<u8>> {
134143955×
        let (ty, body) = self.get(r)?;
134243922×
        if ty != want {
13431×
            return Err(StoreError::WrongType(*r, ty.name(), want.name()));
134443921×
        }
134543921×
        Ok(body)
134643955×
    }
1347
1348
    /// Store a change and the text blobs (and their chunks) its large
1349
    /// inserts are stored as (spec §6.1).
13502932×
    pub fn put_change(&self, c: &Change) -> Result<ObjectRef> {
13512932×
        let encoding = crate::timing::phase("store: encode the change and its texts");
13522932×
        let bytes = c.encode_streaming(|batch| self.put_texts(&batch))?;
13532932×
        drop(encoding);
13542932×
        let _t = crate::timing::phase("store: put the encoding");
13552932×
        self.put(ObjectType::Change, &bytes)
13562932×
    }
1357
1358
    /// Store text blobs and their chunks: every chunk at once (compressed
1359
    /// on every core in a large batch), then the blobs.
136062×
    pub fn put_texts(&self, texts: &[crate::objects::StoredText]) -> Result<()> {
136162×
        let mut chunks: Vec<(ObjectType, &[u8])> = Vec::new();
136263×
        for t in texts {
136363×
            if !self.has(&t.blob.hash()) {
136452×
                chunks.extend(
136552×
                    crate::chunk::chunks(&t.content)
136652×
                        .into_iter()
136793×
                        .map(|c| (ObjectType::Chunk, c)),
1368
                );
136911×
            }
1370
        }
137162×
        self.put_many(&chunks)?;
137263×
        for t in texts {
137363×
            self.put(ObjectType::Blob, &t.blob.encode())?;
1374
        }
137562×
        Ok(())
137662×
    }
1377
13784579×
    pub fn get_change(&self, r: &ObjectRef) -> Result<Change> {
13794579×
        Ok(self.get_change_parts(r)?.0)
13804579×
    }
1381
1382
    /// A change and the objects its stored inserts are made of: each text
1383
    /// blob and its chunks, which travel and are kept with it.
13844698×
    pub fn get_change_parts(&self, r: &ObjectRef) -> Result<(Change, Vec<ObjectRef>)> {
13854698×
        let bytes = self.get_typed(r, ObjectType::Change)?;
13864686×
        let mut parts = Vec::new();
13874686×
        let mut failed = None;
13884686×
        let decoded = Change::decode_with(&bytes, &mut |text, count, form| match self
138959×
            .read_text(text)
1390
        {
139158×
            Ok((blob, content)) => {
139258×
                parts.push(*text);
139358×
                parts.extend(blob.chunks);
139458×
                crate::objects::text_lines(&content, count, form)
1395
            }
13961×
            Err(e) => {
13971×
                let why = e.to_string();
13981×
                failed = Some(e);
13991×
                Err(ObjectError::Schema(format!(
14001×
                    "text blob {}: {why}",
14011×
                    text.short(12)
14021×
                )))
1403
            }
140459×
        });
14054686×
        match (decoded, failed) {
14064685×
            (Ok(c), _) => Ok((c, parts)),
14071×
            (Err(_), Some(StoreError::NotFound(t))) => Err(StoreError::Corrupt(
14081×
                *r,
14091×
                format!("its text blob {} is missing", t.short(12)),
14101×
            )),
14110×
            (Err(e), _) => Err(StoreError::Corrupt(*r, e.to_string())),
1412
        }
14134698×
    }
1414
1415
    /// A change with every insert left without its lines (empty
1416
    /// placeholders), and where each one's lines are, in order: for
1417
    /// history-only states, which read text only for the graphs they build
1418
    /// (see [`Change::decode_skeleton`]).
1419537×
    pub fn get_change_lazy(&self, r: &ObjectRef) -> Result<(Change, Vec<crate::objects::TextRef>)> {
1420537×
        let bytes = self.get_typed(r, ObjectType::Change)?;
1421535×
        Change::decode_skeleton(&bytes, *r).map_err(|e| StoreError::Corrupt(*r, e.to_string()))
1422537×
    }
1423
1424
    /// A change's metadata (id, dependencies, author, clock, message, kind)
1425
    /// without its contents: read from its skeleton, so neither its text
1426
    /// blobs nor its inline lines are read (spec §11).
1427
    ///
1428
    /// Served from the `metas` cache when this store read it before (a
1429
    /// change object can be as large as the tree it adds), as long as the
1430
    /// store still has the change.
14314546×
    pub fn get_change_meta(&self, r: &ObjectRef) -> Result<crate::state::ChangeMeta> {
14324546×
        if self.metas.borrow().is_none() {
1433948×
            *self.metas.borrow_mut() = Some(self.read_metas_cache());
14343598×
        }
14354546×
        let cached = self.metas.borrow().as_ref().and_then(|m| m.get(r).cloned());
14364546×
        if let Some(meta) = cached {
14373265×
            if self.has(r) {
1438
                #[cfg(test)]
14391803×
                assert_eq!(
1440
                    meta,
14411803×
                    Change::decode_meta(&self.get_typed(r, ObjectType::Change)?).unwrap(),
1442
                    "the metadata cache holds a change's metadata"
1443
                );
14443265×
                return Ok(meta);
14450×
            }
14461281×
        }
14471281×
        let bytes = self.get_typed(r, ObjectType::Change)?;
14481281×
        let meta =
14491281×
            Change::decode_meta(&bytes).map_err(|e| StoreError::Corrupt(*r, e.to_string()))?;
14501281×
        self.remember_meta(r, &meta);
14511281×
        Ok(meta)
14524546×
    }
1453
1454
    /// The metadata cache: records of a change hash, its metadata in the
1455
    /// flat layout, and the first 8 bytes of BLAKE3 over both, each
1456
    /// appended in one write. Derived only: a record that does not check
1457
    /// ends the reading (a torn write), and the cache can be deleted. It
1458
    /// lives in the scratch directory, on this disk.
1459948×
    fn read_metas_cache(&self) -> crate::fasthash::FastMap<ObjectRef, crate::state::ChangeMeta> {
1460948×
        let mut out = crate::fasthash::FastMap::default();
1461948×
        let Ok(bytes) = fs::read(self.scratch.join("metas")) else {
1462364×
            return out;
1463
        };
1464584×
        let mut r = crate::flat::Reader::new(&bytes);
1465584×
        let mut good = 0;
14662075×
        while !r.at_end() {
14671492×
            let mut record = || -> std::result::Result<_, ObjectError> {
14681492×
                let h = r.object()?;
14691491×
                let body = r.bytes()?;
14701491×
                let check = r.fixed::<8>()?;
14711491×
                Ok((h, body, check))
14721492×
            };
14731492×
            let Ok((h, body, check)) = record() else {
14741×
                break;
1475
            };
14761491×
            let mut sum = blake3::Hasher::new();
14771491×
            sum.update(&h.0).update(body);
14781491×
            if sum.finalize().as_bytes()[..8] != check {
14790×
                break;
14801491×
            }
14811491×
            match crate::state::ChangeMeta::read_flat(&mut crate::flat::Reader::new(body)) {
14821491×
                Ok(m) => out.insert(h, m),
14830×
                Err(_) => break,
1484
            };
14851491×
            good += 32 + 4 + body.len() + 8;
1486
        }
1487
        // Cut a bad tail, so records appended later are read.
1488584×
        if good < bytes.len() {
14891×
            let _ = fs::OpenOptions::new()
14901×
                .write(true)
14911×
                .open(self.scratch.join("metas"))
14921×
                .and_then(|f| f.set_len(good as u64));
1493583×
        }
1494584×
        out
1495948×
    }
1496
1497
    /// Add a change's metadata to the cache; quietly nothing when it
1498
    /// cannot be written (it is only a cache).
14991281×
    fn remember_meta(&self, r: &ObjectRef, meta: &crate::state::ChangeMeta) {
15001281×
        let mut body = crate::flat::Writer::default();
15011281×
        meta.write_flat(&mut body);
15021281×
        let mut w = crate::flat::Writer::default();
15031281×
        w.object(r);
15041281×
        w.bytes(&body.0);
15051281×
        let mut sum = blake3::Hasher::new();
15061281×
        sum.update(&r.0).update(&body.0);
15071281×
        w.fixed(&sum.finalize().as_bytes()[..8]);
15081281×
        let _ = fs::OpenOptions::new()
15091281×
            .create(true)
15101281×
            .append(true)
15111281×
            .open(self.scratch.join("metas"))
15121281×
            .and_then(|mut f| f.write_all(&w.0));
15131281×
        if let Some(m) = self.metas.borrow_mut().as_mut() {
15141281×
            m.insert(*r, meta.clone());
15151281×
        }
15161281×
    }
1517
151859×
    fn read_text(&self, r: &ObjectRef) -> Result<(Blob, Vec<u8>)> {
151959×
        let blob = Blob::decode(&self.get_typed(r, ObjectType::Blob)?)?;
152058×
        let mut content = Vec::with_capacity(blob.length as usize);
152198×
        for c in &blob.chunks {
152298×
            content.extend(self.get_typed(c, ObjectType::Chunk)?);
1523
        }
152458×
        Ok((blob, content))
152559×
    }
1526
1527194×
    pub fn put_chunk(&self, bytes: &[u8]) -> Result<ObjectRef> {
1528194×
        self.put(ObjectType::Chunk, bytes)
1529194×
    }
1530
1531
    /// Store a chunk that packs will keep uncompressed (`nocompress`, spec §13.3).
1532123×
    pub fn put_chunk_nocompress(&self, bytes: &[u8]) -> Result<ObjectRef> {
1533123×
        let r = hash_object(ObjectType::Chunk, bytes);
1534123×
        put_new_or_existing(
1535123×
            self.backend.as_ref(),
1536123×
            &format!("nocompress/{}", r.to_hex()),
1537123×
            b"",
15380×
        )?;
1539123×
        self.put(ObjectType::Chunk, bytes)
1540123×
    }
1541
15421900×
    fn is_nocompress(&self, r: &ObjectRef) -> bool {
15431900×
        self.backend
15441900×
            .has(&format!("nocompress/{}", r.to_hex()))
15451900×
            .unwrap_or(false)
15461900×
    }
1547
1548
    /// Every chunk marked to be kept raw.
154974×
    fn nocompress_set(&self) -> Result<std::collections::HashSet<ObjectRef>> {
155074×
        Ok(self
155174×
            .backend
155274×
            .list("nocompress/")
155374×
            .map_err(backend_err)?
155474×
            .into_iter()
155574×
            .filter_map(|e| ObjectRef::from_hex(e.key.trim_start_matches("nocompress/")))
155674×
            .collect())
155774×
    }
1558
15592956×
    pub fn put_state(&self, st: &StateObject) -> Result<ObjectRef> {
15602956×
        self.put(ObjectType::State, &st.encode())
15612956×
    }
1562
15635320×
    pub fn get_state(&self, r: &ObjectRef) -> Result<StateObject> {
15645320×
        Ok(StateObject::decode(&self.get_typed(r, ObjectType::State)?)?)
15655320×
    }
1566
1567389×
    pub fn put_attest(&self, a: &Attest) -> Result<ObjectRef> {
1568389×
        self.put(ObjectType::Attest, &a.encode())
1569389×
    }
1570
1571619×
    pub fn get_attest(&self, r: &ObjectRef) -> Result<Attest> {
1572619×
        Ok(Attest::decode(&self.get_typed(r, ObjectType::Attest)?)?)
1573619×
    }
1574
15753×
    pub fn put_receipt(&self, r: &Receipt) -> Result<ObjectRef> {
15763×
        self.put(ObjectType::Receipt, &r.encode())
15773×
    }
1578
157912×
    pub fn get_receipt(&self, r: &ObjectRef) -> Result<Receipt> {
158012×
        Ok(Receipt::decode(&self.get_typed(r, ObjectType::Receipt)?)?)
158112×
    }
1582
1583
    /// Record `value` under `key` in `index`: one write-once entry, so two
1584
    /// writers of a shared store never lose each other's.
1585180×
    fn index_add(&self, index: &str, key: &ObjectRef, value: &ObjectRef) -> Result<()> {
1586180×
        let entry = format!("{}{}", fanout_prefix(index, key), value.to_hex());
1587180×
        put_new_or_existing(self.backend.as_ref(), &entry, b"")
1588180×
    }
1589
1590
    /// The values recorded under `key` in `index`, with those an older
1591
    /// store kept as one list in `legacy`, sorted.
15922123×
    fn index_get(&self, index: &str, legacy: &str, key: &ObjectRef) -> Result<Vec<ObjectRef>> {
15932123×
        let mut values: Vec<ObjectRef> = self
15942123×
            .backend
15952123×
            .list(&fanout_prefix(index, key))
15962123×
            .map_err(backend_err)?
15972123×
            .into_iter()
15982123×
            .filter_map(|e| e.key.rsplit('/').next().and_then(ObjectRef::from_hex))
15992123×
            .collect();
16002123×
        if let Some(bytes) = get_optional(self.backend.as_ref(), &legacy_key(legacy, key))? {
16013×
            values.extend(
16023×
                String::from_utf8_lossy(&bytes)
16033×
                    .lines()
16043×
                    .filter_map(ObjectRef::from_hex),
16053×
            );
16062120×
        }
16072123×
        values.sort();
16082123×
        values.dedup();
16092123×
        Ok(values)
16102123×
    }
1611
1612
    /// Record that seal attestation `attest` covers `change`.
1613151×
    pub fn index_seal(&self, change: &ObjectRef, attest: &ObjectRef) -> Result<()> {
1614151×
        self.index_add("seal", change, attest)
1615151×
    }
1616
1617
    /// Seal attestations known to cover `change`.
16181410×
    pub fn seals_for(&self, change: &ObjectRef) -> Result<Vec<ObjectRef>> {
16191410×
        self.index_get("seal", "seals", change)
16201410×
    }
1621
1622
    /// Record receipt attestation `attest` for `state` (spec §14.6).
1623
    /// Keep a short-lived key's certificate (spec §14.7), with its issuer's
1624
    /// signature, indexed by the key it certifies.
162522×
    pub fn put_certificate(&self, c: &crate::cert::SignedCertificate) -> Result<ObjectRef> {
162622×
        let r = self.put(ObjectType::Certificate, &c.cert.encode())?;
162722×
        self.put_signature(&r, &c.sig)?;
162822×
        self.index_add("cert", &c.cert.key_id(), &r)?;
162922×
        Ok(r)
163022×
    }
1631
1632
    /// A kept certificate with its issuer's signature.
163361×
    pub fn get_certificate(&self, r: &ObjectRef) -> Result<crate::cert::SignedCertificate> {
163461×
        let cert = crate::cert::Certificate::decode(&self.get_typed(r, ObjectType::Certificate)?)?;
163561×
        let sig = self
163661×
            .signatures(r)?
163761×
            .into_iter()
163861×
            .find(|s| s.key == cert.issuer)
163961×
            .ok_or(StoreError::NotFound(*r))?;
164061×
        Ok(crate::cert::SignedCertificate { cert, sig })
164161×
    }
1642
1643
    /// The certificate kept for short-lived key `key`, if any.
1644569×
    pub fn certificate_of(&self, key: &ObjectRef) -> Result<Option<ObjectRef>> {
1645569×
        Ok(self
1646569×
            .index_get("cert", "certs", key)?
1647569×
            .into_iter()
1648569×
            .find(|r| self.has(r)))
1649569×
    }
1650
1651
    /// Record `value` under `key` in index `index` (for repository code).
16521×
    pub(crate) fn index_add_pub(
16531×
        &self,
16541×
        index: &str,
16551×
        key: &ObjectRef,
16561×
        value: &ObjectRef,
16571×
    ) -> Result<()> {
16581×
        self.index_add(index, key, value)
16591×
    }
1660
1661
    /// The values recorded under `key` in index `index` (which has no
1662
    /// older one-list form: the legacy name only has to differ from it).
1663127×
    pub(crate) fn index_get_pub(&self, index: &str, key: &ObjectRef) -> Result<Vec<ObjectRef>> {
1664127×
        self.index_get(index, &format!("{index}-lists"), key)
1665127×
    }
1666
16675×
    pub fn index_receipt(&self, state: &ObjectRef, attest: &ObjectRef) -> Result<()> {
16685×
        self.index_add("receipt", state, attest)
16695×
    }
1670
1671
    /// Receipt attestations recorded for `state`.
167214×
    pub fn receipts_for(&self, state: &ObjectRef) -> Result<Vec<ObjectRef>> {
167314×
        self.index_get("receipt", "receipts", state)
167414×
    }
1675
16761×
    pub fn index_approval(&self, state: &ObjectRef, attest: &ObjectRef) -> Result<()> {
16771×
        self.index_add("approval", state, attest)
16781×
    }
1679
1680
    /// Approval attestations recorded for `state` (OxygenFlow approval
1681
    /// gates, oxygenhub#235).
16823×
    pub fn approvals_for(&self, state: &ObjectRef) -> Result<Vec<ObjectRef>> {
16833×
        self.index_get("approval", "approvals", state)
16843×
    }
1685
1686
    /// Every state with an approval, for garbage collection.
168718×
    pub fn approval_states(&self) -> Result<Vec<ObjectRef>> {
168818×
        self.indexed_states("approval", "approvals")
168918×
    }
1690
1691
    /// Every state with a receipt, for garbage collection.
169224×
    pub fn receipt_states(&self) -> Result<Vec<ObjectRef>> {
169324×
        self.indexed_states("receipt", "receipts")
169424×
    }
1695
1696
    /// The states an index (`one/<hh>/<rest>/<attest>` entries, and the
1697
    /// packed `many/<hh>/<rest>` lists) holds attestations for.
169842×
    fn indexed_states(&self, one: &str, many: &str) -> Result<Vec<ObjectRef>> {
169942×
        let mut out: Vec<ObjectRef> = Vec::new();
170042×
        for e in self.backend.list(&format!("{one}/")).map_err(backend_err)? {
1701
            // receipt/<hh>/<rest>/<attest>
17025×
            let mut parts = e.key.splitn(4, '/');
17035×
            let (Some(_), Some(hh), Some(rest), Some(_)) =
17045×
                (parts.next(), parts.next(), parts.next(), parts.next())
1705
            else {
17060×
                continue;
1707
            };
17085×
            out.extend(ObjectRef::from_hex(&format!("{hh}{rest}")));
1709
        }
171042×
        for e in self
171142×
            .backend
171242×
            .list(&format!("{many}/"))
171342×
            .map_err(backend_err)?
1714
        {
17151×
            let mut parts = e.key.splitn(3, '/');
17161×
            let (Some(_), Some(hh), Some(rest)) = (parts.next(), parts.next(), parts.next()) else {
17170×
                continue;
1718
            };
17191×
            out.extend(ObjectRef::from_hex(&format!("{hh}{rest}")));
1720
        }
172142×
        out.sort();
172242×
        out.dedup();
172342×
        Ok(out)
172442×
    }
1725
1726296×
    pub fn put_identity(&self, id: &Identity) -> Result<ObjectRef> {
1727296×
        self.put(ObjectType::Identity, &id.encode())
1728296×
    }
1729
1730785×
    pub fn get_identity(&self, r: &ObjectRef) -> Result<Identity> {
1731785×
        Ok(Identity::decode(&self.get_typed(r, ObjectType::Identity)?)?)
1732785×
    }
1733
1734
    /// Signatures stored alongside object `r` (spec §5.4), one per signing
1735
    /// key, sorted by key. Empty if none.
17362813×
    pub fn signatures(&self, r: &ObjectRef) -> Result<Vec<Signature>> {
17372813×
        let mut sigs: Vec<Signature> = Vec::new();
17382813×
        for e in self
17392813×
            .backend
17402813×
            .list(&fanout_prefix("sig", r))
17412813×
            .map_err(backend_err)?
1742
        {
17431593×
            let bytes = self.backend.get(&e.key).map_err(backend_err)?;
17441593×
            sigs.extend(Signature::decode_list(&bytes)?);
1745
        }
1746
        // An older store's one list per object; a signer kept there and
1747
        // here counts once, as written here.
17482813×
        if let Some(bytes) = get_optional(self.backend.as_ref(), &legacy_key("sigs", r))? {
17492×
            for sig in Signature::decode_list(&bytes)? {
17502×
                if !sigs.iter().any(|s| s.key == sig.key) {
17511×
                    sigs.push(sig);
17521×
                }
1753
            }
17542811×
        }
17552813×
        sigs.sort_by_key(|s| s.key);
17562813×
        sigs.dedup_by_key(|s| s.key);
17572813×
        Ok(sigs)
17582813×
    }
1759
1760
    /// Attach a signature to object `r`, replacing any earlier one by the
1761
    /// same key: one entry per signer, so the latest by a key is the one
1762
    /// kept, and two signers never touch each other's.
1763599×
    pub fn put_signature(&self, r: &ObjectRef, sig: &Signature) -> Result<()> {
1764599×
        let key = format!("{}{}", fanout_prefix("sig", r), sig.key.to_hex());
1765599×
        let bytes = Signature::encode_list(std::slice::from_ref(sig));
1766599×
        if get_optional(self.backend.as_ref(), &key)?.as_deref() == Some(bytes.as_slice()) {
176723×
            return Ok(());
1768576×
        }
1769576×
        delete(self.backend.as_ref(), &key)?;
1770576×
        put_new_or_existing(self.backend.as_ref(), &key, &bytes)
1771599×
    }
1772
1773
    /// Finish packs a writer left unfinished: a batch writes its objects to
1774
    /// `.tmp-ID.pack` and only indexes it at the end, so a process that
1775
    /// stopped part way (an import killed after hours) leaves every object
1776
    /// in a pack nothing can read. Each such pack untouched for `min_age`
1777
    /// is read through: every whole entry checked against its reference,
1778
    /// a torn last one cut off, then indexed as a pack of the store.
1779
    /// Returns the packs and objects adopted.
17802×
    pub fn adopt_unfinished_packs(
17812×
        &self,
17822×
        min_age: std::time::Duration,
17832×
        progress: &dyn Fn(&str),
17842×
    ) -> Result<(usize, usize)> {
1785
        use std::io::{BufReader, Read};
17862×
        let mut packs = 0;
17872×
        let mut objects = 0;
17882×
        let Ok(dir) = fs::read_dir(&self.scratch) else {
17890×
            return Ok((0, 0));
1790
        };
17914×
        for entry in dir.flatten() {
17924×
            let name = entry.file_name().to_string_lossy().to_string();
17934×
            let Some(id) = name
17944×
                .strip_prefix(".tmp-")
17954×
                .and_then(|n| n.strip_suffix(".pack"))
1796
            else {
17972×
                continue;
1798
            };
17992×
            let path = entry.path();
18002×
            let meta = fs::metadata(&path)?;
18012×
            let age = meta
18022×
                .modified()
18032×
                .ok()
18042×
                .and_then(|m| m.elapsed().ok())
18052×
                .unwrap_or_default();
18062×
            if age < min_age {
18071×
                progress(&format!(
18081×
                    "{name}: written to {}s ago, left alone",
18091×
                    age.as_secs()
18101×
                ));
18111×
                continue;
18121×
            }
18131×
            let mut f = BufReader::with_capacity(1 << 20, fs::File::open(&path)?);
18141×
            let mut magic = [0u8; 8];
18151×
            if f.read_exact(&mut magic).is_err() || &magic != PACK_MAGIC {
18160×
                progress(&format!("{name}: not a pack, left alone"));
18170×
                continue;
18181×
            }
18191×
            let mut offset = PACK_MAGIC.len() as u64;
18201×
            let mut index: Vec<(ObjectRef, PackEntry)> = Vec::new();
1821
            loop {
1822301×
                let mut head = [0u8; 42];
1823301×
                if f.read_exact(&mut head).is_err() {
18240×
                    break;
1825301×
                }
1826301×
                let Some(ty) = ObjectType::ALL.get(head[0] as usize).copied() else {
18270×
                    break;
1828
                };
1829301×
                let flags = head[1];
1830301×
                let stored_len = u32::from_le_bytes(head[2..6].try_into().unwrap());
1831301×
                let raw_len = u32::from_le_bytes(head[6..10].try_into().unwrap());
1832301×
                let r = ObjectRef(head[10..42].try_into().unwrap());
1833301×
                let mut stored = vec![0u8; stored_len as usize];
1834301×
                if flags & FLAG_DICT != 0 || f.read_exact(&mut stored).is_err() {
18351×
                    break;
1836300×
                }
1837300×
                let e = PackEntry {
1838300×
                    offset: offset + 42,
1839300×
                    stored_len,
1840300×
                    raw_len,
1841300×
                    ty,
1842300×
                    flags,
1843300×
                };
1844300×
                if decode_entry(&r, e, stored, None).is_err() {
18450×
                    break;
1846300×
                }
1847300×
                index.push((r, e));
1848300×
                offset += 42 + stored_len as u64;
1849300×
                if index.len().is_multiple_of(200_000) {
18500×
                    progress(&format!(
18510×
                        "{name}: {} objects checked, {} MiB",
18520×
                        index.len(),
18530×
                        offset >> 20
18540×
                    ));
1855300×
                }
1856
            }
18571×
            if index.is_empty() {
18580×
                progress(&format!("{name}: no whole object, left alone"));
18590×
                continue;
18601×
            }
18611×
            if offset < meta.len() {
18621×
                progress(&format!(
18631×
                    "{name}: {} torn bytes at the end cut off",
18641×
                    meta.len() - offset
18651×
                ));
18661×
                fs::OpenOptions::new()
18671×
                    .write(true)
18681×
                    .open(&path)?
18691×
                    .set_len(offset)?;
18700×
            }
18711×
            index.sort_by_key(|(r, _)| *r);
18721×
            index.dedup_by_key(|(r, _)| *r);
18731×
            objects += index.len();
18741×
            packs += 1;
18751×
            progress(&format!("{name}: {} objects indexed", index.len()));
18761×
            self.finish_pack_files(id, &path, &index)?;
1877
        }
18782×
        Ok((packs, objects))
18792×
    }
1880
1881
    /// Every object reference in the store, loose and packed (slow; for gc and tests).
188286×
    pub fn list(&self) -> io::Result<Vec<ObjectRef>> {
188386×
        let mut out: Vec<ObjectRef> = self
188486×
            .loose_objects()
188586×
            .map_err(|e| io::Error::other(e.to_string()))?
188686×
            .into_iter()
188786×
            .map(|(r, _)| r)
188886×
            .collect();
188986×
        for p in self.packs.borrow().iter() {
189010×
            out.extend(p.refs.iter().copied());
189110×
        }
189286×
        out.sort();
189386×
        out.dedup();
189486×
        Ok(out)
189586×
    }
1896
}
1897
1898
/// A loose object's bytes, verified against `r`.
189948926×
fn parse_loose(data: &[u8], r: &ObjectRef) -> Result<Object> {
190048926×
    let nul = data
190148926×
        .iter()
1902696511×
        .position(|&b| b == 0)
190348926×
        .ok_or_else(|| StoreError::Corrupt(*r, "no header".into()))?;
190448926×
    let head = std::str::from_utf8(&data[..nul])
190548926×
        .map_err(|_| StoreError::Corrupt(*r, "header not UTF-8".into()))?;
190648926×
    let mut parts = head.split(' ');
190748926×
    let (Some("oxy1"), Some(ty), Some(len), None) =
190848926×
        (parts.next(), parts.next(), parts.next(), parts.next())
1909
    else {
19100×
        return Err(StoreError::Corrupt(*r, format!("bad header {head:?}")));
1911
    };
191248926×
    let ty = ObjectType::from_name(ty)
191348926×
        .ok_or_else(|| StoreError::Corrupt(*r, format!("unknown type {ty}")))?;
191448926×
    let len: usize = len
191548926×
        .parse()
191648926×
        .map_err(|_| StoreError::Corrupt(*r, "bad length".into()))?;
191748926×
    let body = &data[nul + 1..];
191848926×
    if body.len() != len {
19190×
        return Err(StoreError::Corrupt(
19200×
            *r,
19210×
            format!("length {} != header {len}", body.len()),
19220×
        ));
192348926×
    }
192448926×
    if hash_object(ty, body) != *r {
19254×
        return Err(StoreError::Corrupt(*r, "hash mismatch".into()));
192648922×
    }
192748922×
    Ok((ty, body.to_vec()))
192848926×
}
1929
1930
/// Whether the objects are files that a test may tamper with: not under
1931
/// `OXY_STORE_BACKEND=memory`.
1932
#[cfg(test)]
19336×
fn files_on_disk() -> bool {
19346×
    std::env::var_os("OXY_STORE_BACKEND").is_none()
19356×
}
1936
1937
#[cfg(test)]
1938
mod tests {
1939
    use super::*;
1940
1941
    /// The store under test twice: in a directory, and on a store in
1942
    /// memory with a scratch directory, as over a bucket.
19436×
    fn each_store(dir: &Path, f: impl Fn(&str, Store)) {
19446×
        f("directory", Store::open(dir.join("objects")).unwrap());
19456×
        f(
19466×
            "memory",
19476×
            Store::open_on(Arc::new(MemoryStore::new()), dir.join("scratch")).unwrap(),
19486×
        );
19496×
    }
1950
1951
    #[test]
19521×
    fn put_get_verify() {
19531×
        if !super::files_on_disk() {
19540×
            return;
19551×
        }
19561×
        let dir = tempfile::tempdir().unwrap();
19571×
        let s = Store::open(dir.path().join("objects")).unwrap();
19581×
        let r = s.put(ObjectType::Chunk, b"hello").unwrap();
19591×
        assert!(s.has(&r));
19601×
        assert_eq!(s.get(&r).unwrap(), (ObjectType::Chunk, b"hello".to_vec()));
19611×
        assert_eq!(s.put(ObjectType::Chunk, b"hello").unwrap(), r);
19621×
        assert_eq!(s.list().unwrap(), vec![r]);
19631×
        assert!(matches!(
19641×
            s.get_typed(&r, ObjectType::Blob),
1965
            Err(StoreError::WrongType(..))
1966
        ));
19671×
        assert!(matches!(
19681×
            s.get(&ObjectRef([9; 32])),
1969
            Err(StoreError::NotFound(_))
1970
        ));
1971
1972
        // Tamper: flip a byte, read must fail.
19731×
        let path = s.path_for(&r);
19741×
        let mut data = fs::read(&path).unwrap();
19751×
        let last = data.len() - 1;
19761×
        data[last] ^= 1;
19771×
        fs::write(&path, data).unwrap();
19781×
        assert!(matches!(s.get(&r), Err(StoreError::Corrupt(..))));
19791×
    }
1980
1981
    #[test]
19821×
    fn packing_keeps_every_object_readable() {
19831×
        if !super::files_on_disk() {
19840×
            return;
19851×
        }
19861×
        let dir = tempfile::tempdir().unwrap();
19871×
        let s = Store::open(dir.path()).unwrap();
19881×
        let mut refs = Vec::new();
1989300×
        for i in 0..300u32 {
1990300×
            let body = format!("object {i} ").repeat(1 + (i % 7) as usize);
1991300×
            refs.push((s.put(ObjectType::Chunk, body.as_bytes()).unwrap(), body));
1992300×
        }
19931×
        let c = crate::objects::tests::sample_change();
19941×
        let cr = s.put_change(&c).unwrap();
19951×
        assert_eq!(s.loose_count().unwrap(), 301);
19961×
        let report = s.pack_loose().unwrap();
19971×
        assert_eq!(report.objects, 301);
19981×
        assert!(report.packed_bytes < report.raw_bytes, "{report:?}");
19991×
        assert_eq!(s.loose_count().unwrap(), 0);
20001×
        assert_eq!(s.pack_count(), 1);
2001300×
        for (r, body) in &refs {
2002300×
            assert!(s.has(r));
2003300×
            assert_eq!(
2004300×
                s.get(r).unwrap(),
2005300×
                (ObjectType::Chunk, body.as_bytes().to_vec())
2006
            );
2007
        }
20081×
        assert_eq!(s.get_change(&cr).unwrap(), c);
20091×
        assert_eq!(s.list().unwrap().len(), 301);
2010
        // Re-putting a packed object is a no-op; new objects go loose and a
2011
        // second pack coexists with the first.
20121×
        assert_eq!(
20131×
            s.put(ObjectType::Chunk, refs[0].1.as_bytes()).unwrap(),
20141×
            refs[0].0
2015
        );
20161×
        assert_eq!(s.loose_count().unwrap(), 0);
20171×
        let extra = s.put(ObjectType::Chunk, b"after packing").unwrap();
20181×
        s.pack_loose().unwrap();
20191×
        assert_eq!(s.pack_count(), 2);
20201×
        assert_eq!(s.get(&extra).unwrap().1, b"after packing");
2021
        // A reopened store sees the packs.
20221×
        let s2 = Store::open(dir.path()).unwrap();
20231×
        assert_eq!(s2.pack_count(), 2);
20241×
        assert_eq!(s2.get(&refs[5].0).unwrap().1, refs[5].1.as_bytes());
20251×
        assert!(matches!(
20261×
            s2.get(&ObjectRef([9; 32])),
2027
            Err(StoreError::NotFound(_))
2028
        ));
2029
        // Corrupting an object's body inside the pack is detected on read.
20301×
        let target = refs[5].0;
20311×
        let (pack_path, offset) = {
20321×
            let packs = s2.packs.borrow();
20331×
            let (i, e) = s2.find_packed(&target).unwrap();
20341×
            (dir.path().join(&packs[i].key), e.offset as usize)
20351×
        };
20361×
        let mut data = fs::read(&pack_path).unwrap();
20371×
        data[offset] ^= 0xff;
20381×
        fs::write(&pack_path, data).unwrap();
20391×
        let s3 = Store::open(dir.path()).unwrap();
20401×
        assert!(matches!(s3.get(&target), Err(StoreError::Corrupt(..))));
20411×
        assert_eq!(
20421×
            s3.get(&refs[6].0).unwrap().1,
20431×
            refs[6].1.as_bytes(),
2044
            "other objects are unaffected"
2045
        );
20461×
    }
2047
2048
    #[test]
20491×
    fn a_full_repack_trains_a_dictionary_and_reads_back() {
20501×
        if !super::files_on_disk() {
20510×
            return;
20521×
        }
20531×
        let dir = tempfile::tempdir().unwrap();
20541×
        let s = Store::open(dir.path()).unwrap();
2055
        // Many small, alike files: what a dictionary is for.
20561×
        let bodies: Vec<Vec<u8>> = (0..1500)
20571500×
            .map(|i| {
20581500×
                format!(
2059
                    "// SPDX-License-Identifier: GPL-2.0\n#include <linux/module.h>\n\
2060
                     static int probe_{i}(struct device *dev)\n{{\n\treturn {i};\n}}\n"
2061
                )
20621500×
                .into_bytes()
20631500×
            })
20641×
            .collect();
20651×
        let refs: Vec<ObjectRef> = bodies
20661×
            .iter()
20671500×
            .map(|b| s.put(ObjectType::Chunk, b).unwrap())
20681×
            .collect();
20691×
        s.repack_all(REPACK_ZSTD_LEVEL).unwrap();
20701×
        let dicts = fs::read_dir(dir.path())
20711×
            .unwrap()
20721×
            .flatten()
20733×
            .filter(|e| e.path().extension().is_some_and(|x| x == "dict"))
20741×
            .count();
20751×
        assert_eq!(dicts, 1, "the pack has a dictionary");
20761×
        let with_dict = refs
20771×
            .iter()
20781500×
            .filter(|r| s.find_packed(r).unwrap().1.flags & FLAG_DICT != 0)
20791×
            .count();
20801×
        assert!(with_dict > 1000, "most use it: {with_dict}");
20811×
        let again = Store::open(dir.path()).unwrap();
20821500×
        for (r, b) in refs.iter().zip(&bodies) {
20831500×
            assert_eq!(&again.get(r).unwrap().1, b);
2084
        }
20853×
        let packs = |dir: &Path| {
20863×
            let mut v: Vec<String> = fs::read_dir(dir)
20873×
                .unwrap()
20883×
                .flatten()
208911×
                .map(|e| e.file_name().to_string_lossy().into_owned())
209011×
                .filter(|n| n.starts_with("pack-"))
20913×
                .collect();
20923×
            v.sort();
20933×
            v
20943×
        };
2095
        // With nothing new, a repack leaves that pack as it is: it would
2096
        // only write the same again.
20971×
        let before = packs(dir.path());
20981×
        let (report, merged) = again.repack_all(REPACK_ZSTD_LEVEL).unwrap();
20991×
        assert_eq!((report.objects, merged), (0, 0));
21001×
        assert_eq!(packs(dir.path()), before);
2101
        // One new object: a repack replaces the pack and its dictionary.
21021×
        let extra = again.put(ObjectType::Chunk, b"one more\n").unwrap();
21031×
        again.repack_all(REPACK_ZSTD_LEVEL).unwrap();
21041×
        let files = packs(dir.path());
21051×
        assert_eq!(files.len(), 3, "one pack, index and dictionary: {files:?}");
21061×
        assert_ne!(files, before);
21071×
        assert_eq!(&again.get(&refs[7]).unwrap().1, &bodies[7]);
21081×
        assert_eq!(again.get(&extra).unwrap().1, b"one more\n");
21091×
    }
2110
2111
    #[test]
21121×
    fn nocompress_chunks_pack_raw() {
21131×
        let dir = tempfile::tempdir().unwrap();
21141×
        let s = Store::open(dir.path()).unwrap();
21151×
        let body = vec![b'a'; 10_000];
21161×
        let raw = s.put_chunk_nocompress(&body).unwrap();
21171×
        let compressible = s.put(ObjectType::Chunk, &vec![b'b'; 10_000]).unwrap();
21181×
        s.pack_loose().unwrap();
21191×
        let packs = s.packs.borrow();
21201×
        let (i, e) = s.find_packed(&raw).unwrap();
21211×
        assert_eq!((e.flags, e.stored_len as usize), (0, 10_000));
21221×
        let (_, c) = s.find_packed(&compressible).unwrap();
21231×
        assert_eq!(c.flags, FLAG_ZSTD);
21241×
        assert!(c.stored_len < 10_000);
21251×
        assert_eq!(packs[i].fetch(s.backend.as_ref(), &raw, e).unwrap().1, body);
21261×
    }
2127
2128
    #[test]
21291×
    fn many_packs_merge_into_one() {
21301×
        let dir = tempfile::tempdir().unwrap();
21312×
        each_store(dir.path(), |which, s| {
21322×
            let mut refs = Vec::new();
213336×
            for i in 0..(MAX_PACKS + 2) {
213436×
                refs.push(
213536×
                    s.put(ObjectType::Chunk, format!("pack {i}").as_bytes())
213636×
                        .unwrap(),
213736×
                );
213836×
                s.pack_loose().unwrap();
213936×
            }
21402×
            assert_eq!(s.pack_count(), MAX_PACKS + 2, "{which}");
21412×
            assert_eq!(s.merge_packs().unwrap(), MAX_PACKS + 2);
21422×
            assert_eq!(s.pack_count(), 1);
21432×
            assert_eq!(s.merge_packs().unwrap(), 0);
214436×
            for (i, r) in refs.iter().enumerate() {
214536×
                assert_eq!(s.get(r).unwrap().1, format!("pack {i}").as_bytes());
2146
            }
21472×
            let again = Store::open_on(s.backend().clone(), s.dir()).unwrap();
21482×
            assert_eq!(again.pack_count(), 1, "{which}: reopened");
21492×
        });
21501×
    }
2151
2152
    #[test]
21531×
    fn prune_removes_unreachable_loose_and_packed_objects() {
2154
        use std::collections::HashSet;
21551×
        let dir = tempfile::tempdir().unwrap();
21562×
        each_store(dir.path(), |which, s| {
21572×
            let keep_a = s.put(ObjectType::Chunk, b"keep a").unwrap();
21582×
            let drop_a = s.put(ObjectType::Chunk, b"drop a").unwrap();
21592×
            s.pack_loose().unwrap();
21602×
            let keep_b = s.put(ObjectType::Chunk, b"keep b").unwrap();
21612×
            let drop_b = s.put(ObjectType::Chunk, b"drop b").unwrap();
21622×
            let sk = crate::identity::SecretKey {
21632×
                ed_seed: [1; 32],
21642×
                ml_seed: [2; 32],
21652×
            };
21662×
            let id = sk.identity("a", "b", 0);
21672×
            let key = s.put_identity(&id).unwrap();
21682×
            s.put_signature(
21692×
                &drop_b,
21702×
                &sk.sign(crate::identity::Domain::Change, &key, &drop_b),
2171
            )
21722×
            .unwrap();
21732×
            s.index_seal(&drop_a, &ObjectRef([3; 32])).unwrap();
21742×
            let keep: HashSet<ObjectRef> = [keep_a, keep_b, key].into_iter().collect();
2175
            // Everything is too young under a grace period.
21762×
            let r = s
21772×
                .prune(&keep, std::time::Duration::from_secs(3600))
21782×
                .unwrap();
21792×
            assert_eq!((r.objects, r.deferred), (0, 2), "{which}");
21802×
            let r = s.prune(&keep, std::time::Duration::ZERO).unwrap();
21812×
            assert_eq!(r.objects, 2, "{which}: {r:?}");
21822×
            assert!(s.has(&keep_a) && s.has(&keep_b) && s.has(&key));
21832×
            assert!(!s.has(&drop_a) && !s.has(&drop_b));
21842×
            assert!(s.signatures(&drop_b).unwrap().is_empty());
21852×
            assert!(s.seals_for(&drop_a).unwrap().is_empty());
21862×
            assert_eq!(s.pack_count(), 1);
21872×
            assert_eq!(s.get(&keep_a).unwrap().1, b"keep a");
21882×
            assert_eq!(s.list().unwrap().len(), 3);
21892×
        });
21901×
    }
2191
2192
    #[test]
21931×
    fn signatures_are_stored_beside_objects() {
2194
        use crate::identity::{Domain, SecretKey};
21951×
        let dir = tempfile::tempdir().unwrap();
21962×
        each_store(dir.path(), |which, s| {
21972×
            let r = s.put(ObjectType::Chunk, b"x").unwrap();
21982×
            assert!(s.signatures(&r).unwrap().is_empty());
21992×
            let sk = SecretKey {
22002×
                ed_seed: [1; 32],
22012×
                ml_seed: [2; 32],
22022×
            };
22032×
            let id = sk.identity("a", "b", 0);
22042×
            let key = s.put_identity(&id).unwrap();
22052×
            assert_eq!(s.get_identity(&key).unwrap(), id);
22062×
            let sig = sk.sign(Domain::Change, &key, &r);
22072×
            s.put_signature(&r, &sig).unwrap();
22082×
            s.put_signature(&r, &sig).unwrap();
22092×
            assert_eq!(s.signatures(&r).unwrap(), vec![sig.clone()], "{which}");
2210
            // A second signer is kept beside the first, sorted by key.
22112×
            let sk2 = SecretKey {
22122×
                ed_seed: [3; 32],
22132×
                ml_seed: [4; 32],
22142×
            };
22152×
            let key2 = s.put_identity(&sk2.identity("c", "d", 0)).unwrap();
22162×
            let sig2 = sk2.sign(Domain::Change, &key2, &r);
22172×
            s.put_signature(&r, &sig2).unwrap();
22182×
            let mut both = vec![sig.clone(), sig2.clone()];
22192×
            both.sort_by_key(|s| s.key);
22202×
            assert_eq!(s.signatures(&r).unwrap(), both);
2221
            // The object itself is untouched by its signatures.
22222×
            assert_eq!(s.get(&r).unwrap().1, b"x");
2223
            // Seal and receipt entries, and the states with receipts.
22242×
            s.index_seal(&r, &ObjectRef([5; 32])).unwrap();
22252×
            s.index_seal(&r, &ObjectRef([4; 32])).unwrap();
22262×
            s.index_seal(&r, &ObjectRef([5; 32])).unwrap();
22272×
            assert_eq!(
22282×
                s.seals_for(&r).unwrap(),
22292×
                vec![ObjectRef([4; 32]), ObjectRef([5; 32])]
2230
            );
22312×
            let st = ObjectRef([6; 32]);
22322×
            s.index_receipt(&st, &ObjectRef([7; 32])).unwrap();
22332×
            assert_eq!(s.receipts_for(&st).unwrap(), vec![ObjectRef([7; 32])]);
22342×
            assert_eq!(s.receipt_states().unwrap(), vec![st]);
22352×
        });
22361×
    }
2237
2238
    /// A store written before signatures, seals and receipts were one
2239
    /// entry each (`sigs/`, `seals/`, `receipts/` lists) still reads, and
2240
    /// what is written now sits beside what was there.
2241
    #[test]
22421×
    fn an_older_store_s_sidecar_lists_are_still_read() {
22431×
        if !super::files_on_disk() {
22440×
            return;
22451×
        }
2246
        use crate::identity::{Domain, SecretKey};
22471×
        let dir = tempfile::tempdir().unwrap();
22481×
        let s = Store::open(dir.path()).unwrap();
22491×
        let r = s.put(ObjectType::Chunk, b"old").unwrap();
22501×
        let h = r.to_hex();
22511×
        let sk = SecretKey {
22521×
            ed_seed: [1; 32],
22531×
            ml_seed: [2; 32],
22541×
        };
22551×
        let key = s.put_identity(&sk.identity("a", "b", 0)).unwrap();
22561×
        let sig = sk.sign(Domain::Change, &key, &r);
22571×
        let legacy = dir.path().join("sigs").join(&h[..2]);
22581×
        fs::create_dir_all(&legacy).unwrap();
22591×
        fs::write(
22601×
            legacy.join(&h[2..]),
22611×
            Signature::encode_list(std::slice::from_ref(&sig)),
2262
        )
22631×
        .unwrap();
22641×
        let seals = dir.path().join("seals").join(&h[..2]);
22651×
        fs::create_dir_all(&seals).unwrap();
22661×
        fs::write(
22671×
            seals.join(&h[2..]),
22681×
            format!("{}\n", ObjectRef([9; 32]).to_hex()),
2269
        )
22701×
        .unwrap();
22711×
        let receipts = dir.path().join("receipts").join(&h[..2]);
22721×
        fs::create_dir_all(&receipts).unwrap();
22731×
        fs::write(
22741×
            receipts.join(&h[2..]),
22751×
            format!("{}\n", ObjectRef([8; 32]).to_hex()),
2276
        )
22771×
        .unwrap();
2278
22791×
        assert_eq!(s.signatures(&r).unwrap(), vec![sig.clone()]);
22801×
        assert_eq!(s.seals_for(&r).unwrap(), vec![ObjectRef([9; 32])]);
22811×
        assert_eq!(s.receipts_for(&r).unwrap(), vec![ObjectRef([8; 32])]);
22821×
        assert_eq!(s.receipt_states().unwrap(), vec![r]);
2283
        // New entries join the old ones; the same signer is not doubled.
22841×
        s.put_signature(&r, &sig).unwrap();
22851×
        s.index_seal(&r, &ObjectRef([1; 32])).unwrap();
22861×
        assert_eq!(s.signatures(&r).unwrap().len(), 1);
22871×
        assert_eq!(
22881×
            s.seals_for(&r).unwrap(),
22891×
            vec![ObjectRef([1; 32]), ObjectRef([9; 32])]
2290
        );
2291
        // Pruning the object takes both layouts with it.
22921×
        let r2 = s.put(ObjectType::Chunk, b"kept").unwrap();
22931×
        let keep = [r2, key].into_iter().collect();
22941×
        s.prune(&keep, std::time::Duration::ZERO).unwrap();
22951×
        assert!(s.signatures(&r).unwrap().is_empty());
22961×
        assert!(s.seals_for(&r).unwrap().is_empty());
22971×
        assert!(s.receipt_states().unwrap().is_empty());
22981×
    }
2299
2300
    #[test]
23011×
    fn change_round_trip_through_store() {
23021×
        let dir = tempfile::tempdir().unwrap();
23032×
        each_store(dir.path(), |which, s| {
23042×
            let c = crate::objects::tests::sample_change();
23052×
            let r = s.put_change(&c).unwrap();
23062×
            assert_eq!(r, c.hash(), "{which}");
23072×
            assert_eq!(s.get_change(&r).unwrap(), c);
23082×
            assert_eq!(s.get_change_meta(&r).unwrap().message, c.message);
23092×
        });
23101×
    }
2311
23128057203×
    fn xorshift(state: &mut u64) -> u64 {
23138057203×
        *state ^= *state << 13;
23148057203×
        *state ^= *state >> 7;
23158057203×
        *state ^= *state << 17;
23168057203×
        *state
23178057203×
    }
2318
2319
    /// Random corruption of pack and index files never panics: reads fail
2320
    /// with an error or the store refuses to open, and untouched objects
2321
    /// stay readable when the index survives.
2322
    #[test]
23231×
    fn corrupted_packs_and_indexes_fail_cleanly() {
23241×
        if !super::files_on_disk() {
23250×
            return;
23261×
        }
23271×
        let dir = tempfile::tempdir().unwrap();
23281×
        let objects = dir.path().join("objects");
23291×
        let s = Store::open(&objects).unwrap();
23301×
        let mut refs = Vec::new();
23311×
        let mut seed = 0x9e3779b97f4a7c15u64;
233240×
        for i in 0..40u32 {
233340×
            let len = (xorshift(&mut seed) % 5000) as usize;
233440×
            let mut body = vec![0u8; len];
2335104686×
            for b in body.iter_mut() {
2336104686×
                *b = if i % 2 == 0 {
233756338×
                    (xorshift(&mut seed) & 0xff) as u8
2338
                } else {
233948348×
                    b'a' + (i % 26) as u8
2340
                };
2341
            }
234240×
            refs.push(s.put(ObjectType::Chunk, &body).unwrap());
2343
        }
23441×
        s.pack_loose().unwrap();
23451×
        let pack = fs::read_dir(&objects)
23461×
            .unwrap()
23472×
            .filter_map(|e| e.ok().map(|e| e.path()))
23482×
            .find(|p| p.extension().is_some_and(|x| x == "pack"))
23491×
            .unwrap();
23501×
        let idx = pack.with_extension("idx");
23511×
        let pristine_pack = fs::read(&pack).unwrap();
23521×
        let pristine_idx = fs::read(&idx).unwrap();
2353
2354300×
        for round in 0..300u32 {
2355300×
            let (path, pristine) = if round % 2 == 0 {
2356150×
                (&pack, &pristine_pack)
2357
            } else {
2358150×
                (&idx, &pristine_idx)
2359
            };
2360300×
            let mut data = pristine.clone();
2361300×
            match xorshift(&mut seed) % 4 {
236281×
                0 => {
236381×
                    let at = (xorshift(&mut seed) as usize) % data.len();
236481×
                    data[at] ^= 1 << (xorshift(&mut seed) % 8);
236581×
                }
236673×
                1 => {
236773×
                    let at = (xorshift(&mut seed) as usize) % data.len();
236873×
                    data[at] = (xorshift(&mut seed) & 0xff) as u8;
236973×
                }
237075×
                2 => {
237175×
                    let keep = (xorshift(&mut seed) as usize) % data.len();
237275×
                    data.truncate(keep);
237375×
                }
2374
                _ => {
237571×
                    let at = (xorshift(&mut seed) as usize) % data.len();
237671×
                    let n = ((xorshift(&mut seed) % 8) + 1) as usize;
2377324×
                    for k in 0..n.min(data.len() - at) {
2378324×
                        data[at + k] = 0xff;
2379324×
                    }
2380
                }
2381
            }
2382300×
            fs::write(path, &data).unwrap();
2383
            // Opening may refuse a broken index; a store that opens must
2384
            // either return each object intact or report it corrupt/missing.
2385300×
            if let Ok(s2) = Store::open(&objects) {
23867640×
                for r in &refs {
23877640×
                    match s2.get(r) {
23887452×
                        Ok((ty, body)) => assert_eq!(
23897452×
                            hash_object(ty, &body),
2390
                            *r,
2391
                            "a returned object always matches its reference"
2392
                        ),
2393
                        Err(StoreError::Corrupt(..))
2394
                        | Err(StoreError::NotFound(_))
2395188×
                        | Err(StoreError::Io(_)) => {}
23960×
                        Err(e) => panic!("unexpected error {e}"),
2397
                    }
2398
                }
2399109×
            }
2400300×
            fs::write(path, pristine).unwrap();
2401
        }
2402
        // Restored files read back in full.
24031×
        let s3 = Store::open(&objects).unwrap();
240440×
        for r in &refs {
240540×
            let (ty, body) = s3.get(r).unwrap();
240640×
            assert_eq!(hash_object(ty, &body), *r);
2407
        }
24081×
    }
2409
2410
    /// A small batch becomes loose objects; a large one becomes a single
2411
    /// pack. Objects are readable while pending, and batches nest.
2412
    #[test]
24131×
    fn batches_write_loose_or_one_pack() {
24141×
        let dir = tempfile::tempdir().unwrap();
24152×
        each_store(dir.path(), |which, s| {
24162×
            let outer = s.begin_batch().unwrap();
24172×
            let a = s.put(ObjectType::Chunk, b"small a").unwrap();
2418
            {
24192×
                let inner = s.begin_batch().unwrap();
24202×
                let b = s.put(ObjectType::Chunk, b"small b").unwrap();
24212×
                assert!(s.has(&b));
24222×
                assert_eq!(s.get(&b).unwrap().1, b"small b");
24232×
                inner.finish().unwrap();
24242×
                assert_eq!(
24252×
                    s.loose_count().unwrap(),
2426
                    0,
2427
                    "{which}: the inner batch defers to the outer one"
2428
                );
2429
            }
24302×
            assert_eq!(s.get(&a).unwrap().1, b"small a");
24312×
            assert_eq!(
24322×
                s.put(ObjectType::Chunk, b"small a").unwrap(),
2433
                a,
2434
                "idempotent while pending"
2435
            );
24362×
            outer.finish().unwrap();
24372×
            assert_eq!(s.loose_count().unwrap(), 2);
24382×
            assert_eq!(s.pack_count(), 0);
24392×
            assert_eq!(s.get(&a).unwrap().1, b"small a");
2440
2441
            // Large: spills into one pack, readable during and after.
24422×
            let big = s.begin_batch().unwrap();
24432×
            let mut refs = Vec::new();
24442×
            let mut seed = 7u64;
2445600×
            for i in 0..300u32 {
2446600×
                let mut body = vec![b'z'; 40_000];
2447600×
                if i % 3 == 0 {
24488000000×
                    for b in body.iter_mut() {
24498000000×
                        *b = (xorshift(&mut seed) & 0xff) as u8;
24508000000×
                    }
2451400×
                }
2452600×
                body[..4].copy_from_slice(&i.to_le_bytes());
2453600×
                let r = if i % 5 == 0 {
2454120×
                    s.put_chunk_nocompress(&body).unwrap()
2455
                } else {
2456480×
                    s.put(ObjectType::Chunk, &body).unwrap()
2457
                };
2458600×
                assert_eq!(
2459600×
                    s.get(&r).unwrap().1,
2460
                    body,
2461
                    "readable while pending (object {i})"
2462
                );
2463600×
                refs.push((r, body));
2464
            }
24652×
            assert!(s.has(&refs[0].0));
24662×
            big.finish().unwrap();
24672×
            assert_eq!(s.pack_count(), 1);
24682×
            assert_eq!(
24692×
                s.loose_count().unwrap(),
2470
                2,
2471
                "nothing from the big batch is loose"
2472
            );
24732×
            assert!(fs::read_dir(s.dir())
24742×
                .unwrap()
24752×
                .flatten()
24764×
                .all(|e| !e.file_name().to_string_lossy().contains("tmp-")));
24772×
            let s2 = Store::open_on(s.backend().clone(), s.dir()).unwrap();
2478600×
            for (r, body) in &refs {
2479600×
                assert_eq!(s2.get(r).unwrap().1, *body);
2480
            }
2481
            // A dropped batch still writes its objects.
24822×
            {
24832×
                let _b = s2.begin_batch().unwrap();
24842×
                s2.put(ObjectType::Chunk, b"dropped").unwrap();
24852×
            }
24862×
            assert_eq!(
24872×
                s2.get(&hash_object(ObjectType::Chunk, b"dropped"))
24882×
                    .unwrap()
2489
                    .1,
2490
                b"dropped"
2491
            );
24922×
        });
24931×
    }
2494
2495
    #[test]
24961×
    fn get_many_reads_loose_and_packed_objects_in_order() {
24971×
        let dir = tempfile::tempdir().unwrap();
24982×
        each_store(dir.path(), |which, s| {
24992×
            let packed: Vec<ObjectRef> = (0..20)
250040×
                .map(|i| {
250140×
                    s.put(ObjectType::Chunk, format!("packed {i}").as_bytes())
250240×
                        .unwrap()
250340×
                })
25042×
                .collect();
25052×
            s.pack_loose().unwrap();
25062×
            let loose: Vec<ObjectRef> = (0..20)
250740×
                .map(|i| {
250840×
                    s.put(ObjectType::Chunk, format!("loose {i}").as_bytes())
250940×
                        .unwrap()
251040×
                })
25112×
                .collect();
25122×
            let missing = hash_object(ObjectType::Chunk, b"never stored");
25132×
            let mut refs: Vec<ObjectRef> = Vec::new();
251440×
            for i in 0..20 {
251540×
                refs.push(loose[i]);
251640×
                refs.push(packed[i]);
251740×
            }
25182×
            refs.push(missing);
25192×
            let got = s.get_many(&refs);
25202×
            assert_eq!(got.len(), refs.len());
252180×
            for (i, r) in refs.iter().enumerate().take(40) {
252280×
                let want = if i % 2 == 0 {
252340×
                    format!("loose {}", i / 2)
2524
                } else {
252540×
                    format!("packed {}", i / 2)
2526
                };
252780×
                let (ty, body) = got[i].as_ref().unwrap();
252880×
                assert_eq!(
252980×
                    (*ty, body.as_slice()),
253080×
                    (ObjectType::Chunk, want.as_bytes()),
2531
                    "{which}: {r:?}"
2532
                );
2533
            }
25342×
            assert!(matches!(got[40], Err(StoreError::NotFound(_))));
25352×
        });
25361×
    }
2537
2538
    /// A packed object is read from its pack without first asking the
2539
    /// backend for a loose copy (oxygenhub#228: on a bucket each such ask
2540
    /// was a request that found nothing, every time).
2541
    #[test]
25421×
    fn packed_objects_cost_no_loose_lookups() {
2543
        use std::sync::atomic::{AtomicUsize, Ordering};
2544
        struct Counting(MemoryStore, AtomicUsize);
2545
        impl crate::strata::Store for Counting {
25460×
            fn describe(&self) -> String {
25470×
                "counting".into()
25480×
            }
254910×
            fn has(&self, k: &str) -> crate::strata::Result<bool> {
255010×
                self.0.has(k)
255110×
            }
255217×
            fn get(&self, k: &str) -> crate::strata::Result<Vec<u8>> {
255317×
                if k.starts_with("loose/") {
255411×
                    self.1.fetch_add(1, Ordering::Relaxed);
255511×
                }
255617×
                self.0.get(k)
255717×
            }
255820×
            fn get_range(
255920×
                &self,
256020×
                k: &str,
256120×
                r: std::ops::Range<u64>,
256220×
            ) -> crate::strata::Result<Vec<u8>> {
256320×
                self.0.get_range(k, r)
256420×
            }
256512×
            fn put_new(&self, k: &str, b: &[u8]) -> crate::strata::Result<()> {
256612×
                self.0.put_new(k, b)
256712×
            }
25686×
            fn list(&self, p: &str) -> crate::strata::Result<Vec<crate::strata::Entry>> {
25696×
                self.0.list(p)
25706×
            }
257110×
            fn delete(&self, k: &str) -> crate::strata::Result<()> {
257210×
                self.0.delete(k)
257310×
            }
25740×
            fn read_ref(
25750×
                &self,
25760×
                k: &str,
25770×
            ) -> crate::strata::Result<Option<(Vec<u8>, crate::strata::Version)>> {
25780×
                self.0.read_ref(k)
25790×
            }
25800×
            fn write_ref(
25810×
                &self,
25820×
                k: &str,
25830×
                b: &[u8],
25840×
                e: Option<&crate::strata::Version>,
25850×
            ) -> crate::strata::Result<crate::strata::Version> {
25860×
                self.0.write_ref(k, b, e)
25870×
            }
2588
        }
25891×
        let dir = tempfile::tempdir().unwrap();
25901×
        let backend = Arc::new(Counting(MemoryStore::new(), AtomicUsize::new(0)));
25911×
        let s = Store::open_on(backend.clone(), dir.path().join("scratch")).unwrap();
25921×
        let refs: Vec<ObjectRef> = (0..10)
259310×
            .map(|i| {
259410×
                s.put(ObjectType::Chunk, format!("p {i}").as_bytes())
259510×
                    .unwrap()
259610×
            })
25971×
            .collect();
25981×
        s.pack_loose().unwrap();
25991×
        let s = Store::open_on(backend.clone(), dir.path().join("scratch")).unwrap();
26001×
        backend.1.store(0, Ordering::Relaxed);
260110×
        for r in &refs {
260210×
            assert!(s.get(r).is_ok());
2603
        }
260410×
        assert!(s.get_many(&refs).iter().all(|g| g.is_ok()));
26051×
        assert_eq!(backend.1.load(Ordering::Relaxed), 0, "no loose lookups");
2606
        // An object no pack holds is still looked for loose.
26071×
        let missing = hash_object(ObjectType::Chunk, b"never stored");
26081×
        assert!(matches!(s.get(&missing), Err(StoreError::NotFound(_))));
26091×
        assert!(backend.1.load(Ordering::Relaxed) > 0);
26101×
    }
2611
}
2612
2613
#[cfg(test)]
2614
mod batch_tests {
2615
    use super::*;
2616
2617
    #[test]
26181×
    fn a_large_batch_packs_every_object_and_indexes_them_all() {
26191×
        let dir = tempfile::tempdir().unwrap();
26201×
        let store = Store::open(dir.path()).unwrap();
26211×
        let mut refs = Vec::new();
2622
        {
26231×
            let batch = store.begin_batch().unwrap();
26241000×
            for i in 0..1000u32 {
26251000×
                let body = format!("object {i} {}", "x".repeat(20_000 + i as usize)).into_bytes();
26261000×
                refs.push(store.put(ObjectType::Blob, &body).unwrap());
26271000×
            }
26281×
            batch.finish().unwrap();
2629
        }
26301×
        let again = Store::open(dir.path()).unwrap();
26311000×
        for r in &refs {
26321000×
            again
26331000×
                .get(r)
26341000×
                .unwrap_or_else(|e| panic!("{}: {e}", r.short(12)));
2635
        }
26361×
    }
2637
2638
    #[test]
26391×
    fn a_pack_a_stopped_batch_left_is_adopted_and_its_torn_end_cut() {
2640
        // oxygen#79: a batch never finished (a process killed part way)
2641
        // leaves its objects in an unindexed .tmp pack.
26421×
        let dir = tempfile::tempdir().unwrap();
26431×
        let mut refs = Vec::new();
2644
        {
26451×
            let store = Store::open(dir.path()).unwrap();
26461×
            let batch = store.begin_batch().unwrap();
2647300×
            for i in 0..300u32 {
2648300×
                let body = format!("object {i} {}", "y".repeat(20_000 + i as usize)).into_bytes();
2649300×
                refs.push(store.put(ObjectType::Blob, &body).unwrap());
2650300×
            }
26511×
            std::mem::forget(batch);
2652
        }
26531×
        let tmp: Vec<_> = fs::read_dir(dir.path())
26541×
            .unwrap()
26551×
            .flatten()
26562×
            .filter(|e| e.file_name().to_string_lossy().starts_with(".tmp-"))
26571×
            .map(|e| e.path())
26581×
            .collect();
26591×
        assert_eq!(tmp.len(), 1, "one unfinished pack");
2660
        // A torn last entry: a header and part of its body.
26611×
        let mut f = fs::OpenOptions::new().append(true).open(&tmp[0]).unwrap();
26621×
        f.write_all(&[0, 1, 0xff, 0, 0, 0, 9, 9, 0, 0]).unwrap();
26631×
        f.write_all(&[7; 40]).unwrap();
26641×
        drop(f);
26651×
        let store = Store::open(dir.path()).unwrap();
26661×
        assert!(store.get(&refs[0]).is_err(), "unreadable before");
26671×
        let young = store
26681×
            .adopt_unfinished_packs(std::time::Duration::from_secs(3600), &|_| {})
26691×
            .unwrap();
26701×
        assert_eq!(young, (0, 0), "a pack written to lately is left alone");
26711×
        let (packs, objects) = store
26722×
            .adopt_unfinished_packs(std::time::Duration::ZERO, &|_| {})
26731×
            .unwrap();
26741×
        assert_eq!((packs, objects), (1, 300));
2675300×
        for r in &refs {
2676300×
            store
2677300×
                .get(r)
2678300×
                .unwrap_or_else(|e| panic!("{}: {e}", r.short(12)));
2679
        }
26801×
        let again = Store::open(dir.path()).unwrap();
26811×
        assert!(again.get(&refs[299]).is_ok(), "indexed for good");
26821×
    }
2683
}
2684
2685
#[cfg(test)]
2686
mod size_tests {
2687
    use super::*;
2688
2689
    #[test]
26901×
    fn an_object_over_the_limit_is_refused_and_nothing_is_written() {
26911×
        let dir = tempfile::tempdir().unwrap();
26921×
        let store = Store::open(dir.path()).unwrap();
26931×
        let big = vec![b'x'; MAX_OBJECT_LEN as usize + 1];
2694
        {
26951×
            let batch = store.begin_batch().unwrap();
26961×
            store.put(ObjectType::Blob, b"small").unwrap();
26971×
            let e = store.put(ObjectType::Change, &big).unwrap_err();
26981×
            assert!(
26991×
                matches!(e, StoreError::TooLarge { kind: "change", .. }),
2700
                "{e}"
2701
            );
27021×
            assert!(e.to_string().contains("commit it in parts"), "{e}");
27031×
            batch.finish().unwrap();
2704
        }
2705
        // The store still opens and reads what was written.
27061×
        let again = Store::open(dir.path()).unwrap();
27071×
        again.get(&hash_object(ObjectType::Blob, b"small")).unwrap();
27081×
    }
2709
}
2710
2711
#[cfg(test)]
2712
mod text_tests {
2713
    use super::*;
2714
    use crate::objects::{AtomRef, Edit, CHANGE_VERSION};
2715
27162×
    fn large_change() -> Change {
27172×
        let mut c = crate::objects::tests::sample_change();
27182×
        assert_eq!(c.version, CHANGE_VERSION);
27192×
        c.edits[0] = Edit::Insert {
27202×
            node: crate::ids::NodeId([7; 16]),
27212×
            after: AtomRef::Start,
27222×
            before: AtomRef::End,
27232×
            lines: (0..2000)
27244000×
                .map(|i| format!("line {i}\n").into_bytes())
27252×
                .collect(),
2726
        };
27272×
        c
27282×
    }
2729
2730
    #[test]
27311×
    fn a_large_insert_is_stored_beside_its_change_and_read_back() {
27321×
        let dir = tempfile::tempdir().unwrap();
27331×
        let s = Store::open(dir.path()).unwrap();
27341×
        let c = large_change();
27351×
        let r = s.put_change(&c).unwrap();
27361×
        assert_eq!(r, c.hash());
27371×
        let (_, body) = s.get(&r).unwrap();
27381×
        assert!(
27391×
            body.len() < 1024,
2740
            "the change holds a reference, not {} bytes of lines",
27410×
            body.len()
2742
        );
27431×
        let (back, parts) = s.get_change_parts(&r).unwrap();
27441×
        assert_eq!(back, c);
2745
        // The text blob and its chunk.
27461×
        assert_eq!(parts.len(), 2);
27472×
        for p in &parts {
27482×
            assert!(s.has(p));
2749
        }
2750
        // Storing it again writes nothing new.
27511×
        let n = s.list().unwrap().len();
27521×
        s.put_change(&c).unwrap();
27531×
        assert_eq!(s.list().unwrap().len(), n);
27541×
    }
2755
2756
    #[test]
27571×
    fn a_change_whose_text_blob_is_missing_is_reported_not_misread() {
27581×
        if !super::files_on_disk() {
27590×
            return;
27601×
        }
27611×
        let dir = tempfile::tempdir().unwrap();
27621×
        let s = Store::open(dir.path()).unwrap();
27631×
        let c = large_change();
27641×
        let r = s.put_change(&c).unwrap();
27651×
        let (_, parts) = s.get_change_parts(&r).unwrap();
27661×
        fs::remove_file(s.path_for(&parts[0])).unwrap();
27671×
        let e = s.get_change(&r).unwrap_err();
27681×
        assert!(
27691×
            matches!(&e, StoreError::Corrupt(h, why) if *h == r && why.contains("missing")),
2770
            "{e}"
2771
        );
27721×
    }
2773
}
2774
2775
#[cfg(test)]
2776
mod put_many_tests {
2777
    use super::*;
2778
2779
    #[test]
27801×
    fn many_objects_in_a_batch_are_compressed_together_and_all_read_back() {
27811×
        let dir = tempfile::tempdir().unwrap();
27821×
        let s = Store::open(dir.path()).unwrap();
27831×
        let before = s.put(ObjectType::Chunk, b"already here").unwrap();
27841×
        let bodies: Vec<Vec<u8>> = (0..200u32)
2785200×
            .map(|i| format!("chunk {i} ").repeat(6000).into_bytes())
27861×
            .collect();
27871×
        let mut objects: Vec<(ObjectType, &[u8])> = bodies
27881×
            .iter()
2789200×
            .map(|b| (ObjectType::Chunk, b.as_slice()))
27901×
            .collect();
27911×
        objects.push((ObjectType::Chunk, bodies[0].as_slice()));
27921×
        objects.push((ObjectType::Chunk, b"already here"));
27931×
        let refs = {
27941×
            let batch = s.begin_batch().unwrap();
27951×
            let refs = s.put_many(&objects).unwrap();
27961×
            batch.finish().unwrap();
27971×
            refs
2798
        };
27991×
        assert_eq!(refs.len(), objects.len());
28001×
        assert_eq!(refs[200], refs[0], "a repeat is the same object");
28011×
        assert_eq!(refs[201], before);
28021×
        let again = Store::open(dir.path()).unwrap();
2803202×
        for (r, b) in refs.iter().zip(&objects) {
2804202×
            assert_eq!(again.get(r).unwrap().1, b.1);
2805
        }
28061×
    }
2807
}