OXYGENHUB
◆ v0.3.0 oxygen / crates / oxy-core / src / store.rs
80.9 KiB · Rust · last changed First commit: 12 s and 2.1 GB on the Linux tree (refs #41) 1 month ago source blame history raw
d02e545d Andy McCall ↶ 1
//! Object store (spec §13.1, §13.3, §13.4): loose objects, one file per
2
//! object under `objects/loose/<hh>/<rest>`, and packs, `pack-<id>.pack`
3
//! with a sorted `pack-<id>.idx`, holding zstd-compressed objects. Reads
4
//! check loose objects first, then every pack; every read re-hashes.
5
//!
6
//! Pack file: `OXYPACK1` then entries of
7
//! `type u8 | flags u8 (bit 0: zstd) | stored_len u32 | raw_len u32 | ref 32 | body`.
8
//! Index file: `OXYIDX01` then `count u32` then sorted entries of
9
//! `ref 32 | offset u64 | stored_len u32 | raw_len u32 | type u8 | flags u8`.
ff0af1ee unknown 10
5b9c6ad4 Andy McCall ↶ 11
use crate::attest::Attest;
ff0af1ee unknown 12
use crate::hash::{hash_object, header, ObjectRef, ObjectType};
1983686f Andy McCall ↶ 13
use crate::identity::{Identity, Signature};
6f8aeb99 Andy McCall ↶ 14
use crate::objects::{Blob, Change, ObjectError, StateObject};
fc1a466a Andy McCall ↶ 15
use crate::receipt::Receipt;
ff0af1ee unknown 16
use rand::RngCore;
d02e545d Andy McCall ↶ 17
use std::cell::RefCell;
ff0af1ee unknown 18
use std::fs;
d02e545d Andy McCall ↶ 19
use std::io::{self, Read, Seek, SeekFrom, Write};
ff0af1ee unknown 20
use std::path::{Path, PathBuf};
21
d02e545d Andy McCall ↶ 22
const PACK_MAGIC: &[u8; 8] = b"OXYPACK1";
23
const IDX_MAGIC: &[u8; 8] = b"OXYIDX01";
24
const IDX_ENTRY: usize = 32 + 8 + 4 + 4 + 1 + 1;
25
const FLAG_ZSTD: u8 = 1;
a17fc622 Andy McCall ↶ 26
/// Compressed with the pack's trained dictionary (`pack-ID.dict`), which a
27
/// full repack makes from a sample of the objects it packs: small files
28
/// share most of their vocabulary, which each one alone cannot use.
29
const FLAG_DICT: u8 = 2;
30
/// The size of a pack's dictionary, and the least a repack must pack for
31
/// one to pay.
32
const DICT_BYTES: usize = 110 << 10;
33
const DICT_MIN_OBJECTS: usize = 1000;
34
/// Objects up to this size are compressed with the dictionary.
35
const DICT_OBJECT_MAX: usize = 64 << 10;
d02e545d Andy McCall ↶ 36
const ZSTD_LEVEL: i32 = 3;
da652d63 Andy McCall ↶ 37
/// The level of a full repack (`oxy gc`): slower, once, for a smaller store.
38
pub const REPACK_ZSTD_LEVEL: i32 = 9;
2b872d07 Andy McCall ↶ 39
/// Largest object a pack entry may claim; anything bigger is corruption,
40
/// and refusing it keeps a bad index from driving huge allocations.
41
const MAX_OBJECT_LEN: u32 = 1 << 30;
d02e545d Andy McCall ↶ 42
afaba5f1 Andy McCall ↶ 43
/// An object read from the store: its type and body.
44
pub type Object = (ObjectType, Vec<u8>);
45
d02e545d Andy McCall ↶ 46
#[derive(Clone, Copy, Debug)]
47
struct PackEntry {
48
    offset: u64,
49
    stored_len: u32,
50
    raw_len: u32,
51
    ty: ObjectType,
52
    flags: u8,
53
}
54
55
/// One pack's index, fully in memory and sorted by ObjectRef.
56
struct PackIndex {
57
    path: PathBuf,
58
    refs: Vec<ObjectRef>,
59
    entries: Vec<PackEntry>,
a17fc622 Andy McCall ↶ 60
    /// The pack's dictionary, prepared once, when it has one.
61
    dict: Option<std::sync::Arc<zstd::dict::DecoderDictionary<'static>>>,
d02e545d Andy McCall ↶ 62
}
63
a17fc622 Andy McCall ↶ 64
impl std::fmt::Debug for PackIndex {
65
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
66
        f.debug_struct("PackIndex")
67
            .field("path", &self.path)
68
            .field("objects", &self.refs.len())
69
            .field("dict", &self.dict.is_some())
70
            .finish()
71
    }
72
}
73
d02e545d Andy McCall ↶ 74
impl PackIndex {
75
    fn load(idx_path: &Path, pack_path: &Path) -> Result<PackIndex> {
76
        let data = fs::read(idx_path)?;
75167672 Andy McCall 77
        let bad =
78
            |m: &str| StoreError::Corrupt(ObjectRef::ZERO, format!("{}: {m}", idx_path.display()));
d02e545d Andy McCall ↶ 79
        if data.len() < 12 || &data[..8] != IDX_MAGIC {
80
            return Err(bad("bad index header"));
81
        }
82
        let count = u32::from_le_bytes(data[8..12].try_into().unwrap()) as usize;
83
        if data.len() != 12 + count * IDX_ENTRY {
84
            return Err(bad("index length mismatch"));
85
        }
2b872d07 Andy McCall ↶ 86
        let pack_len = fs::metadata(pack_path)?.len();
d02e545d Andy McCall ↶ 87
        let mut refs = Vec::with_capacity(count);
88
        let mut entries = Vec::with_capacity(count);
89
        for i in 0..count {
90
            let e = &data[12 + i * IDX_ENTRY..12 + (i + 1) * IDX_ENTRY];
91
            let r = ObjectRef::from_slice(&e[..32]).unwrap();
92
            if let Some(prev) = refs.last() {
93
                if r <= *prev {
94
                    return Err(bad("index not sorted"));
95
                }
96
            }
97
            refs.push(r);
2b872d07 Andy McCall ↶ 98
            let entry = PackEntry {
d02e545d Andy McCall ↶ 99
                offset: u64::from_le_bytes(e[32..40].try_into().unwrap()),
100
                stored_len: u32::from_le_bytes(e[40..44].try_into().unwrap()),
101
                raw_len: u32::from_le_bytes(e[44..48].try_into().unwrap()),
75167672 Andy McCall 102
                ty: ObjectType::ALL
103
                    .get(e[48] as usize)
104
                    .copied()
105
                    .ok_or_else(|| bad("unknown object type"))?,
d02e545d Andy McCall ↶ 106
                flags: e[49],
2b872d07 Andy McCall ↶ 107
            };
a9f686ed Andy McCall ↶ 108
            if entry.stored_len > MAX_OBJECT_LEN || entry.raw_len > MAX_OBJECT_LEN {
109
                return Err(StoreError::Corrupt(
110
                    r,
111
                    format!(
112
                        "{}: the index claims an object of {} bytes ({} stored), more than the {MAX_OBJECT_LEN} an object may be",
113
                        idx_path.display(),
114
                        entry.raw_len,
115
                        entry.stored_len
116
                    ),
117
                ));
118
            }
119
            if entry.offset.saturating_add(entry.stored_len as u64) > pack_len {
2b872d07 Andy McCall ↶ 120
                return Err(bad("index entry points outside the pack"));
121
            }
122
            entries.push(entry);
d02e545d Andy McCall ↶ 123
        }
a17fc622 Andy McCall ↶ 124
        let dict = match fs::read(pack_path.with_extension("dict")) {
125
            Ok(d) => Some(std::sync::Arc::new(zstd::dict::DecoderDictionary::copy(&d))),
126
            Err(e) if e.kind() == io::ErrorKind::NotFound => None,
127
            Err(e) => return Err(e.into()),
128
        };
75167672 Andy McCall 129
        Ok(PackIndex {
130
            path: pack_path.to_path_buf(),
131
            refs,
132
            entries,
a17fc622 Andy McCall ↶ 133
            dict,
75167672 Andy McCall 134
        })
d02e545d Andy McCall ↶ 135
    }
136
137
    fn find(&self, r: &ObjectRef) -> Option<PackEntry> {
138
        self.refs.binary_search(r).ok().map(|i| self.entries[i])
139
    }
140
141
    fn read(&self, r: &ObjectRef, e: PackEntry) -> Result<(ObjectType, Vec<u8>)> {
142
        let mut f = fs::File::open(&self.path)?;
143
        f.seek(SeekFrom::Start(e.offset))?;
144
        let mut stored = vec![0u8; e.stored_len as usize];
145
        f.read_exact(&mut stored)?;
a17fc622 Andy McCall ↶ 146
        let body = if e.flags & FLAG_DICT != 0 {
147
            let dict = self.dict.as_ref().ok_or_else(|| {
148
                StoreError::Corrupt(*r, "packed with a dictionary the pack lacks".into())
149
            })?;
150
            zstd::bulk::Decompressor::with_prepared_dictionary(dict)
151
                .and_then(|mut d| d.decompress(&stored, e.raw_len as usize))
152
                .map_err(|err| StoreError::Corrupt(*r, format!("zstd: {err}")))?
153
        } else if e.flags & FLAG_ZSTD != 0 {
75167672 Andy McCall 154
            zstd::bulk::decompress(&stored, e.raw_len as usize)
155
                .map_err(|err| StoreError::Corrupt(*r, format!("zstd: {err}")))?
d02e545d Andy McCall ↶ 156
        } else {
157
            stored
158
        };
159
        if body.len() != e.raw_len as usize || hash_object(e.ty, &body) != *r {
75167672 Andy McCall 160
            return Err(StoreError::Corrupt(
161
                *r,
162
                "packed object does not match its reference".into(),
163
            ));
d02e545d Andy McCall ↶ 164
        }
165
        Ok((e.ty, body))
166
    }
167
}
168
169
fn type_index(ty: ObjectType) -> u8 {
170
    ObjectType::ALL.iter().position(|t| *t == ty).unwrap() as u8
171
}
172
173
/// What `pack_loose` did.
174
#[derive(Debug, Clone, Default, PartialEq, Eq)]
175
pub struct PackReport {
176
    pub objects: usize,
177
    pub raw_bytes: u64,
178
    pub packed_bytes: u64,
179
}
180
135212de Andy McCall ↶ 181
/// Packs tolerated before `merge_packs` folds them into one (spec §13.4).
182
pub const MAX_PACKS: usize = 16;
183
ba160f60 Andy McCall ↶ 184
/// What `prune` did.
185
#[derive(Debug, Clone, Default, PartialEq, Eq)]
186
pub struct PruneReport {
187
    pub objects: usize,
188
    pub bytes: u64,
189
    /// Unreachable objects left alone because they are younger than the grace period.
190
    pub deferred: usize,
191
}
192
ff0af1ee unknown 193
#[derive(Debug, thiserror::Error)]
194
pub enum StoreError {
195
    #[error("io: {0}")]
196
    Io(#[from] io::Error),
197
    #[error("corrupt object {0}: {1}")]
198
    Corrupt(ObjectRef, String),
199
    #[error("object {0}")]
200
    Object(#[from] ObjectError),
201
    #[error("object {0} is a {1}, expected {2}")]
202
    WrongType(ObjectRef, &'static str, &'static str),
203
    #[error("object {0} not found")]
204
    NotFound(ObjectRef),
6f8aeb99 Andy McCall ↶ 205
    #[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)")]
a9f686ed Andy McCall ↶ 206
    TooLarge {
207
        kind: &'static str,
208
        size: usize,
209
        limit: u32,
210
    },
ff0af1ee unknown 211
}
212
213
pub type Result<T> = std::result::Result<T, StoreError>;
214
d02e545d Andy McCall ↶ 215
#[derive(Debug)]
ff0af1ee unknown 216
pub struct Store {
217
    dir: PathBuf,
d02e545d Andy McCall ↶ 218
    packs: RefCell<Vec<PackIndex>>,
8f10a07a Andy McCall ↶ 219
    /// An open write batch (spec §13.4): objects go here instead of to loose
220
    /// files, and are written out with one `fsync` when the batch finishes.
221
    pending: RefCell<Option<Pending>>,
222
    batch_depth: std::cell::Cell<u32>,
ce14f0d6 Andy McCall ↶ 223
    /// Change metadata already read, from the `metas` cache file.
224
    metas: RefCell<Option<crate::fasthash::FastMap<ObjectRef, crate::state::ChangeMeta>>>,
8f10a07a Andy McCall ↶ 225
}
226
227
/// Objects written since `begin_batch`. Small batches stay in memory and
228
/// become loose objects; once `SPILL_OBJECTS` or `SPILL_BYTES` is exceeded
229
/// the batch streams into a pack file instead.
230
#[derive(Debug)]
231
struct Pending {
232
    buffered: Vec<(ObjectType, Vec<u8>)>,
233
    buffered_bytes: usize,
234
    lookup: std::collections::HashMap<ObjectRef, PendingEntry>,
235
    pack: Option<PendingPack>,
236
    report: PackReport,
ff0af1ee unknown 237
}
238
8f10a07a Andy McCall ↶ 239
#[derive(Debug, Clone, Copy)]
240
enum PendingEntry {
241
    Buffered(usize),
242
    Packed(PackEntry),
243
}
244
245
#[derive(Debug)]
246
struct PendingPack {
247
    id: String,
248
    tmp_path: PathBuf,
249
    writer: io::BufWriter<fs::File>,
250
    offset: u64,
251
    index: Vec<(ObjectRef, PackEntry)>,
252
}
253
254
const SPILL_OBJECTS: usize = 256;
255
const SPILL_BYTES: usize = 8 << 20;
256
257
/// Finishes the enclosing batch when dropped; call `finish` to see errors.
258
pub struct Batch<'a> {
259
    store: &'a Store,
260
    done: bool,
261
}
262
263
impl Batch<'_> {
264
    pub fn finish(mut self) -> Result<()> {
265
        self.done = true;
266
        self.store.finish_batch()
267
    }
268
}
269
270
impl Drop for Batch<'_> {
271
    fn drop(&mut self) {
272
        if !self.done {
273
            let _ = self.store.finish_batch();
274
        }
275
    }
