OXYGENHUB
# 7e0e8e5ad882 oxygen / crates / oxy-core / src / store.rs
93.1 KiB · Rust · last changed Opening a bucket store sends no request per pack (refs #116) 18 days ago source blame history raw
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
72
fn silent() -> impl FnMut(&str) {
73
    |_| {}
74
}
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.
78
fn backend_err(e: BackendError) -> StoreError {
79
    match e {
80
        BackendError::Io(e) => StoreError::Io(e),
81
        other => StoreError::Io(io::Error::other(other.to_string())),
82
    }
83
}
84
85
/// `put_new` where an object already there is what we wanted.
86
fn put_new_or_existing(backend: &Backend, key: &str, bytes: &[u8]) -> Result<()> {
87
    match backend.put_new(key, bytes) {
88
        Ok(()) | Err(BackendError::Exists(_)) => Ok(()),
89
        Err(e) => Err(backend_err(e)),
90
    }
91
}
92
93
/// `get` as an option: `None` when there is no such key.
94
fn get_optional(backend: &Backend, key: &str) -> Result<Option<Vec<u8>>> {
95
    match backend.get(key) {
96
        Ok(b) => Ok(Some(b)),
97
        Err(BackendError::NotFound(_)) => Ok(None),
98
        Err(e) => Err(backend_err(e)),
99
    }
100
}
101
102
fn delete(backend: &Backend, key: &str) -> Result<()> {
103
    backend.delete(key).map_err(backend_err)
104
}
105
106
/// `loose/<hh>/<rest>`.
107
fn loose_key(r: &ObjectRef) -> String {
108
    let h = r.to_hex();
109
    format!("loose/{}/{}", &h[..2], &h[2..])
110
}
111
112
/// The object a loose key names, if it is one.
113
fn loose_ref(key: &str) -> Option<ObjectRef> {
114
    let rest = key.strip_prefix("loose/")?;
115
    let (hh, tail) = rest.split_once('/')?;
116
    if hh.len() != 2 || tail.contains('/') {
117
        return None;
118
    }
119
    ObjectRef::from_hex(&format!("{hh}{tail}"))
120
}
121
122
/// `<index>/<hh>/<rest>/` for the entries kept under an object.
123
fn fanout_prefix(index: &str, r: &ObjectRef) -> String {
124
    let h = r.to_hex();
125
    format!("{index}/{}/{}/", &h[..2], &h[2..])
126
}
127
128
/// `<index>/<hh>/<rest>`: the one-file-per-object layout of older stores.
129
fn legacy_key(index: &str, r: &ObjectRef) -> String {
130
    let h = r.to_hex();
131
    format!("{index}/{}/{}", &h[..2], &h[2..])
132
}
133
134
fn now_ms() -> u64 {
135
    std::time::SystemTime::now()
136
        .duration_since(std::time::UNIX_EPOCH)
137
        .map(|d| d.as_millis() as u64)
138
        .unwrap_or(0)
139
}
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 {
161
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
162
        f.debug_struct("PackIndex")
163
            .field("key", &self.key)
164
            .field("objects", &self.refs.len())
165
            .field("dict", &self.dict.is_some())
166
            .finish()
167
    }
168
}
169
170
/// Decode a stored pack entry and check it against its reference.
171
fn decode_entry(
172
    r: &ObjectRef,
173
    e: PackEntry,
174
    stored: Vec<u8>,
175
    dict: Option<&Arc<zstd::dict::DecoderDictionary<'static>>>,
176
) -> Result<Object> {
177
    let body = if e.flags & FLAG_DICT != 0 {
178
        let dict = dict.ok_or_else(|| {
179
            StoreError::Corrupt(*r, "packed with a dictionary the pack lacks".into())
180
        })?;
181
        zstd::bulk::Decompressor::with_prepared_dictionary(dict)
182
            .and_then(|mut d| d.decompress(&stored, e.raw_len as usize))
183
            .map_err(|err| StoreError::Corrupt(*r, format!("zstd: {err}")))?
184
    } else if e.flags & FLAG_ZSTD != 0 {
185
        zstd::bulk::decompress(&stored, e.raw_len as usize)
186
            .map_err(|err| StoreError::Corrupt(*r, format!("zstd: {err}")))?
187
    } else {
188
        stored
189
    };
190
    if body.len() != e.raw_len as usize || hash_object(e.ty, &body) != *r {
191
        return Err(StoreError::Corrupt(
192
            *r,
193
            "packed object does not match its reference".into(),
194
        ));
195
    }
196
    Ok((e.ty, body))
197
}
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).
202
    fn load(backend: &Backend, idx_key: &str, pack_key: &str, pack_len: u64) -> Result<PackIndex> {
203
        let data = backend.get(idx_key).map_err(backend_err)?;
204
        let bad = |m: &str| StoreError::Corrupt(ObjectRef::ZERO, format!("{idx_key}: {m}"));
205
        if data.len() < 12 || &data[..8] != IDX_MAGIC {
206
            return Err(bad("bad index header"));
207
        }
208
        let count = u32::from_le_bytes(data[8..12].try_into().unwrap()) as usize;
209
        if data.len() != 12 + count * IDX_ENTRY {
210
            return Err(bad("index length mismatch"));
211
        }
212
        let mut refs = Vec::with_capacity(count);
213
        let mut entries = Vec::with_capacity(count);
214
        for i in 0..count {
215
            let e = &data[12 + i * IDX_ENTRY..12 + (i + 1) * IDX_ENTRY];
216
            let r = ObjectRef::from_slice(&e[..32]).unwrap();
217
            if let Some(prev) = refs.last() {
218
                if r <= *prev {
219
                    return Err(bad("index not sorted"));
220
                }
221
            }
222
            refs.push(r);
223
            let entry = PackEntry {
224
                offset: u64::from_le_bytes(e[32..40].try_into().unwrap()),
225
                stored_len: u32::from_le_bytes(e[40..44].try_into().unwrap()),
226
                raw_len: u32::from_le_bytes(e[44..48].try_into().unwrap()),
227
                ty: ObjectType::ALL
228
                    .get(e[48] as usize)
229
                    .copied()
230
                    .ok_or_else(|| bad("unknown object type"))?,
231
                flags: e[49],
232
            };
233
            if entry.stored_len > MAX_OBJECT_LEN || entry.raw_len > MAX_OBJECT_LEN {
234
                return Err(StoreError::Corrupt(
235
                    r,
236
                    format!(
237
                        "{idx_key}: the index claims an object of {} bytes ({} stored), more than the {MAX_OBJECT_LEN} an object may be",
238
                        entry.raw_len, entry.stored_len
239
                    ),
240
                ));
241
            }
242
            if entry.offset.saturating_add(entry.stored_len as u64) > pack_len {
243
                return Err(bad("index entry points outside the pack"));
244
            }
245
            entries.push(entry);
246
        }
247
        let dict_key = format!("{}.dict", pack_key.trim_end_matches(".pack"));
248
        let dict = get_optional(backend, &dict_key)?
249
            .map(|d| Arc::new(zstd::dict::DecoderDictionary::copy(&d)));
250
        Ok(PackIndex {
251
            key: pack_key.to_string(),
252
            refs,
253
            entries,
254
            dict,
255
        })
256
    }
257
258
    fn find(&self, r: &ObjectRef) -> Option<PackEntry> {
259
        self.refs.binary_search(r).ok().map(|i| self.entries[i])
260
    }
261
262
    /// Read one entry from the pack: one ranged read, then decode.
263
    fn fetch(&self, backend: &Backend, r: &ObjectRef, e: PackEntry) -> Result<Object> {
264
        let stored = backend
265
            .get_range(&self.key, e.offset..e.offset + e.stored_len as u64)
266
            .map_err(|err| match err {
267
                BackendError::NotFound(_) => StoreError::NotFound(*r),
268
                other => backend_err(other),
269
            })?;
270
        decode_entry(r, e, stored, self.dict.as_ref())
271
    }
272
}
273
274
fn type_index(ty: ObjectType) -> u8 {
275
    ObjectType::ALL.iter().position(|t| *t == ty).unwrap() as u8
276
}
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 {
337
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
338
        f.debug_struct("Store")
339
            .field("backend", &self.backend.describe())
340
            .field("scratch", &self.scratch)
341
            .field("packs", &self.packs.borrow().len())
342
            .finish()
343
    }
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<'_> {
383
    pub fn finish(mut self) -> Result<()> {
384
        self.done = true;
385
        self.store.finish_batch()
386
    }
387
}
388
389
impl Drop for Batch<'_> {
390
    fn drop(&mut self) {
391
        if !self.done {
392
            let _ = self.store.finish_batch();
393
        }
394
    }
395
}
396
397
/// Append one stored object to a pending pack and index it.
398
fn write_pack_entry(
399
    pack: &mut PendingPack,
400
    r: &ObjectRef,
401
    ty: ObjectType,
402
    raw_len: usize,
403
    flags: u8,
404
    stored: &[u8],
405
    report: &mut PackReport,
406
) -> Result<PackEntry> {
407
    let header_len = 1 + 1 + 4 + 4 + 32;
408
    pack.writer.write_all(&[type_index(ty), flags])?;
409
    pack.writer
410
        .write_all(&(stored.len() as u32).to_le_bytes())?;
411
    pack.writer.write_all(&(raw_len as u32).to_le_bytes())?;
412
    pack.writer.write_all(&r.0)?;
413
    pack.writer.write_all(stored)?;
414
    let entry = PackEntry {
415
        offset: pack.offset + header_len,
416
        stored_len: stored.len() as u32,
417
        raw_len: raw_len as u32,
418
        ty,
419
        flags,
420
    };
421
    pack.offset += header_len + stored.len() as u64;
422
    pack.index.push((*r, entry));
423
    report.objects += 1;
424
    report.raw_bytes += raw_len as u64;
425
    report.packed_bytes += stored.len() as u64;
426
    Ok(entry)
427
}
428
429
/// A random name for a temporary file or a pack.
430
fn fresh_id() -> String {
431
    let mut nonce = [0u8; 8];
432
    rand::rngs::OsRng.fill_bytes(&mut nonce);
433
    hex::encode(nonce)
434
}
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.
440
fn memory_backend_for(dir: &Path) -> Option<Arc<Backend>> {
441
    use std::collections::HashMap;
442
    use std::sync::{Mutex, OnceLock};
443
    if std::env::var_os("OXY_STORE_BACKEND").as_deref() != Some(std::ffi::OsStr::new("memory")) {
444
        return None;
445
    }
446
    static STORES: OnceLock<Mutex<HashMap<PathBuf, Arc<MemoryStore>>>> = OnceLock::new();
447
    let key = dir.canonicalize().unwrap_or_else(|_| dir.to_path_buf());
448
    let mut map = STORES
449
        .get_or_init(Default::default)
450
        .lock()
451
        .unwrap_or_else(|p| p.into_inner());
452
    let store = map
453
        .entry(key)
454
        .or_insert_with(|| Arc::new(MemoryStore::new()))
455
        .clone();
456
    Some(store)
457
}
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.
469
    pub fn open(dir: impl Into<PathBuf>) -> io::Result<Store> {
470
        let dir = dir.into();
471
        fs::create_dir_all(dir.join("loose"))?;
472
        #[cfg(test)]
473
        if let Some(backend) = TEST_BACKEND.with(|t| t.borrow().clone()) {
474
            return Store::open_on(backend, dir);
475
        }
476
        if let Some(backend) = memory_backend_for(&dir) {
477
            return Store::open_on(backend, dir);
478
        }
479
        let backend: Arc<Backend> =
480
            Arc::new(LocalStore::open(&dir).map_err(|e| io::Error::other(e.to_string()))?);
481
        let mut store = Store::open_on(backend, dir)?;
482
        store.local = true;
483
        Ok(store)
484
    }
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.
488
    pub fn open_on(backend: Arc<Backend>, scratch: impl Into<PathBuf>) -> io::Result<Store> {
489
        let scratch = scratch.into();
490
        fs::create_dir_all(&scratch)?;
491
        let store = Store {
492
            backend,
493
            scratch,
494
            local: false,
495
            packs: RefCell::new(Vec::new()),
496
            pending: RefCell::new(None),
497
            batch_depth: std::cell::Cell::new(0),
498
            metas: RefCell::new(None),
499
        };
500
        store
501
            .reload_packs()
502
            .map_err(|e| io::Error::other(e.to_string()))?;
503
        Ok(store)
504
    }
505
506
    /// The storage under this store.
507
    pub fn backend(&self) -> &Arc<Backend> {
508
        &self.backend
509
    }
