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