276
}
277
192fad1e Andy McCall ↶ 278
/// Append one stored object to a pending pack and index it.
279
fn write_pack_entry(
280
    pack: &mut PendingPack,
281
    r: &ObjectRef,
282
    ty: ObjectType,
283
    raw_len: usize,
284
    flags: u8,
285
    stored: &[u8],
286
    report: &mut PackReport,
287
) -> Result<PackEntry> {
288
    let header_len = 1 + 1 + 4 + 4 + 32;
289
    pack.writer.write_all(&[type_index(ty), flags])?;
290
    pack.writer
291
        .write_all(&(stored.len() as u32).to_le_bytes())?;
292
    pack.writer.write_all(&(raw_len as u32).to_le_bytes())?;
293
    pack.writer.write_all(&r.0)?;
294
    pack.writer.write_all(stored)?;
295
    let entry = PackEntry {
296
        offset: pack.offset + header_len,
297
        stored_len: stored.len() as u32,
298
        raw_len: raw_len as u32,
299
        ty,
300
        flags,
301
    };
302
    pack.offset += header_len + stored.len() as u64;
303
    pack.index.push((*r, entry));
304
    report.objects += 1;
305
    report.raw_bytes += raw_len as u64;
306
    report.packed_bytes += stored.len() as u64;
307
    Ok(entry)
308
}
309
ff0af1ee unknown 310
impl Store {
311
    pub fn open(dir: impl Into<PathBuf>) -> io::Result<Store> {
312
        let dir = dir.into();
313
        fs::create_dir_all(dir.join("loose"))?;
75167672 Andy McCall 314
        let store = Store {
315
            dir,
316
            packs: RefCell::new(Vec::new()),
317
            pending: RefCell::new(None),
318
            batch_depth: std::cell::Cell::new(0),
ce14f0d6 Andy McCall ↶ 319
            metas: RefCell::new(None),
75167672 Andy McCall 320
        };
321
        store
322
            .reload_packs()
323
            .map_err(|e| io::Error::other(e.to_string()))?;
d02e545d Andy McCall ↶ 324
        Ok(store)
325
    }
326
327
    fn reload_packs(&self) -> Result<()> {
328
        let mut packs = Vec::new();
329
        let mut names: Vec<PathBuf> = fs::read_dir(&self.dir)?
330
            .filter_map(|e| e.ok().map(|e| e.path()))
75167672 Andy McCall 331
            .filter(|p| {
332
                p.extension().is_some_and(|x| x == "idx")
333
                    && p.file_name()
334
                        .is_some_and(|n| n.to_string_lossy().starts_with("pack-"))
335
            })
d02e545d Andy McCall ↶ 336
            .collect();
337
        names.sort();
338
        for idx in names {
339
            let pack = idx.with_extension("pack");
340
            if pack.is_file() {
341
                packs.push(PackIndex::load(&idx, &pack)?);
342
            }
343
        }
344
        *self.packs.borrow_mut() = packs;
345
        Ok(())
346
    }
347
348
    pub fn pack_count(&self) -> usize {
349
        self.packs.borrow().len()
350
    }
351
352
    fn find_packed(&self, r: &ObjectRef) -> Option<(usize, PackEntry)> {
75167672 Andy McCall 353
        self.packs
354
            .borrow()
355
            .iter()
356
            .enumerate()
357
            .find_map(|(i, p)| p.find(r).map(|e| (i, e)))
d02e545d Andy McCall ↶ 358
    }
359
360
    /// Loose objects on disk, with their paths.
361
    fn loose_objects(&self) -> io::Result<Vec<(ObjectRef, PathBuf)>> {
362
        let mut out = Vec::new();
363
        let loose = self.dir.join("loose");
364
        for d in fs::read_dir(&loose)? {
365
            let d = d?;
366
            if !d.file_type()?.is_dir() {
367
                continue;
368
            }
369
            let prefix = d.file_name().to_string_lossy().to_string();
370
            for f in fs::read_dir(d.path())? {
371
                let f = f?;
372
                let name = f.file_name().to_string_lossy().to_string();
373
                if let Some(r) = ObjectRef::from_hex(&format!("{prefix}{name}")) {
374
                    out.push((r, f.path()));
375
                }
376
            }
377
        }
378
        out.sort();
379
        Ok(out)
380
    }
381
382
    pub fn loose_count(&self) -> io::Result<usize> {
383
        Ok(self.loose_objects()?.len())
384
    }
385
386
    /// Move every loose object into a new pack (spec §13.4). Loose files are
387
    /// deleted only after the pack and its index are durable.
388
    pub fn pack_loose(&self) -> Result<PackReport> {
389
        let loose = self.loose_objects()?;
75167672 Andy McCall 390
        self.write_pack(
391
            &loose.iter().map(|(r, _)| *r).collect::<Vec<_>>(),
392
            &loose.iter().map(|(_, p)| p.clone()).collect::<Vec<_>>(),
393
            &[],
da652d63 Andy McCall ↶ 394
            ZSTD_LEVEL,
75167672 Andy McCall 395
        )
da652d63 Andy McCall ↶ 396
    }
397
398
    /// Fold every pack and loose object into one pack compressed at
399
    /// `level` (spec §13.4): what an explicit `oxy gc` does, as `git gc`
400
    /// repacks. Returns the report and how many packs were folded in.
401
    pub fn repack_all(&self, level: i32) -> Result<(PackReport, usize)> {
402
        let loose = self.loose_objects()?;
403
        let (mut refs, paths): (Vec<ObjectRef>, Vec<PathBuf>) = {
404
            let packs = self.packs.borrow();
405
            (
406
                packs.iter().flat_map(|p| p.refs.iter().copied()).collect(),
407
                packs.iter().map(|p| p.path.clone()).collect(),
408
            )
409
        };
410
        if loose.is_empty() && paths.len() <= 1 && level == ZSTD_LEVEL {
411
            return Ok((PackReport::default(), 0));
412
        }
413
        refs.extend(loose.iter().map(|(r, _)| *r));
414
        let loose_paths: Vec<PathBuf> = loose.into_iter().map(|(_, p)| p).collect();
415
        let n = paths.len();
416
        let report = self.write_pack(&refs, &loose_paths, &paths, level)?;
417
        Ok((report, n))
ba160f60 Andy McCall ↶ 418
    }
419
135212de Andy McCall ↶ 420
    /// Fold every pack into one when there are more than `MAX_PACKS`
421
    /// (spec §13.4). Returns how many packs were merged.
422
    pub fn merge_packs(&self) -> Result<usize> {
423
        let (refs, paths): (Vec<ObjectRef>, Vec<PathBuf>) = {
424
            let packs = self.packs.borrow();
425
            if packs.len() <= MAX_PACKS {
426
                return Ok(0);
427
            }
75167672 Andy McCall 428
            (
429
                packs.iter().flat_map(|p| p.refs.iter().copied()).collect(),
430
                packs.iter().map(|p| p.path.clone()).collect(),
431
            )
135212de Andy McCall ↶ 432
        };
433
        let n = paths.len();
da652d63 Andy McCall ↶ 434
        self.write_pack(&refs, &[], &paths, ZSTD_LEVEL)?;
135212de Andy McCall ↶ 435
        Ok(n)
436
    }
437
ba160f60 Andy McCall ↶ 438
    /// Delete every object not in `keep` (spec §21). Loose objects and packs
439
    /// younger than `grace` are left alone, since a concurrent writer may not
440
    /// have recorded its references yet. Packs holding unreachable objects
441
    /// are rewritten without them. Signature sidecars and seal-index entries
442
    /// of deleted objects go too.
75167672 Andy McCall 443
    pub fn prune(
444
        &self,
445
        keep: &std::collections::HashSet<ObjectRef>,
446
        grace: std::time::Duration,
447
    ) -> Result<PruneReport> {
ba160f60 Andy McCall ↶ 448
        let now = std::time::SystemTime::now();
75167672 Andy McCall 449
        let old_enough = |m: &fs::Metadata| {
450
            m.modified()
451
                .map(|t| now.duration_since(t).unwrap_or_default() >= grace)
452
                .unwrap_or(true)
453
        };
ba160f60 Andy McCall ↶ 454
        let mut report = PruneReport::default();
455
        // Loose objects.
456
        for (r, path) in self.loose_objects()? {
457
            if keep.contains(&r) {
458
                continue;
459
            }
460
            let meta = fs::metadata(&path)?;
461
            if !old_enough(&meta) {
462
                report.deferred += 1;
463
                continue;
464
            }
465
            report.bytes += meta.len();
466
            report.objects += 1;
b30a880f Andy McCall ↶ 467
            // The fan-out directory stays: recreating it costs a sync of
468
            // `loose` on the next write.
ba160f60 Andy McCall ↶ 469
            fs::remove_file(&path)?;
470
            self.remove_sidecars(&r)?;
471
        }
472
        // Packs: rewrite those that hold unreachable objects.
473
        let mut keep_refs: Vec<ObjectRef> = Vec::new();
474
        let mut old_paths: Vec<PathBuf> = Vec::new();
475
        let mut rewrite = false;
476
        for p in self.packs.borrow().iter() {
477
            let meta = fs::metadata(&p.path)?;
478
            let unreachable = p.refs.iter().filter(|r| !keep.contains(r)).count();
479
            if unreachable == 0 {
480
                continue;
481
            }
482
            if !old_enough(&meta) {
483
                report.deferred += unreachable;
484
                continue;
485
            }
486
            rewrite = true;
487
            old_paths.push(p.path.clone());
488
            for (r, e) in p.refs.iter().zip(p.entries.iter()) {
489
                if keep.contains(r) {
490
                    keep_refs.push(*r);
491
                } else {
492
                    report.objects += 1;
493
                    report.bytes += e.stored_len as u64;
494
                }
495
            }
496
        }
497
        if rewrite {
498
            let dropped: Vec<ObjectRef> = {
499
                let packs = self.packs.borrow();
75167672 Andy McCall 500
                packs
501
                    .iter()
502
                    .filter(|p| old_paths.contains(&p.path))
503
                    .flat_map(|p| p.refs.iter().copied())
504
                    .filter(|r| !keep.contains(r))
505
                    .collect()
ba160f60 Andy McCall ↶ 506
            };
da652d63 Andy McCall ↶ 507
            self.write_pack(&keep_refs, &[], &old_paths, ZSTD_LEVEL)?;
ba160f60 Andy McCall ↶ 508
            for r in dropped {
509
                self.remove_sidecars(&r)?;
510
            }
511
        }
512
        Ok(report)
513
    }
514
515
    fn remove_sidecars(&self, r: &ObjectRef) -> Result<()> {
fc1a466a Andy McCall ↶ 516
        let sidecars = [
517
            self.sig_path(r),
518
            self.index_path("seals", r),
519
            self.index_path("receipts", r),
520
        ];
521
        for path in sidecars {
ba160f60 Andy McCall ↶ 522
            match fs::remove_file(&path) {
523
                Ok(()) => {
524
                    let _ = fs::remove_dir(path.parent().unwrap());
525
                }
526
                Err(e) if e.kind() == io::ErrorKind::NotFound => {}
527
                Err(e) => return Err(e.into()),
528
            }
529
        }
530
        Ok(())
531
    }
532
533
    /// Write `refs` into a new pack, then delete `loose_paths` and `old_packs`.
8f10a07a Andy McCall ↶ 534
    /// Write the index for a complete, synced temporary pack file and move
535
    /// both into place; `index` must be sorted by ObjectRef.
75167672 Andy McCall 536
    fn finish_pack_files(
537
        &self,
538
        id: &str,
539
        tmp_pack: &Path,
540
        index: &[(ObjectRef, PackEntry)],
541
    ) -> Result<()> {
8f10a07a Andy McCall ↶ 542
        let pack_path = self.dir.join(format!("pack-{id}.pack"));
543
        let idx_path = self.dir.join(format!("pack-{id}.idx"));
544
        let tmp_idx = self.dir.join(format!("tmp-{id}.idx"));
545
        {
546
            let mut f = io::BufWriter::new(fs::File::create(&tmp_idx)?);
547
            f.write_all(IDX_MAGIC)?;
548
            f.write_all(&(index.len() as u32).to_le_bytes())?;
549
            for (r, e) in index {
550
                f.write_all(&r.0)?;
551
                f.write_all(&e.offset.to_le_bytes())?;
552
                f.write_all(&e.stored_len.to_le_bytes())?;
553
                f.write_all(&e.raw_len.to_le_bytes())?;
554
                f.write_all(&[type_index(e.ty), e.flags])?;
555
            }
556
            f.flush()?;
557
            f.get_ref().sync_all()?;
558
        }
4dcb8c09 Andy McCall ↶ 559
        // The pack first: an index never names a pack that is not there.
560
        crate::durable::rename(tmp_pack, &pack_path)?;
561
        crate::durable::rename(&tmp_idx, &idx_path)?;
8f10a07a Andy McCall ↶ 562
        self.reload_packs()
563
    }
564
75167672 Andy McCall 565
    fn write_pack(
566
        &self,
567
        refs: &[ObjectRef],
568
        loose_paths: &[PathBuf],
569
        old_packs: &[PathBuf],
da652d63 Andy McCall ↶ 570
        level: i32,
75167672 Andy McCall 571
    ) -> Result<PackReport> {
ba160f60 Andy McCall ↶ 572
        if refs.is_empty() {
573
            for p in old_packs {
a17fc622 Andy McCall ↶ 574
                remove_pack_files(p)?;
ba160f60 Andy McCall ↶ 575
            }
576
            self.reload_packs()?;
d02e545d Andy McCall ↶ 577
            return Ok(PackReport::default());
578
        }
ba160f60 Andy McCall ↶ 579
        let mut refs: Vec<ObjectRef> = refs.to_vec();
580
        refs.sort();
581
        refs.dedup();
d02e545d Andy McCall ↶ 582
        let mut nonce = [0u8; 8];
583
        rand::rngs::OsRng.fill_bytes(&mut nonce);
584
        let id = hex::encode(nonce);
585
        let tmp_pack = self.dir.join(format!("tmp-{id}.pack"));
586
        let mut report = PackReport::default();
ba160f60 Andy McCall ↶ 587
        let mut index: Vec<(ObjectRef, PackEntry)> = Vec::with_capacity(refs.len());
a17fc622 Andy McCall ↶ 588
        // A full repack of enough objects trains a dictionary for them.
589
        let dict = if level == REPACK_ZSTD_LEVEL && refs.len() >= DICT_MIN_OBJECTS {
590
            let _t = crate::timing::phase("pack: train a dictionary");
591
            self.train_dictionary(&refs)
592
        } else {
593
            None
594
        };
595
        let encoder = dict
596
            .as_ref()
597
            .map(|d| zstd::dict::EncoderDictionary::copy(d, level));
d02e545d Andy McCall ↶ 598
        {
599
            let mut f = io::BufWriter::new(fs::File::create(&tmp_pack)?);
600
            f.write_all(PACK_MAGIC)?;
601
            let mut offset = PACK_MAGIC.len() as u64;
da652d63 Andy McCall ↶ 602
            // Read a batch (the store is not shared between threads),
603
            // compress it on every core, write it in order.
604
            let mut batches = refs.chunks(1024);
605
            let mut compressed_batch = Vec::new();
606
            loop {
607
                if compressed_batch.is_empty() {
608
                    let Some(batch) = batches.next() else { break };
609
                    let mut read: Vec<(ObjectRef, ObjectType, Vec<u8>, bool)> =
610
                        Vec::with_capacity(batch.len());
611
                    for r in batch {
612
                        let (ty, body) = self.get(r)?;
613
                        read.push((*r, ty, body, self.is_nocompress(r)));
75167672 Andy McCall 614
                    }
da652d63 Andy McCall ↶ 615
                    use rayon::prelude::*;
616
                    compressed_batch = read
617
                        .into_par_iter()
a17fc622 Andy McCall ↶ 618
                        .map_init(
619
                            || {
620
                                encoder.as_ref().and_then(|e| {
621
                                    zstd::bulk::Compressor::with_prepared_dictionary(e).ok()
622
                                })
623
                            },
624
                            |with_dict, (r, ty, body, raw)| {
625
                                let packed = if raw {
626
                                    None
627
                                } else {
628
                                    // Small objects with the dictionary;
629
                                    // a large one has context enough.
630
                                    let dicted = with_dict
631
                                        .as_mut()
632
                                        .filter(|_| body.len() <= DICT_OBJECT_MAX)
633
                                        .and_then(|c| c.compress(&body).ok())
634
                                        .map(|c| (FLAG_ZSTD | FLAG_DICT, c));
635
                                    dicted
636
                                        .or_else(|| {
637
                                            zstd::bulk::compress(&body, level)
638
                                                .ok()
639
                                                .map(|c| (FLAG_ZSTD, c))
640
                                        })
641
                                        .filter(|(_, c)| c.len() < body.len())
642
                                };
643
                                (r, ty, body, packed)
644
                            },
645
                        )
da652d63 Andy McCall ↶ 646
                        .collect::<Vec<_>>();
647
                    compressed_batch.reverse();
648
                }
649
                let Some((r, ty, body, packed)) = compressed_batch.pop() else {
650
                    continue;
651
                };
652
                let r = &r;
653
                let (flags, stored) = match packed {
a17fc622 Andy McCall ↶ 654
                    Some(c) => c,
da652d63 Andy McCall ↶ 655
                    None => (0, body.clone()),
cdb5beba Andy McCall ↶ 656
                };
d02e545d Andy McCall ↶ 657
                let header_len = 1 + 1 + 4 + 4 + 32;
658
                f.write_all(&[type_index(ty), flags])?;
659
                f.write_all(&(stored.len() as u32).to_le_bytes())?;
660
                f.write_all(&(body.len() as u32).to_le_bytes())?;
661
                f.write_all(&r.0)?;
662
                f.write_all(&stored)?;
75167672 Andy McCall 663
                index.push((
664
                    *r,
665
                    PackEntry {
666
                        offset: offset + header_len,
667
                        stored_len: stored.len() as u32,
668
                        raw_len: body.len() as u32,
669
                        ty,
670
                        flags,
671
                    },
672
                ));
d02e545d Andy McCall ↶ 673
                offset += header_len + stored.len() as u64;
674
                report.objects += 1;
675
                report.raw_bytes += body.len() as u64;
676
                report.packed_bytes += stored.len() as u64;
677
            }
678
            f.flush()?;
679
            f.get_ref().sync_all()?;
a17fc622 Andy McCall ↶ 680
        }
681
        // The dictionary before the pack that needs it becomes visible.
682
        if let Some(d) = &dict {
683
            crate::durable::write_atomic(&self.dir.join(format!("pack-{id}.dict")), d)?;
d02e545d Andy McCall ↶ 684
        }
8f10a07a Andy McCall ↶ 685
        self.finish_pack_files(&id, &tmp_pack, &index)?;
ba160f60 Andy McCall ↶ 686
        for p in old_packs {
a17fc622 Andy McCall ↶ 687
            remove_pack_files(p)?;
ba160f60 Andy McCall ↶ 688
        }
d02e545d Andy McCall ↶ 689
        self.reload_packs()?;
b30a880f Andy McCall ↶ 690
        // Fan-out directories stay, emptied: see `ensure_fanouts`.
ba160f60 Andy McCall ↶ 691
        for path in loose_paths {
d02e545d Andy McCall ↶ 692
            fs::remove_file(path)?;
693
        }
694
        Ok(report)
ff0af1ee unknown 695
    }
696
a17fc622 Andy McCall ↶ 697
    /// A dictionary for these objects, trained on a sample of them (at
698
    /// most 10 000, 32 MB); `None` when training fails.
699
    fn train_dictionary(&self, refs: &[ObjectRef]) -> Option<Vec<u8>> {
700
        let step = (refs.len() / 10_000).max(1);
701
        let mut samples: Vec<Vec<u8>> = Vec::new();
702
        let mut total = 0usize;
703
        for r in refs.iter().step_by(step) {
704
            if total > 32 << 20 {
705
                break;
706
            }
707
            let Ok((ty, body)) = self.get(r) else {
708
                continue;
709
            };
710
            if matches!(
711
                ty,
712
                ObjectType::Chunk | ObjectType::Blob | ObjectType::Change
713
            ) && !body.is_empty()
714
                && body.len() < 1 << 20
715
            {
716
                total += body.len();
717
                samples.push(body);
718
            }
719
        }
720
        zstd::dict::from_samples(&samples, DICT_BYTES).ok()
721
    }
722
ff0af1ee unknown 723
    pub fn dir(&self) -> &Path {
724
        &self.dir
725
    }
726
387335ff Andy McCall ↶ 727
    pub(crate) fn path_for(&self, r: &ObjectRef) -> PathBuf {
ff0af1ee unknown 728
        let h = r.to_hex();
729
        self.dir.join("loose").join(&h[..2]).join(&h[2..])
730
    }
731
732
    pub fn has(&self, r: &ObjectRef) -> bool {
75167672 Andy McCall 733
        self.path_for(r).is_file()
734
            || self.find_packed(r).is_some()
735
            || self
736
                .pending
737
                .borrow()
738
                .as_ref()
739
                .is_some_and(|p| p.lookup.contains_key(r))
ff0af1ee unknown 740
    }
741
8f10a07a Andy McCall ↶ 742
    /// Start a write batch: until the matching `finish`, `put` collects
743
    /// objects and writes them out together, with one `fsync` for a large
744
    /// batch (a pack) instead of one per object. Batches nest; only the
745
    /// outermost does the work. Objects are readable while pending.
746
    pub fn begin_batch(&self) -> Result<Batch<'_>> {
747
        let depth = self.batch_depth.get();
748
        if depth == 0 {
75167672 Andy McCall 749
            *self.pending.borrow_mut() = Some(Pending {
750
                buffered: vec![],
751
                buffered_bytes: 0,
752
                lookup: Default::default(),
753
                pack: None,
754
                report: PackReport::default(),
755
            });
8f10a07a Andy McCall ↶ 756
        }
757
        self.batch_depth.set(depth + 1);
75167672 Andy McCall 758
        Ok(Batch {
759
            store: self,
760
            done: false,
761
        })
8f10a07a Andy McCall ↶ 762
    }