510
511
    /// Whether the store is the directory `dir()` names, where every key
512
    /// is a file.
513
    pub fn is_local(&self) -> bool {
514
        self.local
515
    }
516
517
    fn reload_packs(&self) -> Result<()> {
518
        let listed = self.backend.list("pack-").map_err(backend_err)?;
519
        let mut idx_keys: Vec<&str> = listed
520
            .iter()
521
            .map(|e| e.key.as_str())
522
            .filter(|k| k.ends_with(".idx"))
523
            .collect();
524
        idx_keys.sort();
525
        let mut packs = Vec::new();
526
        for idx in idx_keys {
527
            let pack = format!("{}.pack", idx.trim_end_matches(".idx"));
528
            if let Some(p) = listed.iter().find(|e| e.key == pack) {
529
                packs.push(PackIndex::load(self.backend.as_ref(), idx, &pack, p.size)?);
530
            }
531
        }
532
        *self.packs.borrow_mut() = packs;
533
        Ok(())
534
    }
535
536
    pub fn pack_count(&self) -> usize {
537
        self.packs.borrow().len()
538
    }
539
540
    fn find_packed(&self, r: &ObjectRef) -> Option<(usize, PackEntry)> {
541
        self.packs
542
            .borrow()
543
            .iter()
544
            .enumerate()
545
            .find_map(|(i, p)| p.find(r).map(|e| (i, e)))
546
    }
547
548
    /// Loose objects, with their keys and ages.
549
    fn loose_objects(&self) -> Result<Vec<(ObjectRef, strata::Entry)>> {
550
        let mut out: Vec<(ObjectRef, strata::Entry)> = self
551
            .backend
552
            .list("loose/")
553
            .map_err(backend_err)?
554
            .into_iter()
555
            .filter_map(|e| loose_ref(&e.key).map(|r| (r, e)))
556
            .collect();
557
        out.sort_by_key(|(r, _)| *r);
558
        Ok(out)
559
    }
560
561
    pub fn loose_count(&self) -> io::Result<usize> {
562
        self.loose_objects()
563
            .map(|l| l.len())
564
            .map_err(|e| io::Error::other(e.to_string()))
565
    }
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.
569
    pub fn pack_loose(&self) -> Result<PackReport> {
570
        let loose = self.loose_objects()?;
571
        self.write_pack(
572
            &loose.iter().map(|(r, _)| *r).collect::<Vec<_>>(),
573
            &loose.iter().map(|(_, e)| e.key.clone()).collect::<Vec<_>>(),
574
            &[],
575
            ZSTD_LEVEL,
576
            &mut silent(),
577
        )
578
    }
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.
583
    pub fn repack_all(&self, level: i32) -> Result<(PackReport, usize)> {
584
        self.repack_all_with(level, &mut silent())
585
    }
586
587
    /// [`Store::repack_all`], reporting progress.
588
    pub fn repack_all_with(&self, level: i32, progress: Progress) -> Result<(PackReport, usize)> {
589
        let loose = self.loose_objects()?;
590
        let (mut refs, keys): (Vec<ObjectRef>, Vec<String>) = {
591
            let packs = self.packs.borrow();
592
            (
593
                packs.iter().flat_map(|p| p.refs.iter().copied()).collect(),
594
                packs.iter().map(|p| p.key.clone()).collect(),
595
            )
596
        };
597
        if loose.is_empty() && keys.len() <= 1 && level == ZSTD_LEVEL {
598
            return Ok((PackReport::default(), 0));
599
        }
600
        refs.extend(loose.iter().map(|(r, _)| *r));
601
        let loose_keys: Vec<String> = loose.into_iter().map(|(_, e)| e.key).collect();
602
        let n = keys.len();
603
        let report = self.write_pack(&refs, &loose_keys, &keys, level, progress)?;
604
        Ok((report, n))
605
    }
606
607
    /// Fold every pack into one when there are more than `MAX_PACKS`
608
    /// (spec §13.4). Returns how many packs were merged.
609
    pub fn merge_packs(&self) -> Result<usize> {
610
        let (refs, keys): (Vec<ObjectRef>, Vec<String>) = {
611
            let packs = self.packs.borrow();
612
            if packs.len() <= MAX_PACKS {
613
                return Ok(0);
614
            }
615
            (
616
                packs.iter().flat_map(|p| p.refs.iter().copied()).collect(),
617
                packs.iter().map(|p| p.key.clone()).collect(),
618
            )
619
        };
620
        let n = keys.len();
621
        self.write_pack(&refs, &[], &keys, ZSTD_LEVEL, &mut silent())?;
622
        Ok(n)
623
    }
624
625
    /// Delete every object not in `keep` (spec §21). Loose objects and packs
626
    /// younger than `grace` are left alone, since a concurrent writer may not
627
    /// have recorded its references yet. Packs holding unreachable objects
628
    /// are rewritten without them. Signatures, seal and receipt entries of
629
    /// deleted objects go too.
630
    pub fn prune(
631
        &self,
632
        keep: &std::collections::HashSet<ObjectRef>,
633
        grace: std::time::Duration,
634
    ) -> Result<PruneReport> {
635
        self.prune_with(keep, grace, &mut silent())
636
    }
637
638
    /// [`Store::prune`], reporting progress.
639
    pub fn prune_with(
640
        &self,
641
        keep: &std::collections::HashSet<ObjectRef>,
642
        grace: std::time::Duration,
643
        progress: Progress,
644
    ) -> Result<PruneReport> {
645
        progress("pruning loose objects");
646
        let now = now_ms();
647
        let grace_ms = grace.as_millis() as u64;
648
        // An object whose age the store does not know is old enough.
649
        let old_enough =
650
            |modified: Option<u64>| modified.is_none_or(|t| now.saturating_sub(t) >= grace_ms);
651
        let mut report = PruneReport::default();
652
        // Loose objects.
653
        for (r, entry) in self.loose_objects()? {
654
            if keep.contains(&r) {
655
                continue;
656
            }
657
            if !old_enough(entry.modified_ms) {
658
                report.deferred += 1;
659
                continue;
660
            }
661
            report.bytes += entry.size;
662
            report.objects += 1;
663
            delete(self.backend.as_ref(), &entry.key)?;
664
            self.remove_sidecars(&r)?;
665
        }
666
        // Packs: rewrite those that hold unreachable objects.
667
        let mut keep_refs: Vec<ObjectRef> = Vec::new();
668
        let mut old_keys: Vec<String> = Vec::new();
669
        let mut rewrite = false;
670
        progress("pruning packs");
671
        for p in self.packs.borrow().iter() {
672
            let unreachable = p.refs.iter().filter(|r| !keep.contains(r)).count();
673
            if unreachable == 0 {
674
                continue;
675
            }
676
            let modified = self
677
                .backend
678
                .stat(&p.key)
679
                .map_err(backend_err)?
680
                .and_then(|e| e.modified_ms);
681
            if !old_enough(modified) {
682
                report.deferred += unreachable;
683
                continue;
684
            }
685
            rewrite = true;
686
            old_keys.push(p.key.clone());
687
            for (r, e) in p.refs.iter().zip(p.entries.iter()) {
688
                if keep.contains(r) {
689
                    keep_refs.push(*r);
690
                } else {
691
                    report.objects += 1;
692
                    report.bytes += e.stored_len as u64;
693
                }
694
            }
695
        }
696
        if rewrite {
697
            let dropped: Vec<ObjectRef> = {
698
                let packs = self.packs.borrow();
699
                packs
700
                    .iter()
701
                    .filter(|p| old_keys.contains(&p.key))
702
                    .flat_map(|p| p.refs.iter().copied())
703
                    .filter(|r| !keep.contains(r))
704
                    .collect()
705
            };
706
            self.write_pack(&keep_refs, &[], &old_keys, ZSTD_LEVEL, progress)?;
707
            for r in dropped {
708
                self.remove_sidecars(&r)?;
709
            }
710
        }
711
        Ok(report)
712
    }
713
714
    /// Everything kept beside an object: its signatures and the seal and
715
    /// receipt entries under it, in both layouts.
716
    fn remove_sidecars(&self, r: &ObjectRef) -> Result<()> {
717
        for index in ["sigs", "seals", "receipts"] {
718
            delete(self.backend.as_ref(), &legacy_key(index, r))?;
719
        }
720
        for index in ["sig", "seal", "receipt"] {
721
            for e in self
722
                .backend
723
                .list(&fanout_prefix(index, r))
724
                .map_err(backend_err)?
725
            {
726
                delete(self.backend.as_ref(), &e.key)?;
727
            }
728
        }
729
        Ok(())
730
    }
731
732
    /// Put a complete, synced temporary pack file and its index into the
733
    /// store; `index` must be sorted by ObjectRef. The pack goes first: an
734
    /// index never names a pack that is not there.
735
    fn finish_pack_files(
736
        &self,
737
        id: &str,
738
        tmp_pack: &Path,
739
        index: &[(ObjectRef, PackEntry)],
740
    ) -> Result<()> {
741
        let mut idx = Vec::with_capacity(12 + index.len() * IDX_ENTRY);
742
        idx.extend_from_slice(IDX_MAGIC);
743
        idx.extend_from_slice(&(index.len() as u32).to_le_bytes());
744
        for (r, e) in index {
745
            idx.extend_from_slice(&r.0);
746
            idx.extend_from_slice(&e.offset.to_le_bytes());
747
            idx.extend_from_slice(&e.stored_len.to_le_bytes());
748
            idx.extend_from_slice(&e.raw_len.to_le_bytes());
749
            idx.extend_from_slice(&[type_index(e.ty), e.flags]);
750
        }
751
        let put = self
752
            .backend
753
            .put_new_file(&format!("pack-{id}.pack"), tmp_pack)
754
            .map_err(backend_err);
755
        // A store on another disk copied it; the one in this directory
756
        // moved it and this finds nothing.
757
        let _ = fs::remove_file(tmp_pack);
758
        put?;
759
        self.backend
760
            .put_new(&format!("pack-{id}.idx"), &idx)
761
            .map_err(backend_err)?;
762
        self.reload_packs()
763
    }
764
765
    /// Remove a pack with its index and dictionary.
766
    fn remove_pack(&self, pack_key: &str) -> Result<()> {
767
        let stem = pack_key.trim_end_matches(".pack");
768
        delete(self.backend.as_ref(), pack_key)?;
769
        delete(self.backend.as_ref(), &format!("{stem}.idx"))?;
770
        delete(self.backend.as_ref(), &format!("{stem}.dict"))
771
    }
772
773
    /// Write `refs` into a new pack, then delete `loose_keys` and `old_packs`.