763
764
    fn finish_batch(&self) -> Result<()> {
765
        let depth = self.batch_depth.get();
766
        if depth == 0 {
767
            return Ok(());
768
        }
769
        self.batch_depth.set(depth - 1);
770
        if depth > 1 {
771
            return Ok(());
772
        }
75167672 Andy McCall 773
        let Some(pending) = self.pending.borrow_mut().take() else {
774
            return Ok(());
775
        };
4dcb8c09 Andy McCall ↶ 776
        self.write_pending(pending)
777
    }
778
779
    /// Make every object written so far durable, even inside an open
780
    /// batch, which stays open. Call before a reference that names them
781
    /// is moved (spec §13.6): the operation head, a bookmark pointer.
782
    pub fn flush(&self) -> Result<()> {
783
        let Some(pending) = self.pending.borrow_mut().take() else {
784
            return Ok(());
785
        };
786
        self.write_pending(pending)?;
787
        *self.pending.borrow_mut() = Some(Pending {
788
            buffered: vec![],
789
            buffered_bytes: 0,
790
            lookup: Default::default(),
791
            pack: None,
792
            report: PackReport::default(),
793
        });
794
        Ok(())
795
    }
796
797
    fn write_pending(&self, pending: Pending) -> Result<()> {
8f10a07a Andy McCall ↶ 798
        match pending.pack {
799
            None => {
b30a880f Andy McCall ↶ 800
                let objects: Vec<(ObjectType, &[u8])> = pending
801
                    .buffered
802
                    .iter()
803
                    .map(|(ty, body)| (*ty, body.as_slice()))
804
                    .collect();
805
                self.put_loose_many(&objects)
8f10a07a Andy McCall ↶ 806
            }
807
            Some(mut pack) => {
808
                pack.writer.flush()?;
809
                pack.writer.get_ref().sync_all()?;
810
                drop(pack.writer);
811
                pack.index.sort_by_key(|(r, _)| *r);
812
                self.finish_pack_files(&pack.id, &pack.tmp_path, &pack.index)?;
813
                Ok(())
814
            }
815
        }
816
    }
817
818
    /// Whether a batch is open and holds `r`.
819
    fn pending_get(&self, r: &ObjectRef) -> Option<Result<(ObjectType, Vec<u8>)>> {
820
        let mut guard = self.pending.borrow_mut();
821
        let pending = guard.as_mut()?;
822
        let entry = *pending.lookup.get(r)?;
823
        Some(match entry {
824
            PendingEntry::Buffered(i) => {
825
                let (ty, body) = &pending.buffered[i];
826
                Ok((*ty, body.clone()))
827
            }
828
            PendingEntry::Packed(e) => {
75167672 Andy McCall 829
                let Some(pack) = pending.pack.as_mut() else {
830
                    return Some(Err(StoreError::NotFound(*r)));
831
                };
8f10a07a Andy McCall ↶ 832
                match pack.writer.flush() {
75167672 Andy McCall 833
                    Ok(()) => PackIndex {
834
                        path: pack.tmp_path.clone(),
835
                        refs: vec![],
836
                        entries: vec![],
a17fc622 Andy McCall ↶ 837
                        dict: None,
75167672 Andy McCall 838
                    }
839
                    .read(r, e),
8f10a07a Andy McCall ↶ 840
                    Err(err) => Err(err.into()),
841
                }
842
            }
843
        })
844
    }
845
846
    /// Append one object to the pending pack file.
75167672 Andy McCall 847
    fn pending_pack_write(
848
        &self,
849
        pack: &mut PendingPack,
850
        r: &ObjectRef,
851
        ty: ObjectType,
852
        body: &[u8],
853
        report: &mut PackReport,
854
    ) -> Result<PackEntry> {
8f10a07a Andy McCall ↶ 855
        let (flags, stored) = if self.is_nocompress(r) {
856
            (0, body.to_vec())
857
        } else {
75167672 Andy McCall 858
            let compressed = zstd::bulk::compress(body, ZSTD_LEVEL)
859
                .map_err(|e| StoreError::Corrupt(*r, format!("zstd: {e}")))?;
860
            if compressed.len() < body.len() {
861
                (FLAG_ZSTD, compressed)
862
            } else {
863
                (0, body.to_vec())
864
            }
8f10a07a Andy McCall ↶ 865
        };
192fad1e Andy McCall ↶ 866
        write_pack_entry(pack, r, ty, body.len(), flags, &stored, report)
8f10a07a Andy McCall ↶ 867
    }