774
    fn write_pack(
775
        &self,
776
        refs: &[ObjectRef],
777
        loose_keys: &[String],
778
        old_packs: &[String],
779
        level: i32,
780
        progress: Progress,
781
    ) -> Result<PackReport> {
782
        if refs.is_empty() {
783
            for p in old_packs {
784
                self.remove_pack(p)?;
785
            }
786
            self.reload_packs()?;
787
            return Ok(PackReport::default());
788
        }
789
        let mut refs: Vec<ObjectRef> = refs.to_vec();
790
        refs.sort();
791
        refs.dedup();
792
        let id = fresh_id();
793
        let tmp_pack = self.scratch.join(format!(".tmp-{id}.pack"));
794
        let mut report = PackReport::default();
795
        let mut index: Vec<(ObjectRef, PackEntry)> = Vec::with_capacity(refs.len());
796
        // A full repack of enough objects trains a dictionary for them.
797
        let dict = if level == REPACK_ZSTD_LEVEL && refs.len() >= DICT_MIN_OBJECTS {
798
            let _t = crate::timing::phase("pack: train a dictionary");
799
            progress(&format!("training a dictionary on {} objects", refs.len()));
800
            self.train_dictionary(&refs)
801
        } else {
802
            None
803
        };
804
        let encoder = dict
805
            .as_ref()
806
            .map(|d| zstd::dict::EncoderDictionary::copy(d, level));
807
        let nocompress = self.nocompress_set()?;
808
        {
809
            let mut f = io::BufWriter::new(fs::File::create(&tmp_pack)?);
810
            f.write_all(PACK_MAGIC)?;
811
            let mut offset = PACK_MAGIC.len() as u64;
812
            // Read a batch (the store is not shared between threads),
813
            // compress it on every core, write it in order.
814
            let mut batches = refs.chunks(1024);
815
            let mut compressed_batch = Vec::new();
816
            let total = refs.len();
817
            let mut reported = usize::MAX;
818
            loop {
819
                if compressed_batch.is_empty() {
820
                    let Some(batch) = batches.next() else { break };
821
                    // Every tenth, and the first batch.
822
                    let tenth = report.objects * 10 / total.max(1);
823
                    if tenth != reported {
824
                        reported = tenth;
825
                        progress(&format!("packing {} of {total} objects", report.objects));
826
                    }
827
                    let mut read: Vec<(ObjectRef, ObjectType, Vec<u8>, bool)> =
828
                        Vec::with_capacity(batch.len());
829
                    for r in batch {
830
                        let (ty, body) = self.get(r)?;
831
                        read.push((*r, ty, body, nocompress.contains(r)));
832
                    }
833
                    use rayon::prelude::*;
834
                    compressed_batch = read
835
                        .into_par_iter()
836
                        .map_init(
837
                            || {
838
                                encoder.as_ref().and_then(|e| {
839
                                    zstd::bulk::Compressor::with_prepared_dictionary(e).ok()
840
                                })
841
                            },
842
                            |with_dict, (r, ty, body, raw)| {
843
                                let packed = if raw {
844
                                    None
845
                                } else {
846
                                    // Small objects with the dictionary;
847
                                    // a large one has context enough.
848
                                    let dicted = with_dict
849
                                        .as_mut()
850
                                        .filter(|_| body.len() <= DICT_OBJECT_MAX)
851
                                        .and_then(|c| c.compress(&body).ok())
852
                                        .map(|c| (FLAG_ZSTD | FLAG_DICT, c));
853
                                    dicted
854
                                        .or_else(|| {
855
                                            zstd::bulk::compress(&body, level)
856
                                                .ok()
857
                                                .map(|c| (FLAG_ZSTD, c))
858
                                        })
859
                                        .filter(|(_, c)| c.len() < body.len())
860
                                };
861
                                (r, ty, body, packed)
862
                            },
863
                        )
864
                        .collect::<Vec<_>>();
865
                    compressed_batch.reverse();
866
                }
867
                let Some((r, ty, body, packed)) = compressed_batch.pop() else {
868
                    continue;
869
                };
870
                let r = &r;
871
                let (flags, stored) = match packed {
872
                    Some(c) => c,
873
                    None => (0, body.clone()),
874
                };
875
                let header_len = 1 + 1 + 4 + 4 + 32;
876
                f.write_all(&[type_index(ty), flags])?;
877
                f.write_all(&(stored.len() as u32).to_le_bytes())?;
878
                f.write_all(&(body.len() as u32).to_le_bytes())?;
879
                f.write_all(&r.0)?;
880
                f.write_all(&stored)?;
881
                index.push((
882
                    *r,
883
                    PackEntry {
884
                        offset: offset + header_len,
885
                        stored_len: stored.len() as u32,
886
                        raw_len: body.len() as u32,
887
                        ty,
888
                        flags,
889
                    },
890
                ));
891
                offset += header_len + stored.len() as u64;
892
                report.objects += 1;
893
                report.raw_bytes += body.len() as u64;
894
                report.packed_bytes += stored.len() as u64;
895
            }
896
            f.flush()?;
897
            f.get_ref().sync_all()?;
898
        }
899
        // The dictionary before the pack that needs it becomes visible.
900
        if let Some(d) = &dict {
901
            self.backend
902
                .put_new(&format!("pack-{id}.dict"), d)
903
                .map_err(backend_err)?;
904
        }
905
        progress(&format!("writing the index of {} objects", index.len()));
906
        self.finish_pack_files(&id, &tmp_pack, &index)?;
907
        for p in old_packs {
908
            self.remove_pack(p)?;
909
        }
910
        self.reload_packs()?;
911
        for key in loose_keys {
912
            delete(self.backend.as_ref(), key)?;
913
        }
914
        Ok(report)
915
    }
916
917
    /// A dictionary for these objects, trained on a sample of them (at
918
    /// most 10 000, 32 MB); `None` when training fails.
919
    fn train_dictionary(&self, refs: &[ObjectRef]) -> Option<Vec<u8>> {
920
        let step = (refs.len() / 10_000).max(1);
921
        let mut samples: Vec<Vec<u8>> = Vec::new();
922
        let mut total = 0usize;
923
        for r in refs.iter().step_by(step) {
924
            if total > 32 << 20 {
925
                break;
926
            }
927
            let Ok((ty, body)) = self.get(r) else {
928
                continue;
929
            };
930
            if matches!(
931
                ty,
932
                ObjectType::Chunk | ObjectType::Blob | ObjectType::Change
933
            ) && !body.is_empty()
934
                && body.len() < 1 << 20
935
            {
936
                total += body.len();
937
                samples.push(body);
938
            }
939
        }
940
        zstd::dict::from_samples(&samples, DICT_BYTES).ok()
941
    }
942
943
    /// The scratch directory: for a store in a directory, that directory.
944
    pub fn dir(&self) -> &Path {
945
        &self.scratch
946
    }
947
948
    /// Where a loose object is kept, for a store in a directory (tests
949
    /// tamper with and remove objects there).
950
    #[cfg(test)]
951
    pub(crate) fn path_for(&self, r: &ObjectRef) -> PathBuf {
952
        self.scratch.join(loose_key(r))
953
    }
954
955
    /// Remove a loose object, whatever the store is on (tests make an
956
    /// object go missing).
957
    #[cfg(test)]
958
    pub(crate) fn remove_loose_for_test(&self, r: &ObjectRef) -> Result<()> {
959
        delete(self.backend.as_ref(), &loose_key(r))
960
    }
961
962
    pub fn has(&self, r: &ObjectRef) -> bool {
963
        self.backend.has(&loose_key(r)).unwrap_or(false)
964
            || self.find_packed(r).is_some()
965
            || self
966
                .pending
967
                .borrow()
968
                .as_ref()
969
                .is_some_and(|p| p.lookup.contains_key(r))
970
    }
971
972
    /// Start a write batch: until the matching `finish`, `put` collects
973
    /// objects and writes them out together, with one `fsync` for a large
974
    /// batch (a pack) instead of one per object. Batches nest; only the
975
    /// outermost does the work. Objects are readable while pending.
976
    pub fn begin_batch(&self) -> Result<Batch<'_>> {
977
        let depth = self.batch_depth.get();
978
        if depth == 0 {
979
            *self.pending.borrow_mut() = Some(Pending {
980
                buffered: vec![],
981
                buffered_bytes: 0,
982
                lookup: Default::default(),
983
                pack: None,
984
                report: PackReport::default(),
985
            });
986
        }
987
        self.batch_depth.set(depth + 1);
988
        Ok(Batch {
989
            store: self,
990
            done: false,
991
        })
992
    }
993
994
    fn finish_batch(&self) -> Result<()> {
995
        let depth = self.batch_depth.get();
996
        if depth == 0 {
997
            return Ok(());
998
        }
999
        self.batch_depth.set(depth - 1);
1000
        if depth > 1 {
1001
            return Ok(());
1002
        }
1003
        let Some(pending) = self.pending.borrow_mut().take() else {
1004
            return Ok(());
1005
        };
1006
        self.write_pending(pending)
1007
    }
1008
1009
    /// Make every object written so far durable, even inside an open
1010
    /// batch, which stays open. Call before a reference that names them
1011
    /// is moved (spec §13.6): the operation head, a bookmark pointer.
1012
    pub fn flush(&self) -> Result<()> {
1013
        let Some(pending) = self.pending.borrow_mut().take() else {
1014
            return Ok(());
1015
        };
1016
        self.write_pending(pending)?;
1017
        *self.pending.borrow_mut() = Some(Pending {
1018
            buffered: vec![],
1019
            buffered_bytes: 0,
1020
            lookup: Default::default(),
1021
            pack: None,
1022
            report: PackReport::default(),
1023
        });
1024
        Ok(())
1025
    }
1026
1027
    fn write_pending(&self, pending: Pending) -> Result<()> {
1028
        match pending.pack {
1029
            None => {
1030
                let objects: Vec<(ObjectType, &[u8])> = pending
1031
                    .buffered
1032
                    .iter()
1033
                    .map(|(ty, body)| (*ty, body.as_slice()))
1034
                    .collect();
1035
                self.put_loose_many(&objects)
1036
            }
1037
            Some(mut pack) => {
1038
                pack.writer.flush()?;
1039
                pack.writer.get_ref().sync_all()?;
1040
                drop(pack.writer);
1041
                pack.index.sort_by_key(|(r, _)| *r);
1042
                self.finish_pack_files(&pack.id, &pack.tmp_path, &pack.index)?;
1043
                Ok(())
1044
            }
1045
        }
1046
    }
1047
1048
    /// Whether a batch is open and holds `r`.
1049
    fn pending_get(&self, r: &ObjectRef) -> Option<Result<Object>> {
1050
        let mut guard = self.pending.borrow_mut();
1051
        let pending = guard.as_mut()?;
1052
        let entry = *pending.lookup.get(r)?;
1053
        Some(match entry {
1054
            PendingEntry::Buffered(i) => {
1055
                let (ty, body) = &pending.buffered[i];
1056
                Ok((*ty, body.clone()))
1057
            }
1058
            PendingEntry::Packed(e) => {
1059
                let Some(pack) = pending.pack.as_mut() else {
1060
                    return Some(Err(StoreError::NotFound(*r)));
1061
                };
1062
                // Read from the pack file being written, on this disk.
1063
                (|| {
1064
                    pack.writer.flush()?;
1065
                    use std::io::{Read, Seek, SeekFrom};
1066
                    let mut f = fs::File::open(&pack.tmp_path)?;
1067
                    f.seek(SeekFrom::Start(e.offset))?;
1068
                    let mut stored = vec![0u8; e.stored_len as usize];
1069
                    f.read_exact(&mut stored)?;
1070
                    decode_entry(r, e, stored, None)
1071
                })()
1072
            }
1073
        })
1074
    }
1075
1076
    /// Append one object to the pending pack file.
1077
    fn pending_pack_write(
1078
        &self,
1079
        pack: &mut PendingPack,
1080
        r: &ObjectRef,
1081
        ty: ObjectType,
1082
        body: &[u8],
1083
        report: &mut PackReport,
1084
    ) -> Result<PackEntry> {
1085
        let (flags, stored) = if self.is_nocompress(r) {
1086
            (0, body.to_vec())
1087
        } else {
1088
            let compressed = zstd::bulk::compress(body, ZSTD_LEVEL)
1089
                .map_err(|e| StoreError::Corrupt(*r, format!("zstd: {e}")))?;
1090
            if compressed.len() < body.len() {
1091
                (FLAG_ZSTD, compressed)
1092
            } else {
1093
                (0, body.to_vec())
1094
            }
1095
        };
1096
        write_pack_entry(pack, r, ty, body.len(), flags, &stored, report)
1097
    }
1098
1099
    fn put_pending(&self, r: ObjectRef, ty: ObjectType, body: &[u8]) -> Result<()> {
1100
        let mut guard = self.pending.borrow_mut();
1101
        let pending = guard.as_mut().expect("batch open");
1102
        if pending.lookup.contains_key(&r) {
1103
            return Ok(());
1104
        }
1105
        if pending.pack.is_none()
1106
            && pending.buffered.len() < SPILL_OBJECTS
1107
            && pending.buffered_bytes + body.len() <= SPILL_BYTES
1108
        {
1109
            pending.buffered.push((ty, body.to_vec()));
1110
            pending.buffered_bytes += body.len();
1111
            pending
1112
                .lookup
1113
                .insert(r, PendingEntry::Buffered(pending.buffered.len() - 1));
1114
            return Ok(());
1115
        }
1116
        self.ensure_pack(pending)?;
1117
        let mut report = std::mem::take(&mut pending.report);
1118
        let pack = pending.pack.as_mut().unwrap();
1119
        let e = self.pending_pack_write(pack, &r, ty, body, &mut report)?;
1120
        pending.report = report;
1121
        pending.lookup.insert(r, PendingEntry::Packed(e));
1122
        Ok(())
1123
    }
1124
1125
    /// Open the pending batch's pack, moving the objects buffered so far
1126
    /// into it (a spill).
1127
    fn ensure_pack(&self, pending: &mut Pending) -> Result<()> {
1128
        if pending.pack.is_none() {
1129
            // Spill: open the pack and move the buffered objects into it.
1130
            let id = fresh_id();
1131
            let tmp_path = self.scratch.join(format!(".tmp-{id}.pack"));
1132
            let mut writer = io::BufWriter::new(fs::File::create(&tmp_path)?);
1133
            writer.write_all(PACK_MAGIC)?;
1134
            let mut pack = PendingPack {
1135
                id,
1136
                tmp_path,
1137
                writer,
1138
                offset: PACK_MAGIC.len() as u64,
1139
                index: vec![],
1140
            };
1141
            let buffered = std::mem::take(&mut pending.buffered);
1142
            pending.buffered_bytes = 0;
1143
            let mut report = std::mem::take(&mut pending.report);
1144
            for (bty, bbody) in &buffered {
1145
                let br = hash_object(*bty, bbody);
1146
                let e = self.pending_pack_write(&mut pack, &br, *bty, bbody, &mut report)?;
1147
                pending.lookup.insert(br, PendingEntry::Packed(e));
1148
            }
1149
            pending.report = report;
1150
            pending.pack = Some(pack);
1151
        }
1152
        Ok(())
1153
    }
1154
1155
    /// Store many objects at once. Inside a large write batch they are
1156
    /// compressed on every core and appended in order; otherwise each is
1157
    /// stored as [`Store::put`] does.
1158
    pub fn put_many(&self, objects: &[(ObjectType, &[u8])]) -> Result<Vec<ObjectRef>> {
1159
        use rayon::prelude::*;
1160
        let total: usize = objects.iter().map(|(_, b)| b.len()).sum();
1161
        if self.pending.borrow().is_none() || total < SPILL_BYTES {
1162
            return objects.iter().map(|(ty, b)| self.put(*ty, b)).collect();
1163
        }
1164
        for (ty, b) in objects {
1165
            if b.len() > MAX_OBJECT_LEN as usize {
1166
                return Err(StoreError::TooLarge {
1167
                    kind: ty.name(),
1168
                    size: b.len(),
1169
                    limit: MAX_OBJECT_LEN,
1170
                });
1171
            }
1172
        }
1173
        let refs: Vec<ObjectRef> = objects
1174
            .par_iter()
1175
            .map(|(ty, b)| hash_object(*ty, b))
1176
            .collect();
1177
        // What is new: not stored already, not earlier in this call.
1178
        let mut seen: std::collections::HashSet<ObjectRef> = std::collections::HashSet::new();
1179
        let mut todo: Vec<usize> = Vec::new();
1180
        for (i, r) in refs.iter().enumerate() {
1181
            let known = self
1182
                .pending
1183
                .borrow()
1184
                .as_ref()
1185
                .is_some_and(|p| p.lookup.contains_key(r));
1186
            if seen.insert(*r)
1187
                && !known
1188
                && self.find_packed(r).is_none()
1189
                && !self.backend.has(&loose_key(r)).unwrap_or(false)
1190
            {
1191
                todo.push(i);
1192
            }
1193
        }
1194
        let raw = self.nocompress_set()?;
1195
        let packed: Vec<(u8, Vec<u8>)> = todo
1196
            .par_iter()
1197
            .map(|i| {
1198
                let body = objects[*i].1;
1199
                if raw.contains(&refs[*i]) {
1200
                    return (0, body.to_vec());
1201
                }
1202
                match zstd::bulk::compress(body, ZSTD_LEVEL) {
1203
                    Ok(c) if c.len() < body.len() => (FLAG_ZSTD, c),
1204
                    _ => (0, body.to_vec()),
1205
                }
1206
            })
1207
            .collect();
1208
        let mut guard = self.pending.borrow_mut();
1209
        let pending = guard.as_mut().expect("batch open");
1210
        self.ensure_pack(pending)?;
1211
        let mut report = std::mem::take(&mut pending.report);
1212
        let pack = pending.pack.as_mut().unwrap();
1213
        for (i, (flags, stored)) in todo.iter().zip(packed) {
1214
            let (ty, body) = objects[*i];
1215
            let e = write_pack_entry(pack, &refs[*i], ty, body.len(), flags, &stored, &mut report)?;
1216
            pending.lookup.insert(refs[*i], PendingEntry::Packed(e));
1217
        }
1218
        pending.report = report;
1219
        Ok(refs)
1220
    }
1221
1222
    /// Store `body` as an object of type `ty`; idempotent.
1223
    pub fn put(&self, ty: ObjectType, body: &[u8]) -> Result<ObjectRef> {
1224
        // Refuse what the store could not read back (spec §13.2), before
1225
        // anything is written: a pack must never hold an entry its index
1226
        // would reject.
1227
        if body.len() > MAX_OBJECT_LEN as usize {
1228
            return Err(StoreError::TooLarge {
1229
                kind: ty.name(),
1230
                size: body.len(),
1231
                limit: MAX_OBJECT_LEN,
1232
            });
1233
        }
1234
        let r = hash_object(ty, body);
1235
        if self.find_packed(&r).is_some() || self.backend.has(&loose_key(&r)).unwrap_or(false) {
1236
            return Ok(r);
1237
        }
1238
        if self.pending.borrow().is_some() {
1239
            self.put_pending(r, ty, body)?;
1240
            return Ok(r);
1241
        }
1242
        self.put_loose(ty, body)?;
1243
        Ok(r)
1244
    }
1245
1246
    /// Write one loose object, durable when this returns.
1247
    fn put_loose(&self, ty: ObjectType, body: &[u8]) -> Result<ObjectRef> {
1248
        let r = hash_object(ty, body);
1249
        let mut bytes = header(ty, body.len());
1250
        bytes.extend_from_slice(body);
1251
        crate::durable::step("write loose object")?;
1252
        put_new_or_existing(self.backend.as_ref(), &loose_key(&r), &bytes)?;
1253
        Ok(r)
1254
    }
1255
1256
    /// Write several loose objects with as few waits on the disk as the
1257
    /// guarantees allow (a directory store writes them all, then syncs
1258
    /// each directory once). Durable when this returns, as `put_loose`
1259
    /// one at a time would be (spec §13.6).
1260
    fn put_loose_many(&self, objects: &[(ObjectType, &[u8])]) -> Result<()> {
1261
        if objects.len() <= 1 {
1262
            for (ty, body) in objects {
1263
                self.put_loose(*ty, body)?;
1264
            }
1265
            return Ok(());
1266
        }
1267
        let mut seen = std::collections::HashSet::new();
1268
        let items: Vec<(String, Vec<u8>)> = objects
1269
            .iter()
1270
            .filter_map(|(ty, body)| {
1271
                let r = hash_object(*ty, body);
1272
                seen.insert(r).then(|| {
1273
                    let mut bytes = header(*ty, body.len());
1274
                    bytes.extend_from_slice(body);
1275
                    (loose_key(&r), bytes)
1276
                })
1277
            })
1278
            .collect();
1279
        self.backend.put_many_new(&items).map_err(backend_err)
1280
    }
1281
1282
    /// Read and verify an object, loose, packed, or pending in a batch.
1283
    pub fn get(&self, r: &ObjectRef) -> Result<Object> {
1284
        match get_optional(self.backend.as_ref(), &loose_key(r))? {
1285
            Some(bytes) => parse_loose(&bytes, r),
1286
            None => match self.find_packed(r) {
1287
                Some((i, e)) => self.packs.borrow()[i].fetch(self.backend.as_ref(), r, e),
1288
                None => self.pending_get(r).unwrap_or(Err(StoreError::NotFound(*r))),
1289
            },
1290
        }
1291
    }
1292
1293
    /// Many objects at once: loose ones are read and verified in parallel,
1294
    /// then packed ones in parallel, then what is pending in a batch.
1295
    /// Results are in the order of `refs`.
1296
    pub fn get_many(&self, refs: &[ObjectRef]) -> Vec<Result<Object>> {
1297
        use rayon::prelude::*;
1298
        let backend = self.backend.as_ref();
1299
        let mut found: Vec<Option<Result<Object>>> = refs
1300
            .par_iter()
1301
            .map(|r| match get_optional(backend, &loose_key(r)) {
1302
                Ok(Some(bytes)) => Some(parse_loose(&bytes, r)),
1303
                Ok(None) => None,
1304
                Err(e) => Some(Err(e)),
1305
            })
1306
            .collect();
1307
        // Packed ones: look them up in order, then read them in parallel.
1308
        let packs = self.packs.borrow();
1309
        let packed: Vec<(usize, usize, PackEntry)> = found
1310
            .iter()
1311
            .enumerate()
1312
            .filter(|(_, f)| f.is_none())
1313
            .filter_map(|(i, _)| {
1314
                packs
1315
                    .iter()
1316
                    .enumerate()
1317
                    .find_map(|(p, pack)| pack.find(&refs[i]).map(|e| (i, p, e)))
1318
            })
1319
            .collect();
1320
        let indexes: &[PackIndex] = &packs;
1321
        let read: Vec<(usize, Result<Object>)> = packed
1322
            .par_iter()
1323
            .map(|(i, p, e)| (*i, indexes[*p].fetch(backend, &refs[*i], *e)))
1324
            .collect();
1325
        drop(packs);
1326
        for (i, r) in read {
1327
            found[i] = Some(r);
1328
        }
1329
        found
1330
            .into_iter()
1331
            .zip(refs)
1332
            .map(|(f, r)| match f {
1333
                Some(f) => f,
1334
                None => self.get(r),
1335
            })
1336
            .collect()
1337
    }
1338
1339
    pub fn get_typed(&self, r: &ObjectRef, want: ObjectType) -> Result<Vec<u8>> {
1340
        let (ty, body) = self.get(r)?;
1341
        if ty != want {
1342
            return Err(StoreError::WrongType(*r, ty.name(), want.name()));
1343
        }
1344
        Ok(body)
1345
    }
1346
1347
    /// Store a change and the text blobs (and their chunks) its large
1348
    /// inserts are stored as (spec §6.1).
1349
    pub fn put_change(&self, c: &Change) -> Result<ObjectRef> {
1350
        let bytes = c.encode_streaming(|batch| self.put_texts(&batch))?;
1351
        self.put(ObjectType::Change, &bytes)
1352
    }
1353
1354
    /// Store text blobs and their chunks: every chunk at once (compressed
1355
    /// on every core in a large batch), then the blobs.
1356
    pub fn put_texts(&self, texts: &[crate::objects::StoredText]) -> Result<()> {
1357
        let mut chunks: Vec<(ObjectType, &[u8])> = Vec::new();
1358
        for t in texts {
1359
            if !self.has(&t.blob.hash()) {
1360
                chunks.extend(
1361
                    crate::chunk::chunks(&t.content)
1362
                        .into_iter()
1363
                        .map(|c| (ObjectType::Chunk, c)),
1364
                );
1365
            }
1366
        }
1367
        self.put_many(&chunks)?;
1368
        for t in texts {
1369
            self.put(ObjectType::Blob, &t.blob.encode())?;
1370
        }
1371
        Ok(())
1372
    }
1373
1374
    pub fn get_change(&self, r: &ObjectRef) -> Result<Change> {
1375
        Ok(self.get_change_parts(r)?.0)
1376
    }
1377
1378
    /// A change and the objects its stored inserts are made of: each text
1379
    /// blob and its chunks, which travel and are kept with it.
1380
    pub fn get_change_parts(&self, r: &ObjectRef) -> Result<(Change, Vec<ObjectRef>)> {
1381
        let bytes = self.get_typed(r, ObjectType::Change)?;
1382
        let mut parts = Vec::new();
1383
        let mut failed = None;
1384
        let decoded = Change::decode_with(&bytes, &mut |text, count, form| match self
1385
            .read_text(text)
1386
        {
1387
            Ok((blob, content)) => {
1388
                parts.push(*text);
1389
                parts.extend(blob.chunks);
1390
                crate::objects::text_lines(&content, count, form)
1391
            }
1392
            Err(e) => {
1393
                let why = e.to_string();
1394
                failed = Some(e);
1395
                Err(ObjectError::Schema(format!(
1396
                    "text blob {}: {why}",
1397
                    text.short(12)
1398
                )))
1399
            }
1400
        });
1401
        match (decoded, failed) {
1402
            (Ok(c), _) => Ok((c, parts)),
1403
            (Err(_), Some(StoreError::NotFound(t))) => Err(StoreError::Corrupt(
1404
                *r,
1405
                format!("its text blob {} is missing", t.short(12)),
1406
            )),
1407
            (Err(e), _) => Err(StoreError::Corrupt(*r, e.to_string())),
1408
        }
1409
    }
1410
1411
    /// A change with every insert left without its lines (empty
1412
    /// placeholders), and where each one's lines are, in order: for
1413
    /// history-only states, which read text only for the graphs they build
1414
    /// (see [`Change::decode_skeleton`]).
1415
    pub fn get_change_lazy(&self, r: &ObjectRef) -> Result<(Change, Vec<crate::objects::TextRef>)> {
1416
        let bytes = self.get_typed(r, ObjectType::Change)?;
1417
        Change::decode_skeleton(&bytes, *r).map_err(|e| StoreError::Corrupt(*r, e.to_string()))
1418
    }