868
869
    fn put_pending(&self, r: ObjectRef, ty: ObjectType, body: &[u8]) -> Result<()> {
870
        let mut guard = self.pending.borrow_mut();
871
        let pending = guard.as_mut().expect("batch open");
872
        if pending.lookup.contains_key(&r) {
873
            return Ok(());
874
        }
75167672 Andy McCall 875
        if pending.pack.is_none()
876
            && pending.buffered.len() < SPILL_OBJECTS
877
            && pending.buffered_bytes + body.len() <= SPILL_BYTES
878
        {
8f10a07a Andy McCall ↶ 879
            pending.buffered.push((ty, body.to_vec()));
880
            pending.buffered_bytes += body.len();
75167672 Andy McCall 881
            pending
882
                .lookup
883
                .insert(r, PendingEntry::Buffered(pending.buffered.len() - 1));
8f10a07a Andy McCall ↶ 884
            return Ok(());
885
        }
192fad1e Andy McCall ↶ 886
        self.ensure_pack(pending)?;
887
        let mut report = std::mem::take(&mut pending.report);
888
        let pack = pending.pack.as_mut().unwrap();
889
        let e = self.pending_pack_write(pack, &r, ty, body, &mut report)?;
890
        pending.report = report;
891
        pending.lookup.insert(r, PendingEntry::Packed(e));
892
        Ok(())
893
    }
894
895
    /// Open the pending batch's pack, moving the objects buffered so far
896
    /// into it (a spill).
897
    fn ensure_pack(&self, pending: &mut Pending) -> Result<()> {
8f10a07a Andy McCall ↶ 898
        if pending.pack.is_none() {
899
            // Spill: open the pack and move the buffered objects into it.
900
            let mut nonce = [0u8; 8];
901
            rand::rngs::OsRng.fill_bytes(&mut nonce);
902
            let id = hex::encode(nonce);
903
            let tmp_path = self.dir.join(format!("tmp-{id}.pack"));
904
            let mut writer = io::BufWriter::new(fs::File::create(&tmp_path)?);
905
            writer.write_all(PACK_MAGIC)?;
75167672 Andy McCall 906
            let mut pack = PendingPack {
907
                id,
908
                tmp_path,
909
                writer,
910
                offset: PACK_MAGIC.len() as u64,
911
                index: vec![],
912
            };
8f10a07a Andy McCall ↶ 913
            let buffered = std::mem::take(&mut pending.buffered);
914
            pending.buffered_bytes = 0;
915
            let mut report = std::mem::take(&mut pending.report);
916
            for (bty, bbody) in &buffered {
917
                let br = hash_object(*bty, bbody);
918
                let e = self.pending_pack_write(&mut pack, &br, *bty, bbody, &mut report)?;
919
                pending.lookup.insert(br, PendingEntry::Packed(e));
920
            }
921
            pending.report = report;
922
            pending.pack = Some(pack);
192fad1e Andy McCall ↶ 923
        }
924
        Ok(())
925
    }
926
927
    /// Store many objects at once. Inside a large write batch they are
928
    /// compressed on every core and appended in order; otherwise each is
929
    /// stored as [`Store::put`] does.
930
    pub fn put_many(&self, objects: &[(ObjectType, &[u8])]) -> Result<Vec<ObjectRef>> {
931
        use rayon::prelude::*;
932
        let total: usize = objects.iter().map(|(_, b)| b.len()).sum();
933
        if self.pending.borrow().is_none() || total < SPILL_BYTES {
934
            return objects.iter().map(|(ty, b)| self.put(*ty, b)).collect();
935
        }
936
        for (ty, b) in objects {
937
            if b.len() > MAX_OBJECT_LEN as usize {
938
                return Err(StoreError::TooLarge {
939
                    kind: ty.name(),
940
                    size: b.len(),
941
                    limit: MAX_OBJECT_LEN,
942
                });
943
            }
8f10a07a Andy McCall ↶ 944
        }
192fad1e Andy McCall ↶ 945
        let refs: Vec<ObjectRef> = objects
946
            .par_iter()
947
            .map(|(ty, b)| hash_object(*ty, b))
948
            .collect();
949
        // What is new: not stored already, not earlier in this call.
950
        let mut seen: std::collections::HashSet<ObjectRef> = std::collections::HashSet::new();
951
        let mut todo: Vec<usize> = Vec::new();
952
        for (i, r) in refs.iter().enumerate() {
953
            let known = self
954
                .pending
955
                .borrow()
956
                .as_ref()
957
                .is_some_and(|p| p.lookup.contains_key(r));
958
            if seen.insert(*r)
959
                && !known
960
                && !self.path_for(r).is_file()
961
                && self.find_packed(r).is_none()
962
            {
963
                todo.push(i);
964
            }
965
        }
966
        let raw: std::collections::HashSet<ObjectRef> =
967
            match fs::read_dir(self.dir.join("nocompress")) {
968
                Ok(rd) => rd
969
                    .flatten()
970
                    .filter_map(|e| ObjectRef::from_hex(&e.file_name().to_string_lossy()))
971
                    .collect(),
972
                Err(_) => std::collections::HashSet::new(),
973
            };
974
        let packed: Vec<(u8, Vec<u8>)> = todo
975
            .par_iter()
976
            .map(|i| {
977
                let body = objects[*i].1;
978
                if raw.contains(&refs[*i]) {
979
                    return (0, body.to_vec());
980
                }
981
                match zstd::bulk::compress(body, ZSTD_LEVEL) {
982
                    Ok(c) if c.len() < body.len() => (FLAG_ZSTD, c),
983
                    _ => (0, body.to_vec()),
984
                }
985
            })
986
            .collect();
987
        let mut guard = self.pending.borrow_mut();
988
        let pending = guard.as_mut().expect("batch open");
989
        self.ensure_pack(pending)?;
8f10a07a Andy McCall ↶ 990
        let mut report = std::mem::take(&mut pending.report);
991
        let pack = pending.pack.as_mut().unwrap();
192fad1e Andy McCall ↶ 992
        for (i, (flags, stored)) in todo.iter().zip(packed) {
993
            let (ty, body) = objects[*i];
994
            let e = write_pack_entry(pack, &refs[*i], ty, body.len(), flags, &stored, &mut report)?;
995
            pending.lookup.insert(refs[*i], PendingEntry::Packed(e));
996
        }
8f10a07a Andy McCall ↶ 997
        pending.report = report;
192fad1e Andy McCall ↶ 998
        Ok(refs)
8f10a07a Andy McCall ↶ 999
    }
1000
ff0af1ee unknown 1001
    /// Store `body` as an object of type `ty`; idempotent.
1002
    pub fn put(&self, ty: ObjectType, body: &[u8]) -> Result<ObjectRef> {
a9f686ed Andy McCall ↶ 1003
        // Refuse what the store could not read back (spec §13.2), before
1004
        // anything is written: a pack must never hold an entry its index
1005
        // would reject.
1006
        if body.len() > MAX_OBJECT_LEN as usize {
1007
            return Err(StoreError::TooLarge {
1008
                kind: ty.name(),
1009
                size: body.len(),
1010
                limit: MAX_OBJECT_LEN,
1011
            });
1012
        }
ff0af1ee unknown 1013
        let r = hash_object(ty, body);
1014
        let path = self.path_for(&r);
d02e545d Andy McCall ↶ 1015
        if path.is_file() || self.find_packed(&r).is_some() {
ff0af1ee unknown 1016
            return Ok(r);
1017
        }
8f10a07a Andy McCall ↶ 1018
        if self.pending.borrow().is_some() {
1019
            self.put_pending(r, ty, body)?;
1020
            return Ok(r);
1021
        }
1022
        self.put_loose(ty, body)?;
1023
        Ok(r)
1024
    }
1025
1026
    /// Write one loose object atomically with its own `fsync`.
1027
    fn put_loose(&self, ty: ObjectType, body: &[u8]) -> Result<ObjectRef> {
1028
        let r = hash_object(ty, body);
1029
        let path = self.path_for(&r);
1030
        if path.is_file() {
1031
            return Ok(r);
1032
        }
b30a880f Andy McCall ↶ 1033
        if !path.parent().unwrap().is_dir() {
1034
            self.ensure_fanouts()?;
4dcb8c09 Andy McCall ↶ 1035
        }
ff0af1ee unknown 1036
        let mut nonce = [0u8; 8];
1037
        rand::rngs::OsRng.fill_bytes(&mut nonce);
75167672 Andy McCall 1038
        let tmp = self
1039
            .dir
1040
            .join("loose")
1041
            .join(format!("tmp-{}", hex::encode(nonce)));
4dcb8c09 Andy McCall ↶ 1042
        crate::durable::step("write loose object")?;
ff0af1ee unknown 1043
        {
1044
            let mut f = fs::File::create(&tmp)?;
1045
            f.write_all(&header(ty, body.len()))?;
1046
            f.write_all(body)?;
1047
            f.sync_all()?;
1048
        }
4dcb8c09 Andy McCall ↶ 1049
        crate::durable::rename(&tmp, &path)?;
ff0af1ee unknown 1050
        Ok(r)
b30a880f Andy McCall ↶ 1051
    }
1052
1053
    /// Create every fan-out directory of `loose` (`00` to `ff`) and sync
1054
    /// `loose` once. Syncing it is slow on a busy disk (130 ms seen), and
1055
    /// creating the directories one at a time paid that on most writes
1056
    /// after a gc had emptied them; they are kept from then on.
1057
    fn ensure_fanouts(&self) -> io::Result<()> {
1058
        let loose = self.dir.join("loose");
1059
        for b in 0..=255u8 {
1060
            match fs::create_dir(loose.join(format!("{b:02x}"))) {
1061
                Ok(()) => {}
1062
                Err(e) if e.kind() == io::ErrorKind::AlreadyExists => {}
1063
                Err(e) => return Err(e),
1064
            }
1065
        }
1066
        crate::durable::sync_dir(&loose)
1067
    }
1068
1069
    /// Write several loose objects with as few waits on the disk as the
1070
    /// guarantees allow: every file written, then all synced together, then
1071
    /// each renamed into place and each directory synced once. Durable when
1072
    /// this returns, as `put_loose` one at a time would be (spec §13.6).
1073
    fn put_loose_many(&self, objects: &[(ObjectType, &[u8])]) -> Result<()> {
1074
        use rayon::prelude::*;
1075
        if objects.len() <= 1 {
1076
            for (ty, body) in objects {
1077
                self.put_loose(*ty, body)?;
1078
            }
1079
            return Ok(());
1080
        }
1081
        let mut seen = std::collections::HashSet::new();
1082
        let todo: Vec<(PathBuf, PathBuf, ObjectType, &[u8])> = objects
1083
            .iter()
1084
            .filter_map(|(ty, body)| {
1085
                let r = hash_object(*ty, body);
1086
                let path = self.path_for(&r);
1087
                (seen.insert(r) && !path.is_file()).then(|| {
1088
                    let mut nonce = [0u8; 8];
1089
                    rand::rngs::OsRng.fill_bytes(&mut nonce);
1090
                    let tmp = self
1091
                        .dir
1092
                        .join("loose")
1093
                        .join(format!("tmp-{}", hex::encode(nonce)));
1094
                    (path, tmp, *ty, *body)
1095
                })
1096
            })
1097
            .collect();
1098
        if todo.is_empty() {
1099
            return Ok(());
1100
        }
1101
        let fanouts: std::collections::BTreeSet<&Path> =
1102
            todo.iter().map(|(p, ..)| p.parent().unwrap()).collect();
1103
        if fanouts.iter().any(|f| !f.is_dir()) {
1104
            self.ensure_fanouts()?;
1105
        }
1106
        crate::durable::step("write loose objects")?;
1107
        todo.par_iter()
1108
            .try_for_each(|(_, tmp, ty, body)| -> io::Result<()> {
1109
                let mut f = fs::File::create(tmp)?;
1110
                f.write_all(&header(*ty, body.len()))?;
1111
                f.write_all(body)?;
1112
                f.sync_all()
1113
            })?;
1114
        crate::durable::step("rename loose objects")?;
1115
        for (path, tmp, ..) in &todo {
1116
            fs::rename(tmp, path)?;
1117
        }
1118
        fanouts
1119
            .par_iter()
1120
            .try_for_each(|d| crate::durable::sync_dir(d))?;
1121
        Ok(())
ff0af1ee unknown 1122
    }
1123
8f10a07a Andy McCall ↶ 1124
    /// Read and verify an object, loose, packed, or pending in a batch.
ff0af1ee unknown 1125
    pub fn get(&self, r: &ObjectRef) -> Result<(ObjectType, Vec<u8>)> {
afaba5f1 Andy McCall ↶ 1126
        match read_loose_at(&self.path_for(r), r) {
1127
            Some(found) => found,
1128
            None => match self.find_packed(r) {
1129
                Some((i, e)) => self.packs.borrow()[i].read(r, e),
1130
                None => self.pending_get(r).unwrap_or(Err(StoreError::NotFound(*r))),
1131
            },
1132
        }
1133
    }