1419
1420
    /// A change's metadata (id, dependencies, author, clock, message, kind)
1421
    /// without its contents: read from its skeleton, so neither its text
1422
    /// blobs nor its inline lines are read (spec §11).
1423
    ///
1424
    /// Served from the `metas` cache when this store read it before (a
1425
    /// change object can be as large as the tree it adds), as long as the
1426
    /// store still has the change.
1427
    pub fn get_change_meta(&self, r: &ObjectRef) -> Result<crate::state::ChangeMeta> {
1428
        if self.metas.borrow().is_none() {
1429
            *self.metas.borrow_mut() = Some(self.read_metas_cache());
1430
        }
1431
        let cached = self.metas.borrow().as_ref().and_then(|m| m.get(r).cloned());
1432
        if let Some(meta) = cached {
1433
            if self.has(r) {
1434
                #[cfg(test)]
1435
                assert_eq!(
1436
                    meta,
1437
                    Change::decode_meta(&self.get_typed(r, ObjectType::Change)?).unwrap(),
1438
                    "the metadata cache holds a change's metadata"
1439
                );
1440
                return Ok(meta);
1441
            }
1442
        }
1443
        let bytes = self.get_typed(r, ObjectType::Change)?;
1444
        let meta =
1445
            Change::decode_meta(&bytes).map_err(|e| StoreError::Corrupt(*r, e.to_string()))?;
1446
        self.remember_meta(r, &meta);
1447
        Ok(meta)
1448
    }
1449
1450
    /// The metadata cache: records of a change hash, its metadata in the
1451
    /// flat layout, and the first 8 bytes of BLAKE3 over both, each
1452
    /// appended in one write. Derived only: a record that does not check
1453
    /// ends the reading (a torn write), and the cache can be deleted. It
1454
    /// lives in the scratch directory, on this disk.
1455
    fn read_metas_cache(&self) -> crate::fasthash::FastMap<ObjectRef, crate::state::ChangeMeta> {
1456
        let mut out = crate::fasthash::FastMap::default();
1457
        let Ok(bytes) = fs::read(self.scratch.join("metas")) else {
1458
            return out;
1459
        };
1460
        let mut r = crate::flat::Reader::new(&bytes);
1461
        let mut good = 0;
1462
        while !r.at_end() {
1463
            let mut record = || -> std::result::Result<_, ObjectError> {
1464
                let h = r.object()?;
1465
                let body = r.bytes()?;
1466
                let check = r.fixed::<8>()?;
1467
                Ok((h, body, check))
1468
            };
1469
            let Ok((h, body, check)) = record() else {
1470
                break;
1471
            };
1472
            let mut sum = blake3::Hasher::new();
1473
            sum.update(&h.0).update(body);
1474
            if sum.finalize().as_bytes()[..8] != check {
1475
                break;
1476
            }
1477
            match crate::state::ChangeMeta::read_flat(&mut crate::flat::Reader::new(body)) {
1478
                Ok(m) => out.insert(h, m),
1479
                Err(_) => break,
1480
            };
1481
            good += 32 + 4 + body.len() + 8;
1482
        }
1483
        // Cut a bad tail, so records appended later are read.
1484
        if good < bytes.len() {
1485
            let _ = fs::OpenOptions::new()
1486
                .write(true)
1487
                .open(self.scratch.join("metas"))
1488
                .and_then(|f| f.set_len(good as u64));
1489
        }
1490
        out
1491
    }
1492
1493
    /// Add a change's metadata to the cache; quietly nothing when it
1494
    /// cannot be written (it is only a cache).
1495
    fn remember_meta(&self, r: &ObjectRef, meta: &crate::state::ChangeMeta) {
1496
        let mut body = crate::flat::Writer::default();
1497
        meta.write_flat(&mut body);
1498
        let mut w = crate::flat::Writer::default();
1499
        w.object(r);
1500
        w.bytes(&body.0);
1501
        let mut sum = blake3::Hasher::new();
1502
        sum.update(&r.0).update(&body.0);
1503
        w.fixed(&sum.finalize().as_bytes()[..8]);
1504
        let _ = fs::OpenOptions::new()
1505
            .create(true)
1506
            .append(true)
1507
            .open(self.scratch.join("metas"))
1508
            .and_then(|mut f| f.write_all(&w.0));
1509
        if let Some(m) = self.metas.borrow_mut().as_mut() {
1510
            m.insert(*r, meta.clone());
1511
        }
1512
    }
1513
1514
    fn read_text(&self, r: &ObjectRef) -> Result<(Blob, Vec<u8>)> {
1515
        let blob = Blob::decode(&self.get_typed(r, ObjectType::Blob)?)?;
1516
        let mut content = Vec::with_capacity(blob.length as usize);
1517
        for c in &blob.chunks {
1518
            content.extend(self.get_typed(c, ObjectType::Chunk)?);
1519
        }
1520
        Ok((blob, content))
1521
    }
1522
1523
    pub fn put_chunk(&self, bytes: &[u8]) -> Result<ObjectRef> {
1524
        self.put(ObjectType::Chunk, bytes)
1525
    }
1526
1527
    /// Store a chunk that packs will keep uncompressed (`nocompress`, spec §13.3).
1528
    pub fn put_chunk_nocompress(&self, bytes: &[u8]) -> Result<ObjectRef> {
1529
        let r = hash_object(ObjectType::Chunk, bytes);
1530
        put_new_or_existing(
1531
            self.backend.as_ref(),
1532
            &format!("nocompress/{}", r.to_hex()),
1533
            b"",
1534
        )?;
1535
        self.put(ObjectType::Chunk, bytes)
1536
    }
1537
1538
    fn is_nocompress(&self, r: &ObjectRef) -> bool {
1539
        self.backend
1540
            .has(&format!("nocompress/{}", r.to_hex()))
1541
            .unwrap_or(false)
1542
    }
1543
1544
    /// Every chunk marked to be kept raw.
1545
    fn nocompress_set(&self) -> Result<std::collections::HashSet<ObjectRef>> {
1546
        Ok(self
1547
            .backend
1548
            .list("nocompress/")
1549
            .map_err(backend_err)?
1550
            .into_iter()
1551
            .filter_map(|e| ObjectRef::from_hex(e.key.trim_start_matches("nocompress/")))
1552
            .collect())
1553
    }
1554
1555
    pub fn put_state(&self, st: &StateObject) -> Result<ObjectRef> {
1556
        self.put(ObjectType::State, &st.encode())
1557
    }
1558
1559
    pub fn get_state(&self, r: &ObjectRef) -> Result<StateObject> {
1560
        Ok(StateObject::decode(&self.get_typed(r, ObjectType::State)?)?)
1561
    }
1562
1563
    pub fn put_attest(&self, a: &Attest) -> Result<ObjectRef> {
1564
        self.put(ObjectType::Attest, &a.encode())
1565
    }
1566
1567
    pub fn get_attest(&self, r: &ObjectRef) -> Result<Attest> {
1568
        Ok(Attest::decode(&self.get_typed(r, ObjectType::Attest)?)?)
1569
    }
1570
1571
    pub fn put_receipt(&self, r: &Receipt) -> Result<ObjectRef> {
1572
        self.put(ObjectType::Receipt, &r.encode())
1573
    }
1574
1575
    pub fn get_receipt(&self, r: &ObjectRef) -> Result<Receipt> {
1576
        Ok(Receipt::decode(&self.get_typed(r, ObjectType::Receipt)?)?)
1577
    }
1578
1579
    /// Record `value` under `key` in `index`: one write-once entry, so two
1580
    /// writers of a shared store never lose each other's.
1581
    fn index_add(&self, index: &str, key: &ObjectRef, value: &ObjectRef) -> Result<()> {
1582
        let entry = format!("{}{}", fanout_prefix(index, key), value.to_hex());
1583
        put_new_or_existing(self.backend.as_ref(), &entry, b"")
1584
    }
1585
1586
    /// The values recorded under `key` in `index`, with those an older
1587
    /// store kept as one list in `legacy`, sorted.
1588
    fn index_get(&self, index: &str, legacy: &str, key: &ObjectRef) -> Result<Vec<ObjectRef>> {
1589
        let mut values: Vec<ObjectRef> = self
1590
            .backend
1591
            .list(&fanout_prefix(index, key))
1592
            .map_err(backend_err)?
1593
            .into_iter()
1594
            .filter_map(|e| e.key.rsplit('/').next().and_then(ObjectRef::from_hex))
1595
            .collect();
1596
        if let Some(bytes) = get_optional(self.backend.as_ref(), &legacy_key(legacy, key))? {
1597
            values.extend(
1598
                String::from_utf8_lossy(&bytes)
1599
                    .lines()
1600
                    .filter_map(ObjectRef::from_hex),
1601
            );
1602
        }
1603
        values.sort();
1604
        values.dedup();
1605
        Ok(values)
1606
    }
1607
1608
    /// Record that seal attestation `attest` covers `change`.
1609
    pub fn index_seal(&self, change: &ObjectRef, attest: &ObjectRef) -> Result<()> {
1610
        self.index_add("seal", change, attest)
1611
    }
1612
1613
    /// Seal attestations known to cover `change`.
1614
    pub fn seals_for(&self, change: &ObjectRef) -> Result<Vec<ObjectRef>> {
1615
        self.index_get("seal", "seals", change)
1616
    }
1617
1618
    /// Record receipt attestation `attest` for `state` (spec §14.6).
1619
    pub fn index_receipt(&self, state: &ObjectRef, attest: &ObjectRef) -> Result<()> {
1620
        self.index_add("receipt", state, attest)
1621
    }
1622
1623
    /// Receipt attestations recorded for `state`.
1624
    pub fn receipts_for(&self, state: &ObjectRef) -> Result<Vec<ObjectRef>> {
1625
        self.index_get("receipt", "receipts", state)
1626
    }
1627
1628
    /// Every state with a receipt, for garbage collection.
1629
    pub fn receipt_states(&self) -> Result<Vec<ObjectRef>> {
1630
        let mut out: Vec<ObjectRef> = Vec::new();
1631
        for e in self.backend.list("receipt/").map_err(backend_err)? {
1632
            // receipt/<hh>/<rest>/<attest>
1633
            let mut parts = e.key.splitn(4, '/');
1634
            let (Some(_), Some(hh), Some(rest), Some(_)) =
1635
                (parts.next(), parts.next(), parts.next(), parts.next())
1636
            else {
1637
                continue;
1638
            };
1639
            out.extend(ObjectRef::from_hex(&format!("{hh}{rest}")));
1640
        }
1641
        for e in self.backend.list("receipts/").map_err(backend_err)? {
1642
            let mut parts = e.key.splitn(3, '/');
1643
            let (Some(_), Some(hh), Some(rest)) = (parts.next(), parts.next(), parts.next()) else {
1644
                continue;
1645
            };
1646
            out.extend(ObjectRef::from_hex(&format!("{hh}{rest}")));
1647
        }
1648
        out.sort();
1649
        out.dedup();
1650
        Ok(out)
1651
    }
1652
1653
    pub fn put_identity(&self, id: &Identity) -> Result<ObjectRef> {
1654
        self.put(ObjectType::Identity, &id.encode())
1655
    }
1656
1657
    pub fn get_identity(&self, r: &ObjectRef) -> Result<Identity> {
1658
        Ok(Identity::decode(&self.get_typed(r, ObjectType::Identity)?)?)
1659
    }
1660
1661
    /// Signatures stored alongside object `r` (spec §5.4), one per signing
1662
    /// key, sorted by key. Empty if none.
1663
    pub fn signatures(&self, r: &ObjectRef) -> Result<Vec<Signature>> {
1664
        let mut sigs: Vec<Signature> = Vec::new();
1665
        for e in self
1666
            .backend
1667
            .list(&fanout_prefix("sig", r))
1668
            .map_err(backend_err)?
1669
        {
1670
            let bytes = self.backend.get(&e.key).map_err(backend_err)?;
1671
            sigs.extend(Signature::decode_list(&bytes)?);
1672
        }
1673
        // An older store's one list per object; a signer kept there and
1674
        // here counts once, as written here.
1675
        if let Some(bytes) = get_optional(self.backend.as_ref(), &legacy_key("sigs", r))? {
1676
            for sig in Signature::decode_list(&bytes)? {
1677
                if !sigs.iter().any(|s| s.key == sig.key) {
1678
                    sigs.push(sig);
1679
                }
1680
            }
1681
        }
1682
        sigs.sort_by_key(|s| s.key);
1683
        sigs.dedup_by_key(|s| s.key);
1684
        Ok(sigs)
1685
    }