1134
1135
    /// Many objects at once: loose ones are read and verified in parallel,
1136
    /// the rest (packed, or pending in a batch) one by one. Results are in
1137
    /// the order of `refs`.
1138
    pub fn get_many(&self, refs: &[ObjectRef]) -> Vec<Result<Object>> {
1139
        use rayon::prelude::*;
1140
        let paths: Vec<(ObjectRef, PathBuf)> =
1141
            refs.iter().map(|r| (*r, self.path_for(r))).collect();
1142
        let mut found: Vec<Option<Result<Object>>> =
1143
            paths.par_iter().map(|(r, p)| read_loose_at(p, r)).collect();
1144
        // Packed ones: look them up in order, then read them in parallel.
1145
        let packs = self.packs.borrow();
1146
        let packed: Vec<(usize, usize, PackEntry)> = found
75167672 Andy McCall 1147
            .iter()
afaba5f1 Andy McCall ↶ 1148
            .enumerate()
1149
            .filter(|(_, f)| f.is_none())
1150
            .filter_map(|(i, _)| {
1151
                packs
1152
                    .iter()
1153
                    .enumerate()
1154
                    .find_map(|(p, pack)| pack.find(&refs[i]).map(|e| (i, p, e)))
1155
            })
1156
            .collect();
1157
        let indexes: &[PackIndex] = &packs;
1158
        let read: Vec<(usize, Result<Object>)> = packed
1159
            .par_iter()
1160
            .map(|(i, p, e)| (*i, indexes[*p].read(&refs[*i], *e)))
1161
            .collect();
1162
        drop(packs);
1163
        for (i, r) in read {
1164
            found[i] = Some(r);
ff0af1ee unknown 1165
        }
afaba5f1 Andy McCall ↶ 1166
        found
1167
            .into_iter()
1168
            .zip(refs)
1169
            .map(|(f, r)| match f {
1170
                Some(f) => f,
1171
                None => self.get(r),
1172
            })
1173
            .collect()
ff0af1ee unknown 1174
    }
1175
1176
    pub fn get_typed(&self, r: &ObjectRef, want: ObjectType) -> Result<Vec<u8>> {
1177
        let (ty, body) = self.get(r)?;
1178
        if ty != want {
1179
            return Err(StoreError::WrongType(*r, ty.name(), want.name()));
1180
        }
1181
        Ok(body)
1182
    }
1183
6f8aeb99 Andy McCall ↶ 1184
    /// Store a change and the text blobs (and their chunks) its large
1185
    /// inserts are stored as (spec §6.1).
ff0af1ee unknown 1186
    pub fn put_change(&self, c: &Change) -> Result<ObjectRef> {
5b4c5130 Andy McCall ↶ 1187
        let encoding = crate::timing::phase("store: encode the change and its texts");
17f94b13 Andy McCall ↶ 1188
        let bytes = c.encode_streaming(|batch| self.put_texts(&batch))?;
5b4c5130 Andy McCall ↶ 1189
        drop(encoding);
1190
        let _t = crate::timing::phase("store: put the encoding");
17f94b13 Andy McCall ↶ 1191
        self.put(ObjectType::Change, &bytes)
6f8aeb99 Andy McCall ↶ 1192
    }
1193
17f94b13 Andy McCall ↶ 1194
    /// Store text blobs and their chunks: every chunk at once (compressed
1195
    /// on every core in a large batch), then the blobs.
1196
    pub fn put_texts(&self, texts: &[crate::objects::StoredText]) -> Result<()> {
192fad1e Andy McCall ↶ 1197
        let mut chunks: Vec<(ObjectType, &[u8])> = Vec::new();
1198
        for t in texts {
1199
            if !self.has(&t.blob.hash()) {
1200
                chunks.extend(
1201
                    crate::chunk::chunks(&t.content)
1202
                        .into_iter()
1203
                        .map(|c| (ObjectType::Chunk, c)),
1204
                );
1205
            }
6f8aeb99 Andy McCall ↶ 1206
        }
192fad1e Andy McCall ↶ 1207
        self.put_many(&chunks)?;
1208
        for t in texts {
1209
            self.put(ObjectType::Blob, &t.blob.encode())?;
6f8aeb99 Andy McCall ↶ 1210
        }
17f94b13 Andy McCall ↶ 1211
        Ok(())
ff0af1ee unknown 1212
    }
1213
1214
    pub fn get_change(&self, r: &ObjectRef) -> Result<Change> {
6f8aeb99 Andy McCall ↶ 1215
        Ok(self.get_change_parts(r)?.0)
1216
    }
1217
1218
    /// A change and the objects its stored inserts are made of: each text
1219
    /// blob and its chunks, which travel and are kept with it.
1220
    pub fn get_change_parts(&self, r: &ObjectRef) -> Result<(Change, Vec<ObjectRef>)> {
1221
        let bytes = self.get_typed(r, ObjectType::Change)?;
1222
        let mut parts = Vec::new();
1223
        let mut failed = None;
0f93f49c Andy McCall ↶ 1224
        let decoded = Change::decode_with(&bytes, &mut |text, count, form| match self
1225
            .read_text(text)
1226
        {
6f8aeb99 Andy McCall ↶ 1227
            Ok((blob, content)) => {
1228
                parts.push(*text);
1229
                parts.extend(blob.chunks);
0f93f49c Andy McCall ↶ 1230
                crate::objects::text_lines(&content, count, form)
6f8aeb99 Andy McCall ↶ 1231
            }
1232
            Err(e) => {
1233
                let why = e.to_string();
1234
                failed = Some(e);
1235
                Err(ObjectError::Schema(format!(
1236
                    "text blob {}: {why}",
1237
                    text.short(12)
1238
                )))
1239
            }
1240
        });
1241
        match (decoded, failed) {
1242
            (Ok(c), _) => Ok((c, parts)),
1243
            (Err(_), Some(StoreError::NotFound(t))) => Err(StoreError::Corrupt(
1244
                *r,
1245
                format!("its text blob {} is missing", t.short(12)),
1246
            )),
1247
            (Err(e), _) => Err(StoreError::Corrupt(*r, e.to_string())),
1248
        }
1249
    }
1250
a4fad5b0 Andy McCall ↶ 1251
    /// A change with every insert left without its lines (empty
1252
    /// placeholders), and where each one's lines are, in order: for
1253
    /// history-only states, which read text only for the graphs they build
1254
    /// (see [`Change::decode_skeleton`]).
1255
    pub fn get_change_lazy(&self, r: &ObjectRef) -> Result<(Change, Vec<crate::objects::TextRef>)> {
86e1824f Andy McCall ↶ 1256
        let bytes = self.get_typed(r, ObjectType::Change)?;
a4fad5b0 Andy McCall ↶ 1257
        Change::decode_skeleton(&bytes, *r).map_err(|e| StoreError::Corrupt(*r, e.to_string()))
86e1824f Andy McCall ↶ 1258
    }
1259
69816d68 Andy McCall ↶ 1260
    /// A change's metadata (id, dependencies, author, clock, message, kind)
1261
    /// without its contents: read from its skeleton, so neither its text
1262
    /// blobs nor its inline lines are read (spec §11).
ce14f0d6 Andy McCall ↶ 1263
    ///
1264
    /// Served from the `metas` cache when this store read it before (a
1265
    /// change object can be as large as the tree it adds), as long as the
1266
    /// store still has the change.
69816d68 Andy McCall ↶ 1267
    pub fn get_change_meta(&self, r: &ObjectRef) -> Result<crate::state::ChangeMeta> {
ce14f0d6 Andy McCall ↶ 1268
        if self.metas.borrow().is_none() {
1269
            *self.metas.borrow_mut() = Some(self.read_metas_cache());
1270
        }
1271
        let cached = self.metas.borrow().as_ref().and_then(|m| m.get(r).cloned());
1272
        if let Some(meta) = cached {
1273
            if self.has(r) {
1274
                #[cfg(test)]
1275
                assert_eq!(
1276
                    meta,
1277
                    Change::decode_meta(&self.get_typed(r, ObjectType::Change)?).unwrap(),
1278
                    "the metadata cache holds a change's metadata"
1279
                );
1280
                return Ok(meta);
1281
            }
1282
        }
4d6c9c1b Andy McCall ↶ 1283
        let bytes = self.get_typed(r, ObjectType::Change)?;
ce14f0d6 Andy McCall ↶ 1284
        let meta =
1285
            Change::decode_meta(&bytes).map_err(|e| StoreError::Corrupt(*r, e.to_string()))?;
1286
        self.remember_meta(r, &meta);
1287
        Ok(meta)
1288
    }
1289
1290
    /// The metadata cache: records of a change hash, its metadata in the
1291
    /// flat layout, and the first 8 bytes of BLAKE3 over both, each
1292
    /// appended in one write. Derived only: a record that does not check
1293
    /// ends the reading (a torn write), and the cache can be deleted.
1294
    fn read_metas_cache(&self) -> crate::fasthash::FastMap<ObjectRef, crate::state::ChangeMeta> {
1295
        let mut out = crate::fasthash::FastMap::default();
1296
        let Ok(bytes) = fs::read(self.dir.join("metas")) else {
1297
            return out;
1298
        };
1299
        let mut r = crate::flat::Reader::new(&bytes);
1300
        let mut good = 0;
1301
        while !r.at_end() {
1302
            let mut record = || -> std::result::Result<_, ObjectError> {
1303
                let h = r.object()?;
1304
                let body = r.bytes()?;
1305
                let check = r.fixed::<8>()?;
1306
                Ok((h, body, check))
1307
            };
1308
            let Ok((h, body, check)) = record() else {
1309
                break;
1310
            };
1311
            let mut sum = blake3::Hasher::new();
1312
            sum.update(&h.0).update(body);
1313
            if sum.finalize().as_bytes()[..8] != check {
1314
                break;
1315
            }
1316
            match crate::state::ChangeMeta::read_flat(&mut crate::flat::Reader::new(body)) {
1317
                Ok(m) => out.insert(h, m),
1318
                Err(_) => break,
1319
            };
1320
            good += 32 + 4 + body.len() + 8;
1321
        }
1322
        // Cut a bad tail, so records appended later are read.
1323
        if good < bytes.len() {
1324
            let _ = fs::OpenOptions::new()
1325
                .write(true)
1326
                .open(self.dir.join("metas"))
1327
                .and_then(|f| f.set_len(good as u64));
1328
        }
1329
        out
1330
    }
1331
1332
    /// Add a change's metadata to the cache; quietly nothing when it
1333
    /// cannot be written (it is only a cache).
1334
    fn remember_meta(&self, r: &ObjectRef, meta: &crate::state::ChangeMeta) {
1335
        let mut body = crate::flat::Writer::default();
1336
        meta.write_flat(&mut body);
1337
        let mut w = crate::flat::Writer::default();
1338
        w.object(r);
1339
        w.bytes(&body.0);
1340
        let mut sum = blake3::Hasher::new();
1341
        sum.update(&r.0).update(&body.0);
1342
        w.fixed(&sum.finalize().as_bytes()[..8]);
1343
        let _ = fs::OpenOptions::new()
1344
            .create(true)
1345
            .append(true)
1346
            .open(self.dir.join("metas"))
1347
            .and_then(|mut f| f.write_all(&w.0));
1348
        if let Some(m) = self.metas.borrow_mut().as_mut() {
1349
            m.insert(*r, meta.clone());
1350
        }
69816d68 Andy McCall ↶ 1351
    }
1352
6f8aeb99 Andy McCall ↶ 1353
    fn read_text(&self, r: &ObjectRef) -> Result<(Blob, Vec<u8>)> {
1354
        let blob = Blob::decode(&self.get_typed(r, ObjectType::Blob)?)?;
1355
        let mut content = Vec::with_capacity(blob.length as usize);
1356
        for c in &blob.chunks {
1357
            content.extend(self.get_typed(c, ObjectType::Chunk)?);
1358
        }
1359
        Ok((blob, content))
ff0af1ee unknown 1360
    }
1361
1362
    pub fn put_chunk(&self, bytes: &[u8]) -> Result<ObjectRef> {
1363
        self.put(ObjectType::Chunk, bytes)
1364
    }
1365
cdb5beba Andy McCall ↶ 1366
    /// Store a chunk that packs will keep uncompressed (`nocompress`, spec §13.3).
1367
    pub fn put_chunk_nocompress(&self, bytes: &[u8]) -> Result<ObjectRef> {
8f10a07a Andy McCall ↶ 1368
        let r = hash_object(ObjectType::Chunk, bytes);
cdb5beba Andy McCall ↶ 1369
        let dir = self.dir.join("nocompress");
1370
        fs::create_dir_all(&dir)?;
1371
        let _ = fs::File::create(dir.join(r.to_hex()));
8f10a07a Andy McCall ↶ 1372
        self.put(ObjectType::Chunk, bytes)
cdb5beba Andy McCall ↶ 1373
    }
1374
1375
    fn is_nocompress(&self, r: &ObjectRef) -> bool {
1376
        self.dir.join("nocompress").join(r.to_hex()).exists()
1377
    }
1378
86e682c2 Andy McCall ↶ 1379
    pub fn put_state(&self, st: &StateObject) -> Result<ObjectRef> {
1380
        self.put(ObjectType::State, &st.encode())
1381
    }
1382
1383
    pub fn get_state(&self, r: &ObjectRef) -> Result<StateObject> {
1384
        Ok(StateObject::decode(&self.get_typed(r, ObjectType::State)?)?)
1385
    }