1686
1687
    /// Attach a signature to object `r`, replacing any earlier one by the
1688
    /// same key: one entry per signer, so the latest by a key is the one
1689
    /// kept, and two signers never touch each other's.
1690
    pub fn put_signature(&self, r: &ObjectRef, sig: &Signature) -> Result<()> {
1691
        let key = format!("{}{}", fanout_prefix("sig", r), sig.key.to_hex());
1692
        let bytes = Signature::encode_list(std::slice::from_ref(sig));
1693
        if get_optional(self.backend.as_ref(), &key)?.as_deref() == Some(bytes.as_slice()) {
1694
            return Ok(());
1695
        }
1696
        delete(self.backend.as_ref(), &key)?;
1697
        put_new_or_existing(self.backend.as_ref(), &key, &bytes)
1698
    }
1699
1700
    /// Every object reference in the store, loose and packed (slow; for gc and tests).
1701
    pub fn list(&self) -> io::Result<Vec<ObjectRef>> {
1702
        let mut out: Vec<ObjectRef> = self
1703
            .loose_objects()
1704
            .map_err(|e| io::Error::other(e.to_string()))?
1705
            .into_iter()
1706
            .map(|(r, _)| r)
1707
            .collect();
1708
        for p in self.packs.borrow().iter() {
1709
            out.extend(p.refs.iter().copied());
1710
        }
1711
        out.sort();
1712
        out.dedup();
1713
        Ok(out)
1714
    }
1715
}
1716
1717
/// A loose object's bytes, verified against `r`.
1718
fn parse_loose(data: &[u8], r: &ObjectRef) -> Result<Object> {
1719
    let nul = data
1720
        .iter()
1721
        .position(|&b| b == 0)
1722
        .ok_or_else(|| StoreError::Corrupt(*r, "no header".into()))?;
1723
    let head = std::str::from_utf8(&data[..nul])
1724
        .map_err(|_| StoreError::Corrupt(*r, "header not UTF-8".into()))?;
1725
    let mut parts = head.split(' ');
1726
    let (Some("oxy1"), Some(ty), Some(len), None) =
1727
        (parts.next(), parts.next(), parts.next(), parts.next())
1728
    else {
1729
        return Err(StoreError::Corrupt(*r, format!("bad header {head:?}")));
1730
    };
1731
    let ty = ObjectType::from_name(ty)
1732
        .ok_or_else(|| StoreError::Corrupt(*r, format!("unknown type {ty}")))?;
1733
    let len: usize = len
1734
        .parse()
1735
        .map_err(|_| StoreError::Corrupt(*r, "bad length".into()))?;
1736
    let body = &data[nul + 1..];
1737
    if body.len() != len {
1738
        return Err(StoreError::Corrupt(
1739
            *r,
1740
            format!("length {} != header {len}", body.len()),
1741
        ));
1742
    }
1743
    if hash_object(ty, body) != *r {
1744
        return Err(StoreError::Corrupt(*r, "hash mismatch".into()));
1745
    }
1746
    Ok((ty, body.to_vec()))
1747
}
1748
1749
/// Whether the objects are files that a test may tamper with: not under
1750
/// `OXY_STORE_BACKEND=memory`.
1751
#[cfg(test)]
1752
fn files_on_disk() -> bool {
1753
    std::env::var_os("OXY_STORE_BACKEND").is_none()
1754
}
1755
1756
#[cfg(test)]
1757
mod tests {
1758
    use super::*;
1759
1760
    /// The store under test twice: in a directory, and on a store in
1761
    /// memory with a scratch directory, as over a bucket.
1762
    fn each_store(dir: &Path, f: impl Fn(&str, Store)) {
1763
        f("directory", Store::open(dir.join("objects")).unwrap());
1764
        f(
1765
            "memory",
1766
            Store::open_on(Arc::new(MemoryStore::new()), dir.join("scratch")).unwrap(),
1767
        );
1768
    }
1769
1770
    #[test]
1771
    fn put_get_verify() {
1772
        if !super::files_on_disk() {
1773
            return;
1774
        }
1775
        let dir = tempfile::tempdir().unwrap();
1776
        let s = Store::open(dir.path().join("objects")).unwrap();
1777
        let r = s.put(ObjectType::Chunk, b"hello").unwrap();
1778
        assert!(s.has(&r));
1779
        assert_eq!(s.get(&r).unwrap(), (ObjectType::Chunk, b"hello".to_vec()));
1780
        assert_eq!(s.put(ObjectType::Chunk, b"hello").unwrap(), r);
1781
        assert_eq!(s.list().unwrap(), vec![r]);
1782
        assert!(matches!(
1783
            s.get_typed(&r, ObjectType::Blob),
1784
            Err(StoreError::WrongType(..))
1785
        ));
1786
        assert!(matches!(
1787
            s.get(&ObjectRef([9; 32])),
1788
            Err(StoreError::NotFound(_))
1789
        ));
1790
1791
        // Tamper: flip a byte, read must fail.
1792
        let path = s.path_for(&r);
1793
        let mut data = fs::read(&path).unwrap();
1794
        let last = data.len() - 1;
1795
        data[last] ^= 1;
1796
        fs::write(&path, data).unwrap();
1797
        assert!(matches!(s.get(&r), Err(StoreError::Corrupt(..))));
1798
    }
1799
1800
    #[test]
1801
    fn packing_keeps_every_object_readable() {
1802
        if !super::files_on_disk() {
1803
            return;
1804
        }
1805
        let dir = tempfile::tempdir().unwrap();
1806
        let s = Store::open(dir.path()).unwrap();
1807
        let mut refs = Vec::new();
1808
        for i in 0..300u32 {
1809
            let body = format!("object {i} ").repeat(1 + (i % 7) as usize);
1810
            refs.push((s.put(ObjectType::Chunk, body.as_bytes()).unwrap(), body));
1811
        }
1812
        let c = crate::objects::tests::sample_change();
1813
        let cr = s.put_change(&c).unwrap();
1814
        assert_eq!(s.loose_count().unwrap(), 301);
1815
        let report = s.pack_loose().unwrap();
1816
        assert_eq!(report.objects, 301);
1817
        assert!(report.packed_bytes < report.raw_bytes, "{report:?}");
1818
        assert_eq!(s.loose_count().unwrap(), 0);
1819
        assert_eq!(s.pack_count(), 1);
1820
        for (r, body) in &refs {
1821
            assert!(s.has(r));
1822
            assert_eq!(
1823
                s.get(r).unwrap(),
1824
                (ObjectType::Chunk, body.as_bytes().to_vec())
1825
            );
1826
        }
1827
        assert_eq!(s.get_change(&cr).unwrap(), c);
1828
        assert_eq!(s.list().unwrap().len(), 301);
1829
        // Re-putting a packed object is a no-op; new objects go loose and a
1830
        // second pack coexists with the first.
1831
        assert_eq!(
1832
            s.put(ObjectType::Chunk, refs[0].1.as_bytes()).unwrap(),
1833
            refs[0].0
1834
        );
1835
        assert_eq!(s.loose_count().unwrap(), 0);
1836
        let extra = s.put(ObjectType::Chunk, b"after packing").unwrap();
1837
        s.pack_loose().unwrap();
1838
        assert_eq!(s.pack_count(), 2);
1839
        assert_eq!(s.get(&extra).unwrap().1, b"after packing");
1840
        // A reopened store sees the packs.
1841
        let s2 = Store::open(dir.path()).unwrap();
1842
        assert_eq!(s2.pack_count(), 2);
1843
        assert_eq!(s2.get(&refs[5].0).unwrap().1, refs[5].1.as_bytes());
1844
        assert!(matches!(
1845
            s2.get(&ObjectRef([9; 32])),
1846
            Err(StoreError::NotFound(_))
1847
        ));
1848
        // Corrupting an object's body inside the pack is detected on read.
1849
        let target = refs[5].0;
1850
        let (pack_path, offset) = {
1851
            let packs = s2.packs.borrow();
1852
            let (i, e) = s2.find_packed(&target).unwrap();
1853
            (dir.path().join(&packs[i].key), e.offset as usize)
1854
        };
1855
        let mut data = fs::read(&pack_path).unwrap();
1856
        data[offset] ^= 0xff;
1857
        fs::write(&pack_path, data).unwrap();
1858
        let s3 = Store::open(dir.path()).unwrap();
1859
        assert!(matches!(s3.get(&target), Err(StoreError::Corrupt(..))));
1860
        assert_eq!(
1861
            s3.get(&refs[6].0).unwrap().1,
1862
            refs[6].1.as_bytes(),
1863
            "other objects are unaffected"
1864
        );
1865
    }
1866
1867
    #[test]
1868
    fn a_full_repack_trains_a_dictionary_and_reads_back() {
1869
        if !super::files_on_disk() {
1870
            return;
1871
        }
1872
        let dir = tempfile::tempdir().unwrap();
1873
        let s = Store::open(dir.path()).unwrap();
1874
        // Many small, alike files: what a dictionary is for.
1875
        let bodies: Vec<Vec<u8>> = (0..1500)
1876
            .map(|i| {
1877
                format!(
1878
                    "// SPDX-License-Identifier: GPL-2.0\n#include <linux/module.h>\n\
1879
                     static int probe_{i}(struct device *dev)\n{{\n\treturn {i};\n}}\n"
1880
                )
1881
                .into_bytes()
1882
            })
1883
            .collect();
1884
        let refs: Vec<ObjectRef> = bodies
1885
            .iter()
1886
            .map(|b| s.put(ObjectType::Chunk, b).unwrap())
1887
            .collect();
1888
        s.repack_all(REPACK_ZSTD_LEVEL).unwrap();
1889
        let dicts = fs::read_dir(dir.path())
1890
            .unwrap()
1891
            .flatten()
1892
            .filter(|e| e.path().extension().is_some_and(|x| x == "dict"))
1893
            .count();
1894
        assert_eq!(dicts, 1, "the pack has a dictionary");
1895
        let with_dict = refs
1896
            .iter()
1897
            .filter(|r| s.find_packed(r).unwrap().1.flags & FLAG_DICT != 0)
1898
            .count();
1899
        assert!(with_dict > 1000, "most use it: {with_dict}");
1900
        let again = Store::open(dir.path()).unwrap();
1901
        for (r, b) in refs.iter().zip(&bodies) {
1902
            assert_eq!(&again.get(r).unwrap().1, b);
1903
        }
1904
        // A repack after that replaces the pack and its dictionary.
1905
        again.repack_all(REPACK_ZSTD_LEVEL).unwrap();
1906
        let files: Vec<String> = fs::read_dir(dir.path())
1907
            .unwrap()
1908
            .flatten()
1909
            .map(|e| e.file_name().to_string_lossy().into_owned())
1910
            .filter(|n| n.starts_with("pack-"))
1911
            .collect();
1912
        assert_eq!(files.len(), 3, "one pack, index and dictionary: {files:?}");
1913
        assert_eq!(&again.get(&refs[7]).unwrap().1, &bodies[7]);
1914
    }
1915
1916
    #[test]
1917
    fn nocompress_chunks_pack_raw() {
1918
        let dir = tempfile::tempdir().unwrap();
1919
        let s = Store::open(dir.path()).unwrap();
1920
        let body = vec![b'a'; 10_000];
1921
        let raw = s.put_chunk_nocompress(&body).unwrap();
1922
        let compressible = s.put(ObjectType::Chunk, &vec![b'b'; 10_000]).unwrap();
1923
        s.pack_loose().unwrap();
1924
        let packs = s.packs.borrow();
1925
        let (i, e) = s.find_packed(&raw).unwrap();
1926
        assert_eq!((e.flags, e.stored_len as usize), (0, 10_000));
1927
        let (_, c) = s.find_packed(&compressible).unwrap();
1928
        assert_eq!(c.flags, FLAG_ZSTD);
1929
        assert!(c.stored_len < 10_000);
1930
        assert_eq!(packs[i].fetch(s.backend.as_ref(), &raw, e).unwrap().1, body);
1931
    }
1932
1933
    #[test]
1934
    fn many_packs_merge_into_one() {
1935
        let dir = tempfile::tempdir().unwrap();
1936
        each_store(dir.path(), |which, s| {
1937
            let mut refs = Vec::new();
1938
            for i in 0..(MAX_PACKS + 2) {
1939
                refs.push(
1940
                    s.put(ObjectType::Chunk, format!("pack {i}").as_bytes())
1941
                        .unwrap(),
1942
                );
1943
                s.pack_loose().unwrap();
1944
            }
1945
            assert_eq!(s.pack_count(), MAX_PACKS + 2, "{which}");
1946
            assert_eq!(s.merge_packs().unwrap(), MAX_PACKS + 2);
1947
            assert_eq!(s.pack_count(), 1);
1948
            assert_eq!(s.merge_packs().unwrap(), 0);
1949
            for (i, r) in refs.iter().enumerate() {
1950
                assert_eq!(s.get(r).unwrap().1, format!("pack {i}").as_bytes());
1951
            }
1952
            let again = Store::open_on(s.backend().clone(), s.dir()).unwrap();
1953
            assert_eq!(again.pack_count(), 1, "{which}: reopened");
1954
        });
1955
    }
1956
1957
    #[test]
1958
    fn prune_removes_unreachable_loose_and_packed_objects() {
1959
        use std::collections::HashSet;
1960
        let dir = tempfile::tempdir().unwrap();
1961
        each_store(dir.path(), |which, s| {
1962
            let keep_a = s.put(ObjectType::Chunk, b"keep a").unwrap();
1963
            let drop_a = s.put(ObjectType::Chunk, b"drop a").unwrap();
1964
            s.pack_loose().unwrap();
1965
            let keep_b = s.put(ObjectType::Chunk, b"keep b").unwrap();
1966
            let drop_b = s.put(ObjectType::Chunk, b"drop b").unwrap();
1967
            let sk = crate::identity::SecretKey {
1968
                ed_seed: [1; 32],
1969
                ml_seed: [2; 32],
1970
            };
1971
            let id = sk.identity("a", "b", 0);
1972
            let key = s.put_identity(&id).unwrap();
1973
            s.put_signature(
1974
                &drop_b,
1975
                &sk.sign(crate::identity::Domain::Change, &key, &drop_b),
1976
            )
1977
            .unwrap();
1978
            s.index_seal(&drop_a, &ObjectRef([3; 32])).unwrap();
1979
            let keep: HashSet<ObjectRef> = [keep_a, keep_b, key].into_iter().collect();
1980
            // Everything is too young under a grace period.
1981
            let r = s
1982
                .prune(&keep, std::time::Duration::from_secs(3600))
1983
                .unwrap();
1984
            assert_eq!((r.objects, r.deferred), (0, 2), "{which}");
1985
            let r = s.prune(&keep, std::time::Duration::ZERO).unwrap();
1986
            assert_eq!(r.objects, 2, "{which}: {r:?}");
1987
            assert!(s.has(&keep_a) && s.has(&keep_b) && s.has(&key));
1988
            assert!(!s.has(&drop_a) && !s.has(&drop_b));
1989
            assert!(s.signatures(&drop_b).unwrap().is_empty());
1990
            assert!(s.seals_for(&drop_a).unwrap().is_empty());
1991
            assert_eq!(s.pack_count(), 1);
1992
            assert_eq!(s.get(&keep_a).unwrap().1, b"keep a");
1993
            assert_eq!(s.list().unwrap().len(), 3);
1994
        });
1995
    }
1996
1997
    #[test]
1998
    fn signatures_are_stored_beside_objects() {
1999
        use crate::identity::{Domain, SecretKey};
2000
        let dir = tempfile::tempdir().unwrap();
2001
        each_store(dir.path(), |which, s| {
2002
            let r = s.put(ObjectType::Chunk, b"x").unwrap();
2003
            assert!(s.signatures(&r).unwrap().is_empty());
2004
            let sk = SecretKey {
2005
                ed_seed: [1; 32],
2006
                ml_seed: [2; 32],
2007
            };
2008
            let id = sk.identity("a", "b", 0);
2009
            let key = s.put_identity(&id).unwrap();
2010
            assert_eq!(s.get_identity(&key).unwrap(), id);
2011
            let sig = sk.sign(Domain::Change, &key, &r);
2012
            s.put_signature(&r, &sig).unwrap();
2013
            s.put_signature(&r, &sig).unwrap();
2014
            assert_eq!(s.signatures(&r).unwrap(), vec![sig.clone()], "{which}");
2015
            // A second signer is kept beside the first, sorted by key.
2016
            let sk2 = SecretKey {
2017
                ed_seed: [3; 32],
2018
                ml_seed: [4; 32],
2019
            };
2020
            let key2 = s.put_identity(&sk2.identity("c", "d", 0)).unwrap();
2021
            let sig2 = sk2.sign(Domain::Change, &key2, &r);
2022
            s.put_signature(&r, &sig2).unwrap();
2023
            let mut both = vec![sig.clone(), sig2.clone()];
2024
            both.sort_by_key(|s| s.key);
2025
            assert_eq!(s.signatures(&r).unwrap(), both);
2026
            // The object itself is untouched by its signatures.
2027
            assert_eq!(s.get(&r).unwrap().1, b"x");
2028
            // Seal and receipt entries, and the states with receipts.
2029
            s.index_seal(&r, &ObjectRef([5; 32])).unwrap();
2030
            s.index_seal(&r, &ObjectRef([4; 32])).unwrap();
2031
            s.index_seal(&r, &ObjectRef([5; 32])).unwrap();
2032
            assert_eq!(
2033
                s.seals_for(&r).unwrap(),
2034
                vec![ObjectRef([4; 32]), ObjectRef([5; 32])]
2035
            );
2036
            let st = ObjectRef([6; 32]);
2037
            s.index_receipt(&st, &ObjectRef([7; 32])).unwrap();
2038
            assert_eq!(s.receipts_for(&st).unwrap(), vec![ObjectRef([7; 32])]);
2039
            assert_eq!(s.receipt_states().unwrap(), vec![st]);
2040
        });
2041
    }
2042
2043
    /// A store written before signatures, seals and receipts were one
2044
    /// entry each (`sigs/`, `seals/`, `receipts/` lists) still reads, and
2045
    /// what is written now sits beside what was there.
2046
    #[test]
2047
    fn an_older_store_s_sidecar_lists_are_still_read() {
2048
        if !super::files_on_disk() {
2049
            return;
2050
        }
2051
        use crate::identity::{Domain, SecretKey};
2052
        let dir = tempfile::tempdir().unwrap();
2053
        let s = Store::open(dir.path()).unwrap();
2054
        let r = s.put(ObjectType::Chunk, b"old").unwrap();
2055
        let h = r.to_hex();
2056
        let sk = SecretKey {
2057
            ed_seed: [1; 32],
2058
            ml_seed: [2; 32],
2059
        };
2060
        let key = s.put_identity(&sk.identity("a", "b", 0)).unwrap();
2061
        let sig = sk.sign(Domain::Change, &key, &r);
2062
        let legacy = dir.path().join("sigs").join(&h[..2]);
2063
        fs::create_dir_all(&legacy).unwrap();
2064
        fs::write(
2065
            legacy.join(&h[2..]),
2066
            Signature::encode_list(std::slice::from_ref(&sig)),
2067
        )
2068
        .unwrap();
2069
        let seals = dir.path().join("seals").join(&h[..2]);
2070
        fs::create_dir_all(&seals).unwrap();
2071
        fs::write(
2072
            seals.join(&h[2..]),
2073
            format!("{}\n", ObjectRef([9; 32]).to_hex()),
2074
        )
2075
        .unwrap();
2076
        let receipts = dir.path().join("receipts").join(&h[..2]);
2077
        fs::create_dir_all(&receipts).unwrap();
2078
        fs::write(
2079
            receipts.join(&h[2..]),
2080
            format!("{}\n", ObjectRef([8; 32]).to_hex()),
2081
        )
2082
        .unwrap();
2083
2084
        assert_eq!(s.signatures(&r).unwrap(), vec![sig.clone()]);
2085
        assert_eq!(s.seals_for(&r).unwrap(), vec![ObjectRef([9; 32])]);
2086
        assert_eq!(s.receipts_for(&r).unwrap(), vec![ObjectRef([8; 32])]);
2087
        assert_eq!(s.receipt_states().unwrap(), vec![r]);
2088
        // New entries join the old ones; the same signer is not doubled.
2089
        s.put_signature(&r, &sig).unwrap();
2090
        s.index_seal(&r, &ObjectRef([1; 32])).unwrap();
2091
        assert_eq!(s.signatures(&r).unwrap().len(), 1);
2092
        assert_eq!(
2093
            s.seals_for(&r).unwrap(),
2094
            vec![ObjectRef([1; 32]), ObjectRef([9; 32])]
2095
        );
2096
        // Pruning the object takes both layouts with it.
2097
        let r2 = s.put(ObjectType::Chunk, b"kept").unwrap();
2098
        let keep = [r2, key].into_iter().collect();
2099
        s.prune(&keep, std::time::Duration::ZERO).unwrap();
2100
        assert!(s.signatures(&r).unwrap().is_empty());
2101
        assert!(s.seals_for(&r).unwrap().is_empty());
2102
        assert!(s.receipt_states().unwrap().is_empty());
2103
    }
2104
2105
    #[test]
2106
    fn change_round_trip_through_store() {
2107
        let dir = tempfile::tempdir().unwrap();
2108
        each_store(dir.path(), |which, s| {
2109
            let c = crate::objects::tests::sample_change();
2110
            let r = s.put_change(&c).unwrap();
2111
            assert_eq!(r, c.hash(), "{which}");
2112
            assert_eq!(s.get_change(&r).unwrap(), c);
2113
            assert_eq!(s.get_change_meta(&r).unwrap().message, c.message);
2114
        });
2115
    }
2116
2117
    fn xorshift(state: &mut u64) -> u64 {
2118
        *state ^= *state << 13;
2119
        *state ^= *state >> 7;
2120
        *state ^= *state << 17;
2121
        *state
2122
    }
2123
2124
    /// Random corruption of pack and index files never panics: reads fail
2125
    /// with an error or the store refuses to open, and untouched objects
2126
    /// stay readable when the index survives.
2127
    #[test]
2128
    fn corrupted_packs_and_indexes_fail_cleanly() {
2129
        if !super::files_on_disk() {
2130
            return;
2131
        }
2132
        let dir = tempfile::tempdir().unwrap();
2133
        let objects = dir.path().join("objects");
2134
        let s = Store::open(&objects).unwrap();
2135
        let mut refs = Vec::new();
2136
        let mut seed = 0x9e3779b97f4a7c15u64;
2137
        for i in 0..40u32 {
2138
            let len = (xorshift(&mut seed) % 5000) as usize;
2139
            let mut body = vec![0u8; len];
2140
            for b in body.iter_mut() {
2141
                *b = if i % 2 == 0 {
2142
                    (xorshift(&mut seed) & 0xff) as u8
2143
                } else {
2144
                    b'a' + (i % 26) as u8
2145
                };
2146
            }
2147
            refs.push(s.put(ObjectType::Chunk, &body).unwrap());
2148
        }
2149
        s.pack_loose().unwrap();
2150
        let pack = fs::read_dir(&objects)
2151
            .unwrap()
2152
            .filter_map(|e| e.ok().map(|e| e.path()))
2153
            .find(|p| p.extension().is_some_and(|x| x == "pack"))
2154
            .unwrap();
2155
        let idx = pack.with_extension("idx");
2156
        let pristine_pack = fs::read(&pack).unwrap();
2157
        let pristine_idx = fs::read(&idx).unwrap();
2158
2159
        for round in 0..300u32 {
2160
            let (path, pristine) = if round % 2 == 0 {
2161
                (&pack, &pristine_pack)
2162
            } else {
2163
                (&idx, &pristine_idx)
2164
            };
2165
            let mut data = pristine.clone();
2166
            match xorshift(&mut seed) % 4 {
2167
                0 => {
2168
                    let at = (xorshift(&mut seed) as usize) % data.len();
2169
                    data[at] ^= 1 << (xorshift(&mut seed) % 8);
2170
                }
2171
                1 => {
2172
                    let at = (xorshift(&mut seed) as usize) % data.len();
2173
                    data[at] = (xorshift(&mut seed) & 0xff) as u8;
2174
                }
2175
                2 => {
2176
                    let keep = (xorshift(&mut seed) as usize) % data.len();
2177
                    data.truncate(keep);
2178
                }
2179
                _ => {
2180
                    let at = (xorshift(&mut seed) as usize) % data.len();
2181
                    let n = ((xorshift(&mut seed) % 8) + 1) as usize;
2182
                    for k in 0..n.min(data.len() - at) {
2183
                        data[at + k] = 0xff;
2184
                    }
2185
                }
2186
            }
2187
            fs::write(path, &data).unwrap();
2188
            // Opening may refuse a broken index; a store that opens must
2189
            // either return each object intact or report it corrupt/missing.
2190
            if let Ok(s2) = Store::open(&objects) {
2191
                for r in &refs {
2192
                    match s2.get(r) {
2193
                        Ok((ty, body)) => assert_eq!(
2194
                            hash_object(ty, &body),
2195
                            *r,
2196
                            "a returned object always matches its reference"
2197
                        ),
2198
                        Err(StoreError::Corrupt(..))
2199
                        | Err(StoreError::NotFound(_))
2200
                        | Err(StoreError::Io(_)) => {}
2201
                        Err(e) => panic!("unexpected error {e}"),
2202
                    }
2203
                }
2204
            }
2205
            fs::write(path, pristine).unwrap();
2206
        }
2207
        // Restored files read back in full.
2208
        let s3 = Store::open(&objects).unwrap();
2209
        for r in &refs {
2210
            let (ty, body) = s3.get(r).unwrap();
2211
            assert_eq!(hash_object(ty, &body), *r);
2212
        }
2213
    }
2214
2215
    /// A small batch becomes loose objects; a large one becomes a single
2216
    /// pack. Objects are readable while pending, and batches nest.
2217
    #[test]
2218
    fn batches_write_loose_or_one_pack() {
2219
        let dir = tempfile::tempdir().unwrap();
2220
        each_store(dir.path(), |which, s| {
2221
            let outer = s.begin_batch().unwrap();
2222
            let a = s.put(ObjectType::Chunk, b"small a").unwrap();
2223
            {
2224
                let inner = s.begin_batch().unwrap();
2225
                let b = s.put(ObjectType::Chunk, b"small b").unwrap();
2226
                assert!(s.has(&b));
2227
                assert_eq!(s.get(&b).unwrap().1, b"small b");
2228
                inner.finish().unwrap();
2229
                assert_eq!(
2230
                    s.loose_count().unwrap(),
2231
                    0,
2232
                    "{which}: the inner batch defers to the outer one"
2233
                );
2234
            }
2235
            assert_eq!(s.get(&a).unwrap().1, b"small a");
2236
            assert_eq!(
2237
                s.put(ObjectType::Chunk, b"small a").unwrap(),
2238
                a,
2239
                "idempotent while pending"
2240
            );
2241
            outer.finish().unwrap();
2242
            assert_eq!(s.loose_count().unwrap(), 2);
2243
            assert_eq!(s.pack_count(), 0);
2244
            assert_eq!(s.get(&a).unwrap().1, b"small a");
2245
2246
            // Large: spills into one pack, readable during and after.
2247
            let big = s.begin_batch().unwrap();
2248
            let mut refs = Vec::new();
2249
            let mut seed = 7u64;
2250
            for i in 0..300u32 {
2251
                let mut body = vec![b'z'; 40_000];
2252
                if i % 3 == 0 {
2253
                    for b in body.iter_mut() {
2254
                        *b = (xorshift(&mut seed) & 0xff) as u8;
2255
                    }
2256
                }
2257
                body[..4].copy_from_slice(&i.to_le_bytes());
2258
                let r = if i % 5 == 0 {
2259
                    s.put_chunk_nocompress(&body).unwrap()
2260
                } else {
2261
                    s.put(ObjectType::Chunk, &body).unwrap()
2262
                };
2263
                assert_eq!(
2264
                    s.get(&r).unwrap().1,
2265
                    body,
2266
                    "readable while pending (object {i})"
2267
                );
2268
                refs.push((r, body));
2269
            }
2270
            assert!(s.has(&refs[0].0));
2271
            big.finish().unwrap();
2272
            assert_eq!(s.pack_count(), 1);
2273
            assert_eq!(
2274
                s.loose_count().unwrap(),
2275
                2,
2276
                "nothing from the big batch is loose"
2277
            );
2278
            assert!(fs::read_dir(s.dir())
2279
                .unwrap()
2280
                .flatten()
2281
                .all(|e| !e.file_name().to_string_lossy().contains("tmp-")));
2282
            let s2 = Store::open_on(s.backend().clone(), s.dir()).unwrap();
2283
            for (r, body) in &refs {
2284
                assert_eq!(s2.get(r).unwrap().1, *body);
2285
            }
2286
            // A dropped batch still writes its objects.
2287
            {
2288
                let _b = s2.begin_batch().unwrap();
2289
                s2.put(ObjectType::Chunk, b"dropped").unwrap();
2290
            }
2291
            assert_eq!(
2292
                s2.get(&hash_object(ObjectType::Chunk, b"dropped"))
2293
                    .unwrap()
2294
                    .1,
2295
                b"dropped"
2296
            );
2297
        });
2298
    }
2299
2300
    #[test]
2301
    fn get_many_reads_loose_and_packed_objects_in_order() {
2302
        let dir = tempfile::tempdir().unwrap();
2303
        each_store(dir.path(), |which, s| {
2304
            let packed: Vec<ObjectRef> = (0..20)
2305
                .map(|i| {
2306
                    s.put(ObjectType::Chunk, format!("packed {i}").as_bytes())
2307
                        .unwrap()
2308
                })
2309
                .collect();
2310
            s.pack_loose().unwrap();
2311
            let loose: Vec<ObjectRef> = (0..20)
2312
                .map(|i| {
2313
                    s.put(ObjectType::Chunk, format!("loose {i}").as_bytes())
2314
                        .unwrap()
2315
                })
2316
                .collect();
2317
            let missing = hash_object(ObjectType::Chunk, b"never stored");
2318
            let mut refs: Vec<ObjectRef> = Vec::new();
2319
            for i in 0..20 {
2320
                refs.push(loose[i]);
2321
                refs.push(packed[i]);
2322
            }
2323
            refs.push(missing);
2324
            let got = s.get_many(&refs);
2325
            assert_eq!(got.len(), refs.len());
2326
            for (i, r) in refs.iter().enumerate().take(40) {
2327
                let want = if i % 2 == 0 {
2328
                    format!("loose {}", i / 2)
2329
                } else {
2330
                    format!("packed {}", i / 2)
2331
                };
2332
                let (ty, body) = got[i].as_ref().unwrap();
2333
                assert_eq!(
2334
                    (*ty, body.as_slice()),
2335
                    (ObjectType::Chunk, want.as_bytes()),
2336
                    "{which}: {r:?}"
2337
                );
2338
            }
2339
            assert!(matches!(got[40], Err(StoreError::NotFound(_))));
2340
        });
2341
    }
2342
}
2343
2344
#[cfg(test)]
2345
mod batch_tests {
2346
    use super::*;
2347
2348
    #[test]
2349
    fn a_large_batch_packs_every_object_and_indexes_them_all() {
2350
        let dir = tempfile::tempdir().unwrap();
2351
        let store = Store::open(dir.path()).unwrap();
2352
        let mut refs = Vec::new();
2353
        {
2354
            let batch = store.begin_batch().unwrap();
2355
            for i in 0..1000u32 {
2356
                let body = format!("object {i} {}", "x".repeat(20_000 + i as usize)).into_bytes();
2357
                refs.push(store.put(ObjectType::Blob, &body).unwrap());
2358
            }
2359
            batch.finish().unwrap();
2360
        }
2361
        let again = Store::open(dir.path()).unwrap();
2362
        for r in &refs {
2363
            again
2364
                .get(r)
2365
                .unwrap_or_else(|e| panic!("{}: {e}", r.short(12)));
2366
        }
2367
    }
2368
}
2369
2370
#[cfg(test)]
2371
mod size_tests {
2372
    use super::*;
2373
2374
    #[test]
2375
    fn an_object_over_the_limit_is_refused_and_nothing_is_written() {
2376
        let dir = tempfile::tempdir().unwrap();
2377
        let store = Store::open(dir.path()).unwrap();
2378
        let big = vec![b'x'; MAX_OBJECT_LEN as usize + 1];
2379
        {
2380
            let batch = store.begin_batch().unwrap();
2381
            store.put(ObjectType::Blob, b"small").unwrap();
2382
            let e = store.put(ObjectType::Change, &big).unwrap_err();
2383
            assert!(
2384
                matches!(e, StoreError::TooLarge { kind: "change", .. }),
2385
                "{e}"
2386
            );
2387
            assert!(e.to_string().contains("commit it in parts"), "{e}");
2388
            batch.finish().unwrap();
2389
        }
2390
        // The store still opens and reads what was written.
2391
        let again = Store::open(dir.path()).unwrap();
2392
        again.get(&hash_object(ObjectType::Blob, b"small")).unwrap();
2393
    }
2394
}
2395
2396
#[cfg(test)]
2397
mod text_tests {
2398
    use super::*;
2399
    use crate::objects::{AtomRef, Edit, CHANGE_VERSION};
2400
2401
    fn large_change() -> Change {
2402
        let mut c = crate::objects::tests::sample_change();
2403
        assert_eq!(c.version, CHANGE_VERSION);
2404
        c.edits[0] = Edit::Insert {
2405
            node: crate::ids::NodeId([7; 16]),
2406
            after: AtomRef::Start,
2407
            before: AtomRef::End,
2408
            lines: (0..2000)
2409
                .map(|i| format!("line {i}\n").into_bytes())
2410
                .collect(),
2411
        };
2412
        c
2413
    }
2414
2415
    #[test]
2416
    fn a_large_insert_is_stored_beside_its_change_and_read_back() {
2417
        let dir = tempfile::tempdir().unwrap();
2418
        let s = Store::open(dir.path()).unwrap();
2419
        let c = large_change();
2420
        let r = s.put_change(&c).unwrap();
2421
        assert_eq!(r, c.hash());
2422
        let (_, body) = s.get(&r).unwrap();
2423
        assert!(
2424
            body.len() < 1024,
2425
            "the change holds a reference, not {} bytes of lines",
2426
            body.len()
2427
        );
2428
        let (back, parts) = s.get_change_parts(&r).unwrap();
2429
        assert_eq!(back, c);
2430
        // The text blob and its chunk.
2431
        assert_eq!(parts.len(), 2);
2432
        for p in &parts {
2433
            assert!(s.has(p));
2434
        }
2435
        // Storing it again writes nothing new.
2436
        let n = s.list().unwrap().len();
2437
        s.put_change(&c).unwrap();
2438
        assert_eq!(s.list().unwrap().len(), n);
2439
    }
2440
2441
    #[test]
2442
    fn a_change_whose_text_blob_is_missing_is_reported_not_misread() {
2443
        if !super::files_on_disk() {
2444
            return;
2445
        }
2446
        let dir = tempfile::tempdir().unwrap();
2447
        let s = Store::open(dir.path()).unwrap();
2448
        let c = large_change();
2449
        let r = s.put_change(&c).unwrap();
2450
        let (_, parts) = s.get_change_parts(&r).unwrap();
2451
        fs::remove_file(s.path_for(&parts[0])).unwrap();
2452
        let e = s.get_change(&r).unwrap_err();
2453
        assert!(
2454
            matches!(&e, StoreError::Corrupt(h, why) if *h == r && why.contains("missing")),
2455
            "{e}"
2456
        );
2457
    }
2458
}
2459
2460
#[cfg(test)]
2461
mod put_many_tests {
2462
    use super::*;
2463
2464
    #[test]
2465
    fn many_objects_in_a_batch_are_compressed_together_and_all_read_back() {
2466
        let dir = tempfile::tempdir().unwrap();
2467
        let s = Store::open(dir.path()).unwrap();
2468
        let before = s.put(ObjectType::Chunk, b"already here").unwrap();
2469
        let bodies: Vec<Vec<u8>> = (0..200u32)
2470
            .map(|i| format!("chunk {i} ").repeat(6000).into_bytes())
2471
            .collect();
2472
        let mut objects: Vec<(ObjectType, &[u8])> = bodies
2473
            .iter()
2474
            .map(|b| (ObjectType::Chunk, b.as_slice()))
2475
            .collect();
2476
        objects.push((ObjectType::Chunk, bodies[0].as_slice()));
2477
        objects.push((ObjectType::Chunk, b"already here"));
2478
        let refs = {
2479
            let batch = s.begin_batch().unwrap();
2480
            let refs = s.put_many(&objects).unwrap();
2481
            batch.finish().unwrap();
2482
            refs
2483
        };
2484
        assert_eq!(refs.len(), objects.len());
2485
        assert_eq!(refs[200], refs[0], "a repeat is the same object");
2486
        assert_eq!(refs[201], before);
2487
        let again = Store::open(dir.path()).unwrap();
2488
        for (r, b) in refs.iter().zip(&objects) {
2489
            assert_eq!(again.get(r).unwrap().1, b.1);
2490
        }
2491
    }
2492
}