1386
5b9c6ad4 Andy McCall ↶ 1387
    pub fn put_attest(&self, a: &Attest) -> Result<ObjectRef> {
1388
        self.put(ObjectType::Attest, &a.encode())
1389
    }
1390
1391
    pub fn get_attest(&self, r: &ObjectRef) -> Result<Attest> {
1392
        Ok(Attest::decode(&self.get_typed(r, ObjectType::Attest)?)?)
fc1a466a Andy McCall ↶ 1393
    }
1394
1395
    pub fn put_receipt(&self, r: &Receipt) -> Result<ObjectRef> {
1396
        self.put(ObjectType::Receipt, &r.encode())
5b9c6ad4 Andy McCall ↶ 1397
    }
1398
fc1a466a Andy McCall ↶ 1399
    pub fn get_receipt(&self, r: &ObjectRef) -> Result<Receipt> {
1400
        Ok(Receipt::decode(&self.get_typed(r, ObjectType::Receipt)?)?)
5b9c6ad4 Andy McCall ↶ 1401
    }
1402
fc1a466a Andy McCall ↶ 1403
    fn index_path(&self, index: &str, key: &ObjectRef) -> PathBuf {
1404
        let h = key.to_hex();
1405
        self.dir.join(index).join(&h[..2]).join(&h[2..])
1406
    }
1407
1408
    /// Add `value` to the sorted list kept for `key` in `index`.
1409
    fn index_add(&self, index: &str, key: &ObjectRef, value: &ObjectRef) -> Result<()> {
1410
        let mut values = self.index_get(index, key)?;
1411
        if values.contains(value) {
5b9c6ad4 Andy McCall ↶ 1412
            return Ok(());
1413
        }
fc1a466a Andy McCall ↶ 1414
        values.push(*value);
1415
        values.sort();
1416
        let path = self.index_path(index, key);
5b9c6ad4 Andy McCall ↶ 1417
        fs::create_dir_all(path.parent().unwrap())?;
fc1a466a Andy McCall ↶ 1418
        let text: String = values.iter().map(|s| format!("{}\n", s.to_hex())).collect();
4dcb8c09 Andy McCall ↶ 1419
        crate::durable::write_atomic(&path, text.as_bytes())?;
5b9c6ad4 Andy McCall ↶ 1420
        Ok(())
1421
    }
1422
fc1a466a Andy McCall ↶ 1423
    fn index_get(&self, index: &str, key: &ObjectRef) -> Result<Vec<ObjectRef>> {
1424
        match fs::read_to_string(self.index_path(index, key)) {
5b9c6ad4 Andy McCall ↶ 1425
            Ok(t) => Ok(t.lines().filter_map(ObjectRef::from_hex).collect()),
1426
            Err(e) if e.kind() == io::ErrorKind::NotFound => Ok(vec![]),
1427
            Err(e) => Err(e.into()),
1428
        }
fc1a466a Andy McCall ↶ 1429
    }
1430
1431
    /// Record that seal attestation `attest` covers `change`.
1432
    pub fn index_seal(&self, change: &ObjectRef, attest: &ObjectRef) -> Result<()> {
1433
        self.index_add("seals", change, attest)
1434
    }
1435
1436
    /// Seal attestations known to cover `change`.
1437
    pub fn seals_for(&self, change: &ObjectRef) -> Result<Vec<ObjectRef>> {
1438
        self.index_get("seals", change)
1439
    }
1440
1441
    /// Record receipt attestation `attest` for `state` (spec §14.6).
1442
    pub fn index_receipt(&self, state: &ObjectRef, attest: &ObjectRef) -> Result<()> {
1443
        self.index_add("receipts", state, attest)
1444
    }
1445
1446
    /// Receipt attestations recorded for `state`.
1447
    pub fn receipts_for(&self, state: &ObjectRef) -> Result<Vec<ObjectRef>> {
1448
        self.index_get("receipts", state)
1449
    }
1450
1451
    /// Every state with a receipt, for garbage collection.
1452
    pub fn receipt_states(&self) -> Result<Vec<ObjectRef>> {
1453
        let mut out = Vec::new();
1454
        let dir = self.dir.join("receipts");
1455
        let top = match fs::read_dir(&dir) {
1456
            Ok(d) => d,
1457
            Err(e) if e.kind() == io::ErrorKind::NotFound => return Ok(out),
1458
            Err(e) => return Err(e.into()),
1459
        };
1460
        for d in top {
1461
            let d = d?;
1462
            let prefix = d.file_name().to_string_lossy().into_owned();
1463
            for f in fs::read_dir(d.path())? {
1464
                let name = f?.file_name().to_string_lossy().into_owned();
1465
                if let Some(r) = ObjectRef::from_hex(&format!("{prefix}{name}")) {
1466
                    out.push(r);
1467
                }
1468
            }
1469
        }
1470
        out.sort();
1471
        Ok(out)
5b9c6ad4 Andy McCall ↶ 1472
    }
1473
1983686f Andy McCall ↶ 1474
    pub fn put_identity(&self, id: &Identity) -> Result<ObjectRef> {
1475
        self.put(ObjectType::Identity, &id.encode())
1476
    }
1477
1478
    pub fn get_identity(&self, r: &ObjectRef) -> Result<Identity> {
1479
        Ok(Identity::decode(&self.get_typed(r, ObjectType::Identity)?)?)
1480
    }
1481
1482
    fn sig_path(&self, r: &ObjectRef) -> PathBuf {
1483
        let h = r.to_hex();
1484
        self.dir.join("sigs").join(&h[..2]).join(&h[2..])
1485
    }
1486
1487
    /// Signatures stored alongside object `r` (spec §5.4). Empty if none.
1488
    pub fn signatures(&self, r: &ObjectRef) -> Result<Vec<Signature>> {
1489
        match fs::read(self.sig_path(r)) {
1490
            Ok(b) => Ok(Signature::decode_list(&b)?),
1491
            Err(e) if e.kind() == io::ErrorKind::NotFound => Ok(vec![]),
1492
            Err(e) => Err(e.into()),
1493
        }
1494
    }
1495
1496
    /// Attach a signature to object `r`, replacing any earlier one by the same key.
1497
    pub fn put_signature(&self, r: &ObjectRef, sig: &Signature) -> Result<()> {
1498
        let mut sigs = self.signatures(r)?;
1499
        sigs.retain(|s| s.key != sig.key);
1500
        sigs.push(sig.clone());
5b1114a0 Andy McCall ↶ 1501
        sigs.sort_by_key(|s| s.key);
1983686f Andy McCall ↶ 1502
        let path = self.sig_path(r);
1503
        fs::create_dir_all(path.parent().unwrap())?;
1504
        let mut nonce = [0u8; 8];
1505
        rand::rngs::OsRng.fill_bytes(&mut nonce);
1506
        let tmp = self.dir.join(format!("sigs-tmp-{}", hex::encode(nonce)));
1507
        {
1508
            let mut f = fs::File::create(&tmp)?;
1509
            f.write_all(&Signature::encode_list(&sigs))?;
1510
            f.sync_all()?;
1511
        }
4dcb8c09 Andy McCall ↶ 1512
        crate::durable::rename(&tmp, &path)?;
1983686f Andy McCall ↶ 1513
        Ok(())
1514
    }
1515
d02e545d Andy McCall ↶ 1516
    /// Every object reference in the store, loose and packed (slow; for gc and tests).
ff0af1ee unknown 1517
    pub fn list(&self) -> io::Result<Vec<ObjectRef>> {
d02e545d Andy McCall ↶ 1518
        let mut out: Vec<ObjectRef> = self.loose_objects()?.into_iter().map(|(r, _)| r).collect();
1519
        for p in self.packs.borrow().iter() {
1520
            out.extend(p.refs.iter().copied());
ff0af1ee unknown 1521
        }
1522
        out.sort();
d02e545d Andy McCall ↶ 1523
        out.dedup();
ff0af1ee unknown 1524
        Ok(out)
1525
    }
afaba5f1 Andy McCall ↶ 1526
}
1527
1528
/// A loose object at `path`, verified against `r`; `None` when there is no
1529
/// such file (it may be packed).
1530
fn read_loose_at(path: &Path, r: &ObjectRef) -> Option<Result<Object>> {
1531
    let data = match fs::read(path) {
1532
        Ok(d) => d,
1533
        Err(e) if e.kind() == io::ErrorKind::NotFound => return None,
1534
        Err(e) => return Some(Err(e.into())),
1535
    };
1536
    Some((|| {
1537
        let nul = data
1538
            .iter()
1539
            .position(|&b| b == 0)
1540
            .ok_or_else(|| StoreError::Corrupt(*r, "no header".into()))?;
1541
        let head = std::str::from_utf8(&data[..nul])
1542
            .map_err(|_| StoreError::Corrupt(*r, "header not UTF-8".into()))?;
1543
        let mut parts = head.split(' ');
1544
        let (Some("oxy1"), Some(ty), Some(len), None) =
1545
            (parts.next(), parts.next(), parts.next(), parts.next())
1546
        else {
1547
            return Err(StoreError::Corrupt(*r, format!("bad header {head:?}")));
1548
        };
1549
        let ty = ObjectType::from_name(ty)
1550
            .ok_or_else(|| StoreError::Corrupt(*r, format!("unknown type {ty}")))?;
1551
        let len: usize = len
1552
            .parse()
1553
            .map_err(|_| StoreError::Corrupt(*r, "bad length".into()))?;
1554
        let body = &data[nul + 1..];
1555
        if body.len() != len {
1556
            return Err(StoreError::Corrupt(
1557
                *r,
1558
                format!("length {} != header {len}", body.len()),
1559
            ));
1560
        }
1561
        if hash_object(ty, body) != *r {
1562
            return Err(StoreError::Corrupt(*r, "hash mismatch".into()));
1563
        }
1564
        Ok((ty, body.to_vec()))
1565
    })())
ff0af1ee unknown 1566
}
1567
1568
#[cfg(test)]
1569
mod tests {
1570
    use super::*;
1571
1572
    #[test]
1573
    fn put_get_verify() {
1574
        let dir = tempfile::tempdir().unwrap();
1575
        let s = Store::open(dir.path().join("objects")).unwrap();
1576
        let r = s.put(ObjectType::Chunk, b"hello").unwrap();
1577
        assert!(s.has(&r));
1578
        assert_eq!(s.get(&r).unwrap(), (ObjectType::Chunk, b"hello".to_vec()));
1579
        assert_eq!(s.put(ObjectType::Chunk, b"hello").unwrap(), r);
1580
        assert_eq!(s.list().unwrap(), vec![r]);
75167672 Andy McCall 1581
        assert!(matches!(
1582
            s.get_typed(&r, ObjectType::Blob),
1583
            Err(StoreError::WrongType(..))
1584
        ));
1585
        assert!(matches!(
1586
            s.get(&ObjectRef([9; 32])),
1587
            Err(StoreError::NotFound(_))
1588
        ));
ff0af1ee unknown 1589
1590
        // Tamper: flip a byte, read must fail.
1591
        let path = s.path_for(&r);
1592
        let mut data = fs::read(&path).unwrap();
1593
        let last = data.len() - 1;
1594
        data[last] ^= 1;
1595
        fs::write(&path, data).unwrap();
1596
        assert!(matches!(s.get(&r), Err(StoreError::Corrupt(..))));
1597
    }
1598
1599
    #[test]
d02e545d Andy McCall ↶ 1600
    fn packing_keeps_every_object_readable() {
1601
        let dir = tempfile::tempdir().unwrap();
1602
        let s = Store::open(dir.path()).unwrap();
1603
        let mut refs = Vec::new();
1604
        for i in 0..300u32 {
1605
            let body = format!("object {i} ").repeat(1 + (i % 7) as usize);
1606
            refs.push((s.put(ObjectType::Chunk, body.as_bytes()).unwrap(), body));
1607
        }
1608
        let c = crate::objects::tests::sample_change();
1609
        let cr = s.put_change(&c).unwrap();
1610
        assert_eq!(s.loose_count().unwrap(), 301);
1611
        let report = s.pack_loose().unwrap();
1612
        assert_eq!(report.objects, 301);
1613
        assert!(report.packed_bytes < report.raw_bytes, "{report:?}");
1614
        assert_eq!(s.loose_count().unwrap(), 0);
1615
        assert_eq!(s.pack_count(), 1);
1616
        for (r, body) in &refs {
1617
            assert!(s.has(r));
75167672 Andy McCall 1618
            assert_eq!(
1619
                s.get(r).unwrap(),
1620
                (ObjectType::Chunk, body.as_bytes().to_vec())
1621
            );
d02e545d Andy McCall ↶ 1622
        }
1623
        assert_eq!(s.get_change(&cr).unwrap(), c);
1624
        assert_eq!(s.list().unwrap().len(), 301);
1625
        // Re-putting a packed object is a no-op; new objects go loose and a
1626
        // second pack coexists with the first.
75167672 Andy McCall 1627
        assert_eq!(
1628
            s.put(ObjectType::Chunk, refs[0].1.as_bytes()).unwrap(),
1629
            refs[0].0
1630
        );
d02e545d Andy McCall ↶ 1631
        assert_eq!(s.loose_count().unwrap(), 0);
1632
        let extra = s.put(ObjectType::Chunk, b"after packing").unwrap();
1633
        s.pack_loose().unwrap();
1634
        assert_eq!(s.pack_count(), 2);
1635
        assert_eq!(s.get(&extra).unwrap().1, b"after packing");
1636
        // A reopened store sees the packs.
1637
        let s2 = Store::open(dir.path()).unwrap();
1638
        assert_eq!(s2.pack_count(), 2);
1639
        assert_eq!(s2.get(&refs[5].0).unwrap().1, refs[5].1.as_bytes());
75167672 Andy McCall 1640
        assert!(matches!(
1641
            s2.get(&ObjectRef([9; 32])),
1642
            Err(StoreError::NotFound(_))
1643
        ));
d02e545d Andy McCall ↶ 1644
        // Corrupting an object's body inside the pack is detected on read.
1645
        let target = refs[5].0;
1646
        let (pack_path, offset) = {
1647
            let packs = s2.packs.borrow();
1648
            let (i, e) = s2.find_packed(&target).unwrap();
1649
            (packs[i].path.clone(), e.offset as usize)
1650
        };
1651
        let mut data = fs::read(&pack_path).unwrap();
1652
        data[offset] ^= 0xff;
1653
        fs::write(&pack_path, data).unwrap();
1654
        let s3 = Store::open(dir.path()).unwrap();
1655
        assert!(matches!(s3.get(&target), Err(StoreError::Corrupt(..))));
75167672 Andy McCall 1656
        assert_eq!(
1657
            s3.get(&refs[6].0).unwrap().1,
1658
            refs[6].1.as_bytes(),
1659
            "other objects are unaffected"
1660
        );
a17fc622 Andy McCall ↶ 1661
    }
1662
1663
    #[test]
1664
    fn a_full_repack_trains_a_dictionary_and_reads_back() {
1665
        let dir = tempfile::tempdir().unwrap();
1666
        let s = Store::open(dir.path()).unwrap();
1667
        // Many small, alike files: what a dictionary is for.
1668
        let bodies: Vec<Vec<u8>> = (0..1500)
1669
            .map(|i| {
1670
                format!(
1671
                    "// SPDX-License-Identifier: GPL-2.0\n#include <linux/module.h>\n\
1672
                     static int probe_{i}(struct device *dev)\n{{\n\treturn {i};\n}}\n"
1673
                )
1674
                .into_bytes()
1675
            })
1676
            .collect();
1677
        let refs: Vec<ObjectRef> = bodies
1678
            .iter()
1679
            .map(|b| s.put(ObjectType::Chunk, b).unwrap())
1680
            .collect();
1681
        s.repack_all(REPACK_ZSTD_LEVEL).unwrap();
1682
        let dicts = fs::read_dir(dir.path())
1683
            .unwrap()
1684
            .flatten()
1685
            .filter(|e| e.path().extension().is_some_and(|x| x == "dict"))
1686
            .count();
1687
        assert_eq!(dicts, 1, "the pack has a dictionary");
1688
        let with_dict = refs
1689
            .iter()
1690
            .filter(|r| s.find_packed(r).unwrap().1.flags & FLAG_DICT != 0)
1691
            .count();
1692
        assert!(with_dict > 1000, "most use it: {with_dict}");
1693
        let again = Store::open(dir.path()).unwrap();
1694
        for (r, b) in refs.iter().zip(&bodies) {
1695
            assert_eq!(&again.get(r).unwrap().1, b);
1696
        }
1697
        // A repack after that replaces the pack and its dictionary.
1698
        again.repack_all(REPACK_ZSTD_LEVEL).unwrap();
1699
        let files: Vec<String> = fs::read_dir(dir.path())
1700
            .unwrap()
1701
            .flatten()
1702
            .map(|e| e.file_name().to_string_lossy().into_owned())
1703
            .filter(|n| n.starts_with("pack-"))
1704
            .collect();
1705
        assert_eq!(files.len(), 3, "one pack, index and dictionary: {files:?}");
1706
        assert_eq!(&again.get(&refs[7]).unwrap().1, &bodies[7]);
d02e545d Andy McCall ↶ 1707
    }
1708
1709
    #[test]
cdb5beba Andy McCall ↶ 1710
    fn nocompress_chunks_pack_raw() {
1711
        let dir = tempfile::tempdir().unwrap();
1712
        let s = Store::open(dir.path()).unwrap();
1713
        let body = vec![b'a'; 10_000];
1714
        let raw = s.put_chunk_nocompress(&body).unwrap();
1715
        let compressible = s.put(ObjectType::Chunk, &vec![b'b'; 10_000]).unwrap();
1716
        s.pack_loose().unwrap();
1717
        let packs = s.packs.borrow();
1718
        let (i, e) = s.find_packed(&raw).unwrap();
1719
        assert_eq!((e.flags, e.stored_len as usize), (0, 10_000));
1720
        let (_, c) = s.find_packed(&compressible).unwrap();
1721
        assert_eq!(c.flags, FLAG_ZSTD);
1722
        assert!(c.stored_len < 10_000);
1723
        assert_eq!(packs[i].read(&raw, e).unwrap().1, body);
1724
    }
1725
1726
    #[test]
135212de Andy McCall ↶ 1727
    fn many_packs_merge_into_one() {
1728
        let dir = tempfile::tempdir().unwrap();
1729
        let s = Store::open(dir.path()).unwrap();
1730
        let mut refs = Vec::new();
1731
        for i in 0..(MAX_PACKS + 2) {
75167672 Andy McCall 1732
            refs.push(
1733
                s.put(ObjectType::Chunk, format!("pack {i}").as_bytes())
1734
                    .unwrap(),
1735
            );
135212de Andy McCall ↶ 1736
            s.pack_loose().unwrap();
1737
        }
1738
        assert_eq!(s.pack_count(), MAX_PACKS + 2);
1739
        assert_eq!(s.merge_packs().unwrap(), MAX_PACKS + 2);
1740
        assert_eq!(s.pack_count(), 1);
1741
        assert_eq!(s.merge_packs().unwrap(), 0);
1742
        for (i, r) in refs.iter().enumerate() {
1743
            assert_eq!(s.get(r).unwrap().1, format!("pack {i}").as_bytes());
1744
        }
1745
        assert_eq!(Store::open(dir.path()).unwrap().pack_count(), 1);
1746
    }
1747
1748
    #[test]
ba160f60 Andy McCall ↶ 1749
    fn prune_removes_unreachable_loose_and_packed_objects() {
1750
        use std::collections::HashSet;
1751
        let dir = tempfile::tempdir().unwrap();
1752
        let s = Store::open(dir.path()).unwrap();
1753
        let keep_a = s.put(ObjectType::Chunk, b"keep a").unwrap();
1754
        let drop_a = s.put(ObjectType::Chunk, b"drop a").unwrap();
1755
        s.pack_loose().unwrap();
1756
        let keep_b = s.put(ObjectType::Chunk, b"keep b").unwrap();
1757
        let drop_b = s.put(ObjectType::Chunk, b"drop b").unwrap();
75167672 Andy McCall 1758
        let sk = crate::identity::SecretKey {
1759
            ed_seed: [1; 32],
1760
            ml_seed: [2; 32],
1761
        };
ba160f60 Andy McCall ↶ 1762
        let id = sk.identity("a", "b", 0);
1763
        let key = s.put_identity(&id).unwrap();
75167672 Andy McCall 1764
        s.put_signature(
1765
            &drop_b,
1766
            &sk.sign(crate::identity::Domain::Change, &key, &drop_b),
1767
        )
1768
        .unwrap();
ba160f60 Andy McCall ↶ 1769
        s.index_seal(&drop_a, &ObjectRef([3; 32])).unwrap();
1770
        let keep: HashSet<ObjectRef> = [keep_a, keep_b, key].into_iter().collect();
1771
        // Everything is too young under a grace period.
75167672 Andy McCall 1772
        let r = s
1773
            .prune(&keep, std::time::Duration::from_secs(3600))
1774
            .unwrap();
ba160f60 Andy McCall ↶ 1775
        assert_eq!((r.objects, r.deferred), (0, 2));
1776
        let r = s.prune(&keep, std::time::Duration::ZERO).unwrap();
1777
        assert_eq!(r.objects, 2, "{r:?}");
1778
        assert!(s.has(&keep_a) && s.has(&keep_b) && s.has(&key));
1779
        assert!(!s.has(&drop_a) && !s.has(&drop_b));
1780
        assert!(s.signatures(&drop_b).unwrap().is_empty());
1781
        assert!(s.seals_for(&drop_a).unwrap().is_empty());
1782
        assert_eq!(s.pack_count(), 1);
1783
        assert_eq!(s.get(&keep_a).unwrap().1, b"keep a");
1784
        assert_eq!(s.list().unwrap().len(), 3);
1785
    }
1786
1787
    #[test]
1983686f Andy McCall ↶ 1788
    fn signatures_are_stored_beside_objects() {
1789
        use crate::identity::{Domain, SecretKey};
1790
        let dir = tempfile::tempdir().unwrap();
1791
        let s = Store::open(dir.path()).unwrap();
1792
        let r = s.put(ObjectType::Chunk, b"x").unwrap();
1793
        assert!(s.signatures(&r).unwrap().is_empty());
75167672 Andy McCall 1794
        let sk = SecretKey {
1795
            ed_seed: [1; 32],
1796
            ml_seed: [2; 32],
1797
        };
1983686f Andy McCall ↶ 1798
        let id = sk.identity("a", "b", 0);
1799
        let key = s.put_identity(&id).unwrap();
1800
        assert_eq!(s.get_identity(&key).unwrap(), id);
1801
        let sig = sk.sign(Domain::Change, &key, &r);
1802
        s.put_signature(&r, &sig).unwrap();
1803
        s.put_signature(&r, &sig).unwrap();
1804
        assert_eq!(s.signatures(&r).unwrap(), vec![sig]);
1805
        // The object itself is untouched by its signatures.
1806
        assert_eq!(s.get(&r).unwrap().1, b"x");
1807
    }
1808
1809
    #[test]
ff0af1ee unknown 1810
    fn change_round_trip_through_store() {
1811
        let dir = tempfile::tempdir().unwrap();
1812
        let s = Store::open(dir.path()).unwrap();
1813
        let c = crate::objects::tests::sample_change();
1814
        let r = s.put_change(&c).unwrap();
1815
        assert_eq!(r, c.hash());
1816
        assert_eq!(s.get_change(&r).unwrap(), c);
1817
    }
2b872d07 Andy McCall ↶ 1818
1819
    fn xorshift(state: &mut u64) -> u64 {
1820
        *state ^= *state << 13;
1821
        *state ^= *state >> 7;
1822
        *state ^= *state << 17;
1823
        *state
1824
    }
1825
1826
    /// Random corruption of pack and index files never panics: reads fail
1827
    /// with an error or the store refuses to open, and untouched objects
1828
    /// stay readable when the index survives.
1829
    #[test]
1830
    fn corrupted_packs_and_indexes_fail_cleanly() {
1831
        let dir = tempfile::tempdir().unwrap();
1832
        let objects = dir.path().join("objects");
1833
        let s = Store::open(&objects).unwrap();
1834
        let mut refs = Vec::new();
1835
        let mut seed = 0x9e3779b97f4a7c15u64;
1836
        for i in 0..40u32 {
1837
            let len = (xorshift(&mut seed) % 5000) as usize;
1838
            let mut body = vec![0u8; len];
1839
            for b in body.iter_mut() {
75167672 Andy McCall 1840
                *b = if i % 2 == 0 {
1841
                    (xorshift(&mut seed) & 0xff) as u8
1842
                } else {
1843
                    b'a' + (i % 26) as u8
1844
                };
2b872d07 Andy McCall ↶ 1845
            }
1846
            refs.push(s.put(ObjectType::Chunk, &body).unwrap());
1847
        }
1848
        s.pack_loose().unwrap();
75167672 Andy McCall 1849
        let pack = fs::read_dir(&objects)
1850
            .unwrap()
1851
            .filter_map(|e| e.ok().map(|e| e.path()))
1852
            .find(|p| p.extension().is_some_and(|x| x == "pack"))
1853
            .unwrap();
2b872d07 Andy McCall ↶ 1854
        let idx = pack.with_extension("idx");
1855
        let pristine_pack = fs::read(&pack).unwrap();
1856
        let pristine_idx = fs::read(&idx).unwrap();
1857
1858
        for round in 0..300u32 {
75167672 Andy McCall 1859
            let (path, pristine) = if round % 2 == 0 {
1860
                (&pack, &pristine_pack)
1861
            } else {
1862
                (&idx, &pristine_idx)
1863
            };
2b872d07 Andy McCall ↶ 1864
            let mut data = pristine.clone();
1865
            match xorshift(&mut seed) % 4 {
1866
                0 => {
1867
                    let at = (xorshift(&mut seed) as usize) % data.len();
1868
                    data[at] ^= 1 << (xorshift(&mut seed) % 8);
1869
                }
1870
                1 => {
1871
                    let at = (xorshift(&mut seed) as usize) % data.len();
1872
                    data[at] = (xorshift(&mut seed) & 0xff) as u8;
1873
                }
1874
                2 => {
1875
                    let keep = (xorshift(&mut seed) as usize) % data.len();
1876
                    data.truncate(keep);
1877
                }
1878
                _ => {
1879
                    let at = (xorshift(&mut seed) as usize) % data.len();
1880
                    let n = ((xorshift(&mut seed) % 8) + 1) as usize;
1881
                    for k in 0..n.min(data.len() - at) {
1882
                        data[at + k] = 0xff;
1883
                    }
1884
                }
1885
            }
1886
            fs::write(path, &data).unwrap();
1887
            // Opening may refuse a broken index; a store that opens must
1888
            // either return each object intact or report it corrupt/missing.
1889
            if let Ok(s2) = Store::open(&objects) {
1890
                for r in &refs {
1891
                    match s2.get(r) {
75167672 Andy McCall 1892
                        Ok((ty, body)) => assert_eq!(
1893
                            hash_object(ty, &body),
1894
                            *r,
1895
                            "a returned object always matches its reference"
1896
                        ),
1897
                        Err(StoreError::Corrupt(..))
1898
                        | Err(StoreError::NotFound(_))
1899
                        | Err(StoreError::Io(_)) => {}
2b872d07 Andy McCall ↶ 1900
                        Err(e) => panic!("unexpected error {e}"),
1901
                    }
1902
                }
1903
            }
1904
            fs::write(path, pristine).unwrap();
1905
        }
1906
        // Restored files read back in full.
1907
        let s3 = Store::open(&objects).unwrap();
1908
        for r in &refs {
1909
            let (ty, body) = s3.get(r).unwrap();
1910
            assert_eq!(hash_object(ty, &body), *r);
1911
        }
1912
    }
8f10a07a Andy McCall ↶ 1913
1914
    /// A small batch becomes loose objects; a large one becomes a single
1915
    /// pack. Objects are readable while pending, and batches nest.
1916
    #[test]
1917
    fn batches_write_loose_or_one_pack() {
1918
        let dir = tempfile::tempdir().unwrap();
1919
        let s = Store::open(dir.path().join("objects")).unwrap();
1920
        let outer = s.begin_batch().unwrap();
1921
        let a = s.put(ObjectType::Chunk, b"small a").unwrap();
1922
        {
1923
            let inner = s.begin_batch().unwrap();
1924
            let b = s.put(ObjectType::Chunk, b"small b").unwrap();
1925
            assert!(s.has(&b));
1926
            assert_eq!(s.get(&b).unwrap().1, b"small b");
1927
            inner.finish().unwrap();
75167672 Andy McCall 1928
            assert_eq!(
1929
                s.loose_count().unwrap(),
1930
                0,
1931
                "the inner batch defers to the outer one"
1932
            );
8f10a07a Andy McCall ↶ 1933
        }
1934
        assert_eq!(s.get(&a).unwrap().1, b"small a");
75167672 Andy McCall 1935
        assert_eq!(
1936
            s.put(ObjectType::Chunk, b"small a").unwrap(),
1937
            a,
1938
            "idempotent while pending"
1939
        );
8f10a07a Andy McCall ↶ 1940
        outer.finish().unwrap();
1941
        assert_eq!(s.loose_count().unwrap(), 2);
1942
        assert_eq!(s.pack_count(), 0);
1943
        assert_eq!(s.get(&a).unwrap().1, b"small a");
1944
1945
        // Large: spills into one pack, readable during and after.
1946
        let big = s.begin_batch().unwrap();
1947
        let mut refs = Vec::new();
1948
        let mut seed = 7u64;
1949
        for i in 0..300u32 {
1950
            let mut body = vec![b'z'; 40_000];
1951
            if i % 3 == 0 {
1952
                for b in body.iter_mut() {
1953
                    *b = (xorshift(&mut seed) & 0xff) as u8;
1954
                }
1955
            }
1956
            body[..4].copy_from_slice(&i.to_le_bytes());
75167672 Andy McCall 1957
            let r = if i % 5 == 0 {
1958
                s.put_chunk_nocompress(&body).unwrap()
1959
            } else {
1960
                s.put(ObjectType::Chunk, &body).unwrap()
1961
            };
1962
            assert_eq!(
1963
                s.get(&r).unwrap().1,
1964
                body,
1965
                "readable while pending (object {i})"
1966
            );
8f10a07a Andy McCall ↶ 1967
            refs.push((r, body));
1968
        }
1969
        assert!(s.has(&refs[0].0));
1970
        big.finish().unwrap();
1971
        assert_eq!(s.pack_count(), 1);
75167672 Andy McCall 1972
        assert_eq!(
1973
            s.loose_count().unwrap(),
1974
            2,
1975
            "nothing from the big batch is loose"
1976
        );
1977
        assert!(fs::read_dir(dir.path().join("objects"))
1978
            .unwrap()
1979
            .flatten()
1980
            .all(|e| !e.file_name().to_string_lossy().starts_with("tmp-")));
8f10a07a Andy McCall ↶ 1981
        let s2 = Store::open(dir.path().join("objects")).unwrap();
1982
        for (r, body) in &refs {
1983
            assert_eq!(s2.get(r).unwrap().1, *body);
1984
        }
1985
        // A dropped batch still writes its objects.
1986
        {
1987
            let _b = s2.begin_batch().unwrap();
1988
            s2.put(ObjectType::Chunk, b"dropped").unwrap();
1989
        }
75167672 Andy McCall 1990
        assert_eq!(
1991
            s2.get(&hash_object(ObjectType::Chunk, b"dropped"))
1992
                .unwrap()
1993
                .1,
1994
            b"dropped"
1995
        );
afaba5f1 Andy McCall ↶ 1996
    }
1997
1998
    #[test]
1999
    fn get_many_reads_loose_and_packed_objects_in_order() {
2000
        let dir = tempfile::tempdir().unwrap();
2001
        let s = Store::open(dir.path().join("objects")).unwrap();
2002
        let packed: Vec<ObjectRef> = (0..20)
2003
            .map(|i| {
2004
                s.put(ObjectType::Chunk, format!("packed {i}").as_bytes())
2005
                    .unwrap()
2006
            })
2007
            .collect();
2008
        s.pack_loose().unwrap();
2009
        let loose: Vec<ObjectRef> = (0..20)
2010
            .map(|i| {
2011
                s.put(ObjectType::Chunk, format!("loose {i}").as_bytes())
2012
                    .unwrap()
2013
            })
2014
            .collect();
2015
        let missing = hash_object(ObjectType::Chunk, b"never stored");
2016
        let mut refs: Vec<ObjectRef> = Vec::new();
2017
        for i in 0..20 {
2018
            refs.push(loose[i]);
2019
            refs.push(packed[i]);
2020
        }
2021
        refs.push(missing);
2022
        let got = s.get_many(&refs);
2023
        assert_eq!(got.len(), refs.len());
2024
        for (i, r) in refs.iter().enumerate().take(40) {
2025
            let want = if i % 2 == 0 {
2026
                format!("loose {}", i / 2)
2027
            } else {
2028
                format!("packed {}", i / 2)
2029
            };
2030
            let (ty, body) = got[i].as_ref().unwrap();
2031
            assert_eq!(
2032
                (*ty, body.as_slice()),
2033
                (ObjectType::Chunk, want.as_bytes()),
2034
                "{r:?}"
2035
            );
2036
        }
2037
        assert!(matches!(got[40], Err(StoreError::NotFound(_))));
a9f686ed Andy McCall ↶ 2038
    }
2039
}
2040
2041
#[cfg(test)]
2042
mod batch_tests {
2043
    use super::*;
2044
2045
    #[test]
2046
    fn a_large_batch_packs_every_object_and_indexes_them_all() {
2047
        let dir = tempfile::tempdir().unwrap();
2048
        let store = Store::open(dir.path()).unwrap();
2049
        let mut refs = Vec::new();
2050
        {
2051
            let batch = store.begin_batch().unwrap();
2052
            for i in 0..1000u32 {
2053
                let body = format!("object {i} {}", "x".repeat(20_000 + i as usize)).into_bytes();
2054
                refs.push(store.put(ObjectType::Blob, &body).unwrap());
2055
            }
2056
            batch.finish().unwrap();
2057
        }
2058
        let again = Store::open(dir.path()).unwrap();
2059
        for r in &refs {
2060
            again
2061
                .get(r)
2062
                .unwrap_or_else(|e| panic!("{}: {e}", r.short(12)));
2063
        }
2064
    }
2065
}
2066
2067
#[cfg(test)]
2068
mod size_tests {
2069
    use super::*;
2070
2071
    #[test]
2072
    fn an_object_over_the_limit_is_refused_and_nothing_is_written() {
2073
        let dir = tempfile::tempdir().unwrap();
2074
        let store = Store::open(dir.path()).unwrap();
2075
        let big = vec![b'x'; MAX_OBJECT_LEN as usize + 1];
2076
        {
2077
            let batch = store.begin_batch().unwrap();
2078
            store.put(ObjectType::Blob, b"small").unwrap();
2079
            let e = store.put(ObjectType::Change, &big).unwrap_err();
2080
            assert!(
2081
                matches!(e, StoreError::TooLarge { kind: "change", .. }),
2082
                "{e}"
2083
            );
2084
            assert!(e.to_string().contains("commit it in parts"), "{e}");
2085
            batch.finish().unwrap();
2086
        }
2087
        // The store still opens and reads what was written.
2088
        let again = Store::open(dir.path()).unwrap();
2089
        again.get(&hash_object(ObjectType::Blob, b"small")).unwrap();
6f8aeb99 Andy McCall ↶ 2090
    }
2091
}
2092
2093
#[cfg(test)]
2094
mod text_tests {
2095
    use super::*;
2096
    use crate::objects::{AtomRef, Edit, CHANGE_VERSION};
2097
2098
    fn large_change() -> Change {
2099
        let mut c = crate::objects::tests::sample_change();
2100
        assert_eq!(c.version, CHANGE_VERSION);
2101
        c.edits[0] = Edit::Insert {
2102
            node: crate::ids::NodeId([7; 16]),
2103
            after: AtomRef::Start,
2104
            before: AtomRef::End,
2105
            lines: (0..2000)
2106
                .map(|i| format!("line {i}\n").into_bytes())
2107
                .collect(),
2108
        };
2109
        c
2110
    }
2111
2112
    #[test]
2113
    fn a_large_insert_is_stored_beside_its_change_and_read_back() {
2114
        let dir = tempfile::tempdir().unwrap();
2115
        let s = Store::open(dir.path()).unwrap();
2116
        let c = large_change();
2117
        let r = s.put_change(&c).unwrap();
2118
        assert_eq!(r, c.hash());
2119
        let (_, body) = s.get(&r).unwrap();
2120
        assert!(
2121
            body.len() < 1024,
2122
            "the change holds a reference, not {} bytes of lines",
2123
            body.len()
2124
        );
2125
        let (back, parts) = s.get_change_parts(&r).unwrap();
2126
        assert_eq!(back, c);
2127
        // The text blob and its chunk.
2128
        assert_eq!(parts.len(), 2);
2129
        for p in &parts {
2130
            assert!(s.has(p));
2131
        }
2132
        // Storing it again writes nothing new.
2133
        let n = s.list().unwrap().len();
2134
        s.put_change(&c).unwrap();
2135
        assert_eq!(s.list().unwrap().len(), n);
2136
    }
2137
2138
    #[test]
2139
    fn a_change_whose_text_blob_is_missing_is_reported_not_misread() {
2140
        let dir = tempfile::tempdir().unwrap();
2141
        let s = Store::open(dir.path()).unwrap();
2142
        let c = large_change();
2143
        let r = s.put_change(&c).unwrap();
2144
        let (_, parts) = s.get_change_parts(&r).unwrap();
2145
        fs::remove_file(s.path_for(&parts[0])).unwrap();
2146
        let e = s.get_change(&r).unwrap_err();
2147
        assert!(
2148
            matches!(&e, StoreError::Corrupt(h, why) if *h == r && why.contains("missing")),
2149
            "{e}"
2150
        );
192fad1e Andy McCall ↶ 2151
    }
2152
}
2153
2154
#[cfg(test)]
2155
mod put_many_tests {
2156
    use super::*;
2157
2158
    #[test]
2159
    fn many_objects_in_a_batch_are_compressed_together_and_all_read_back() {
2160
        let dir = tempfile::tempdir().unwrap();
2161
        let s = Store::open(dir.path()).unwrap();
2162
        let before = s.put(ObjectType::Chunk, b"already here").unwrap();
2163
        let bodies: Vec<Vec<u8>> = (0..200u32)
2164
            .map(|i| format!("chunk {i} ").repeat(6000).into_bytes())
2165
            .collect();
2166
        let mut objects: Vec<(ObjectType, &[u8])> = bodies
2167
            .iter()
2168
            .map(|b| (ObjectType::Chunk, b.as_slice()))
2169
            .collect();
2170
        objects.push((ObjectType::Chunk, bodies[0].as_slice()));
2171
        objects.push((ObjectType::Chunk, b"already here"));
2172
        let refs = {
2173
            let batch = s.begin_batch().unwrap();
2174
            let refs = s.put_many(&objects).unwrap();
2175
            batch.finish().unwrap();
2176
            refs
2177
        };
2178
        assert_eq!(refs.len(), objects.len());
2179
        assert_eq!(refs[200], refs[0], "a repeat is the same object");
2180
        assert_eq!(refs[201], before);
2181
        let again = Store::open(dir.path()).unwrap();
2182
        for (r, b) in refs.iter().zip(&objects) {
2183
            assert_eq!(again.get(r).unwrap().1, b.1);
2184
        }
a17fc622 Andy McCall ↶ 2185
    }
2186
}
2187
2188
/// Remove a pack with its index and dictionary.
2189
fn remove_pack_files(pack: &Path) -> io::Result<()> {
2190
    fs::remove_file(pack)?;
2191
    fs::remove_file(pack.with_extension("idx"))?;
2192
    match fs::remove_file(pack.with_extension("dict")) {
2193
        Err(e) if e.kind() != io::ErrorKind::NotFound => Err(e),
2194
        _ => Ok(()),
8f10a07a Andy McCall ↶ 2195
    }
ff0af1ee unknown 2196
}