OXYGENHUB
# a9f686edf60b oxygen / crates / oxy-core / src / store.rs
53.8 KiB · Rust · last changed Refuse objects the store could not read back 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::{Change, ObjectError, StateObject};
15
use rand::RngCore;
16
use std::cell::RefCell;
17
use std::fs;
18
use std::io::{self, Read, Seek, SeekFrom, Write};
19
use std::path::{Path, PathBuf};
20
21
const PACK_MAGIC: &[u8; 8] = b"OXYPACK1";
22
const IDX_MAGIC: &[u8; 8] = b"OXYIDX01";
23
const IDX_ENTRY: usize = 32 + 8 + 4 + 4 + 1 + 1;
24
const FLAG_ZSTD: u8 = 1;
25
const ZSTD_LEVEL: i32 = 3;
26
/// Largest object a pack entry may claim; anything bigger is corruption,
27
/// and refusing it keeps a bad index from driving huge allocations.
28
const MAX_OBJECT_LEN: u32 = 1 << 30;
29
30
/// An object read from the store: its type and body.
31
pub type Object = (ObjectType, Vec<u8>);
32
33
#[derive(Clone, Copy, Debug)]
34
struct PackEntry {
35
    offset: u64,
36
    stored_len: u32,
37
    raw_len: u32,
38
    ty: ObjectType,
39
    flags: u8,
40
}
41
42
/// One pack's index, fully in memory and sorted by ObjectRef.
43
#[derive(Debug)]
44
struct PackIndex {
45
    path: PathBuf,
46
    refs: Vec<ObjectRef>,
47
    entries: Vec<PackEntry>,
48
}
49
50
impl PackIndex {
51
    fn load(idx_path: &Path, pack_path: &Path) -> Result<PackIndex> {
52
        let data = fs::read(idx_path)?;
53
        let bad =
54
            |m: &str| StoreError::Corrupt(ObjectRef::ZERO, format!("{}: {m}", idx_path.display()));
55
        if data.len() < 12 || &data[..8] != IDX_MAGIC {
56
            return Err(bad("bad index header"));
57
        }
58
        let count = u32::from_le_bytes(data[8..12].try_into().unwrap()) as usize;
59
        if data.len() != 12 + count * IDX_ENTRY {
60
            return Err(bad("index length mismatch"));
61
        }
62
        let pack_len = fs::metadata(pack_path)?.len();
63
        let mut refs = Vec::with_capacity(count);
64
        let mut entries = Vec::with_capacity(count);
65
        for i in 0..count {
66
            let e = &data[12 + i * IDX_ENTRY..12 + (i + 1) * IDX_ENTRY];
67
            let r = ObjectRef::from_slice(&e[..32]).unwrap();
68
            if let Some(prev) = refs.last() {
69
                if r <= *prev {
70
                    return Err(bad("index not sorted"));
71
                }
72
            }
73
            refs.push(r);
74
            let entry = PackEntry {
75
                offset: u64::from_le_bytes(e[32..40].try_into().unwrap()),
76
                stored_len: u32::from_le_bytes(e[40..44].try_into().unwrap()),
77
                raw_len: u32::from_le_bytes(e[44..48].try_into().unwrap()),
78
                ty: ObjectType::ALL
79
                    .get(e[48] as usize)
80
                    .copied()
81
                    .ok_or_else(|| bad("unknown object type"))?,
82
                flags: e[49],
83
            };
84
            if entry.stored_len > MAX_OBJECT_LEN || entry.raw_len > MAX_OBJECT_LEN {
85
                return Err(StoreError::Corrupt(
86
                    r,
87
                    format!(
88
                        "{}: the index claims an object of {} bytes ({} stored), more than the {MAX_OBJECT_LEN} an object may be",
89
                        idx_path.display(),
90
                        entry.raw_len,
91
                        entry.stored_len
92
                    ),
93
                ));
94
            }
95
            if entry.offset.saturating_add(entry.stored_len as u64) > pack_len {
96
                return Err(bad("index entry points outside the pack"));
97
            }
98
            entries.push(entry);
99
        }
100
        Ok(PackIndex {
101
            path: pack_path.to_path_buf(),
102
            refs,
103
            entries,
104
        })
105
    }
106
107
    fn find(&self, r: &ObjectRef) -> Option<PackEntry> {
108
        self.refs.binary_search(r).ok().map(|i| self.entries[i])
109
    }
110
111
    fn read(&self, r: &ObjectRef, e: PackEntry) -> Result<(ObjectType, Vec<u8>)> {
112
        let mut f = fs::File::open(&self.path)?;
113
        f.seek(SeekFrom::Start(e.offset))?;
114
        let mut stored = vec![0u8; e.stored_len as usize];
115
        f.read_exact(&mut stored)?;
116
        let body = if e.flags & FLAG_ZSTD != 0 {
117
            zstd::bulk::decompress(&stored, e.raw_len as usize)
118
                .map_err(|err| StoreError::Corrupt(*r, format!("zstd: {err}")))?
119
        } else {
120
            stored
121
        };
122
        if body.len() != e.raw_len as usize || hash_object(e.ty, &body) != *r {
123
            return Err(StoreError::Corrupt(
124
                *r,
125
                "packed object does not match its reference".into(),
126
            ));
127
        }
128
        Ok((e.ty, body))
129
    }
130
}
131
132
fn type_index(ty: ObjectType) -> u8 {
133
    ObjectType::ALL.iter().position(|t| *t == ty).unwrap() as u8
134
}
135
136
/// What `pack_loose` did.
137
#[derive(Debug, Clone, Default, PartialEq, Eq)]
138
pub struct PackReport {
139
    pub objects: usize,
140
    pub raw_bytes: u64,
141
    pub packed_bytes: u64,
142
}
143
144
/// Packs tolerated before `merge_packs` folds them into one (spec §13.4).
145
pub const MAX_PACKS: usize = 16;
146
147
/// What `prune` did.
148
#[derive(Debug, Clone, Default, PartialEq, Eq)]
149
pub struct PruneReport {
150
    pub objects: usize,
151
    pub bytes: u64,
152
    /// Unreachable objects left alone because they are younger than the grace period.
153
    pub deferred: usize,
154
}
155
156
#[derive(Debug, thiserror::Error)]
157
pub enum StoreError {
158
    #[error("io: {0}")]
159
    Io(#[from] io::Error),
160
    #[error("corrupt object {0}: {1}")]
161
    Corrupt(ObjectRef, String),
162
    #[error("object {0}")]
163
    Object(#[from] ObjectError),
164
    #[error("object {0} is a {1}, expected {2}")]
165
    WrongType(ObjectRef, &'static str, &'static str),
166
    #[error("object {0} not found")]
167
    NotFound(ObjectRef),
168
    #[error("a {kind} object of {size} bytes is larger than the {limit} bytes an object may be; nothing was stored. A first commit of a very large tree makes one change holding every text line: commit it in parts (a directory at a time)")]
169
    TooLarge {
170
        kind: &'static str,
171
        size: usize,
172
        limit: u32,
173
    },
174
}
175
176
pub type Result<T> = std::result::Result<T, StoreError>;
177
178
#[derive(Debug)]
179
pub struct Store {
180
    dir: PathBuf,
181
    packs: RefCell<Vec<PackIndex>>,
182
    /// An open write batch (spec §13.4): objects go here instead of to loose
183
    /// files, and are written out with one `fsync` when the batch finishes.
184
    pending: RefCell<Option<Pending>>,
185
    batch_depth: std::cell::Cell<u32>,
186
}
187
188
/// Objects written since `begin_batch`. Small batches stay in memory and
189
/// become loose objects; once `SPILL_OBJECTS` or `SPILL_BYTES` is exceeded
190
/// the batch streams into a pack file instead.
191
#[derive(Debug)]
192
struct Pending {
193
    buffered: Vec<(ObjectType, Vec<u8>)>,
194
    buffered_bytes: usize,
195
    lookup: std::collections::HashMap<ObjectRef, PendingEntry>,
196
    pack: Option<PendingPack>,
197
    report: PackReport,
198
}
199
200
#[derive(Debug, Clone, Copy)]
201
enum PendingEntry {
202
    Buffered(usize),
203
    Packed(PackEntry),
204
}
205
206
#[derive(Debug)]
207
struct PendingPack {
208
    id: String,
209
    tmp_path: PathBuf,
210
    writer: io::BufWriter<fs::File>,
211
    offset: u64,
212
    index: Vec<(ObjectRef, PackEntry)>,
213
}
214
215
const SPILL_OBJECTS: usize = 256;
216
const SPILL_BYTES: usize = 8 << 20;
217
218
/// Finishes the enclosing batch when dropped; call `finish` to see errors.
219
pub struct Batch<'a> {
220
    store: &'a Store,
221
    done: bool,
222
}
223
224
impl Batch<'_> {
225
    pub fn finish(mut self) -> Result<()> {
226
        self.done = true;
227
        self.store.finish_batch()
228
    }
229
}
230
231
impl Drop for Batch<'_> {
232
    fn drop(&mut self) {
233
        if !self.done {
234
            let _ = self.store.finish_batch();
235
        }
236
    }
237
}
238
239
impl Store {
240
    pub fn open(dir: impl Into<PathBuf>) -> io::Result<Store> {
241
        let dir = dir.into();
242
        fs::create_dir_all(dir.join("loose"))?;
243
        let store = Store {
244
            dir,
245
            packs: RefCell::new(Vec::new()),
246
            pending: RefCell::new(None),
247
            batch_depth: std::cell::Cell::new(0),
248
        };
249
        store
250
            .reload_packs()
251
            .map_err(|e| io::Error::other(e.to_string()))?;
252
        Ok(store)
253
    }
254
255
    fn reload_packs(&self) -> Result<()> {
256
        let mut packs = Vec::new();
257
        let mut names: Vec<PathBuf> = fs::read_dir(&self.dir)?
258
            .filter_map(|e| e.ok().map(|e| e.path()))
259
            .filter(|p| {
260
                p.extension().is_some_and(|x| x == "idx")
261
                    && p.file_name()
262
                        .is_some_and(|n| n.to_string_lossy().starts_with("pack-"))
263
            })
264
            .collect();
265
        names.sort();
266
        for idx in names {
267
            let pack = idx.with_extension("pack");
268
            if pack.is_file() {
269
                packs.push(PackIndex::load(&idx, &pack)?);
270
            }
271
        }
272
        *self.packs.borrow_mut() = packs;
273
        Ok(())
274
    }
275
276
    pub fn pack_count(&self) -> usize {
277
        self.packs.borrow().len()
278
    }
279
280
    fn find_packed(&self, r: &ObjectRef) -> Option<(usize, PackEntry)> {
281
        self.packs
282
            .borrow()
283
            .iter()
284
            .enumerate()
285
            .find_map(|(i, p)| p.find(r).map(|e| (i, e)))
286
    }
287
288
    /// Loose objects on disk, with their paths.
289
    fn loose_objects(&self) -> io::Result<Vec<(ObjectRef, PathBuf)>> {
290
        let mut out = Vec::new();
291
        let loose = self.dir.join("loose");
292
        for d in fs::read_dir(&loose)? {
293
            let d = d?;
294
            if !d.file_type()?.is_dir() {
295
                continue;
296
            }
297
            let prefix = d.file_name().to_string_lossy().to_string();
298
            for f in fs::read_dir(d.path())? {
299
                let f = f?;
300
                let name = f.file_name().to_string_lossy().to_string();
301
                if let Some(r) = ObjectRef::from_hex(&format!("{prefix}{name}")) {
302
                    out.push((r, f.path()));
303
                }
304
            }
305
        }
306
        out.sort();
307
        Ok(out)
308
    }
309
310
    pub fn loose_count(&self) -> io::Result<usize> {
311
        Ok(self.loose_objects()?.len())
312
    }
313
314
    /// Move every loose object into a new pack (spec §13.4). Loose files are
315
    /// deleted only after the pack and its index are durable.
316
    pub fn pack_loose(&self) -> Result<PackReport> {
317
        let loose = self.loose_objects()?;
318
        self.write_pack(
319
            &loose.iter().map(|(r, _)| *r).collect::<Vec<_>>(),
320
            &loose.iter().map(|(_, p)| p.clone()).collect::<Vec<_>>(),
321
            &[],
322
        )
323
    }
324
325
    /// Fold every pack into one when there are more than `MAX_PACKS`
326
    /// (spec §13.4). Returns how many packs were merged.
327
    pub fn merge_packs(&self) -> Result<usize> {
328
        let (refs, paths): (Vec<ObjectRef>, Vec<PathBuf>) = {
329
            let packs = self.packs.borrow();
330
            if packs.len() <= MAX_PACKS {
331
                return Ok(0);
332
            }
333
            (
334
                packs.iter().flat_map(|p| p.refs.iter().copied()).collect(),
335
                packs.iter().map(|p| p.path.clone()).collect(),
336
            )
337
        };
338
        let n = paths.len();
339
        self.write_pack(&refs, &[], &paths)?;
340
        Ok(n)
341
    }
342
343
    /// Delete every object not in `keep` (spec §21). Loose objects and packs
344
    /// younger than `grace` are left alone, since a concurrent writer may not
345
    /// have recorded its references yet. Packs holding unreachable objects
346
    /// are rewritten without them. Signature sidecars and seal-index entries
347
    /// of deleted objects go too.
348
    pub fn prune(
349
        &self,
350
        keep: &std::collections::HashSet<ObjectRef>,
351
        grace: std::time::Duration,
352
    ) -> Result<PruneReport> {
353
        let now = std::time::SystemTime::now();
354
        let old_enough = |m: &fs::Metadata| {
355
            m.modified()
356
                .map(|t| now.duration_since(t).unwrap_or_default() >= grace)
357
                .unwrap_or(true)
358
        };
359
        let mut report = PruneReport::default();
360
        // Loose objects.
361
        for (r, path) in self.loose_objects()? {
362
            if keep.contains(&r) {
363
                continue;
364
            }
365
            let meta = fs::metadata(&path)?;
366
            if !old_enough(&meta) {
367
                report.deferred += 1;
368
                continue;
369
            }
370
            report.bytes += meta.len();
371
            report.objects += 1;
372
            fs::remove_file(&path)?;
373
            let _ = fs::remove_dir(path.parent().unwrap());
374
            self.remove_sidecars(&r)?;
375
        }
376
        // Packs: rewrite those that hold unreachable objects.
377
        let mut keep_refs: Vec<ObjectRef> = Vec::new();
378
        let mut old_paths: Vec<PathBuf> = Vec::new();
379
        let mut rewrite = false;
380
        for p in self.packs.borrow().iter() {
381
            let meta = fs::metadata(&p.path)?;
382
            let unreachable = p.refs.iter().filter(|r| !keep.contains(r)).count();
383
            if unreachable == 0 {
384
                continue;
385
            }
386
            if !old_enough(&meta) {
387
                report.deferred += unreachable;
388
                continue;
389
            }
390
            rewrite = true;
391
            old_paths.push(p.path.clone());
392
            for (r, e) in p.refs.iter().zip(p.entries.iter()) {
393
                if keep.contains(r) {
394
                    keep_refs.push(*r);
395
                } else {
396
                    report.objects += 1;
397
                    report.bytes += e.stored_len as u64;
398
                }
399
            }
400
        }
401
        if rewrite {
402
            let dropped: Vec<ObjectRef> = {
403
                let packs = self.packs.borrow();
404
                packs
405
                    .iter()
406
                    .filter(|p| old_paths.contains(&p.path))
407
                    .flat_map(|p| p.refs.iter().copied())
408
                    .filter(|r| !keep.contains(r))
409
                    .collect()
410
            };
411
            self.write_pack(&keep_refs, &[], &old_paths)?;
412
            for r in dropped {
413
                self.remove_sidecars(&r)?;
414
            }
415
        }
416
        Ok(report)
417
    }
418
419
    fn remove_sidecars(&self, r: &ObjectRef) -> Result<()> {
420
        for path in [self.sig_path(r), self.seal_index_path(r)] {
421
            match fs::remove_file(&path) {
422
                Ok(()) => {
423
                    let _ = fs::remove_dir(path.parent().unwrap());
424
                }
425
                Err(e) if e.kind() == io::ErrorKind::NotFound => {}
426
                Err(e) => return Err(e.into()),
427
            }
428
        }
429
        Ok(())
430
    }
431
432
    /// Write `refs` into a new pack, then delete `loose_paths` and `old_packs`.
433
    /// Write the index for a complete, synced temporary pack file and move
434
    /// both into place; `index` must be sorted by ObjectRef.
435
    fn finish_pack_files(
436
        &self,
437
        id: &str,
438
        tmp_pack: &Path,
439
        index: &[(ObjectRef, PackEntry)],
440
    ) -> Result<()> {
441
        let pack_path = self.dir.join(format!("pack-{id}.pack"));
442
        let idx_path = self.dir.join(format!("pack-{id}.idx"));
443
        let tmp_idx = self.dir.join(format!("tmp-{id}.idx"));
444
        {
445
            let mut f = io::BufWriter::new(fs::File::create(&tmp_idx)?);
446
            f.write_all(IDX_MAGIC)?;
447
            f.write_all(&(index.len() as u32).to_le_bytes())?;
448
            for (r, e) in index {
449
                f.write_all(&r.0)?;
450
                f.write_all(&e.offset.to_le_bytes())?;
451
                f.write_all(&e.stored_len.to_le_bytes())?;
452
                f.write_all(&e.raw_len.to_le_bytes())?;
453
                f.write_all(&[type_index(e.ty), e.flags])?;
454
            }
455
            f.flush()?;
456
            f.get_ref().sync_all()?;
457
        }
458
        // The pack first: an index never names a pack that is not there.
459
        crate::durable::rename(tmp_pack, &pack_path)?;
460
        crate::durable::rename(&tmp_idx, &idx_path)?;
461
        self.reload_packs()
462
    }
463
464
    fn write_pack(
465
        &self,
466
        refs: &[ObjectRef],
467
        loose_paths: &[PathBuf],
468
        old_packs: &[PathBuf],
469
    ) -> Result<PackReport> {
470
        if refs.is_empty() {
471
            for p in old_packs {
472
                fs::remove_file(p)?;
473
                fs::remove_file(p.with_extension("idx"))?;
474
            }
475
            self.reload_packs()?;
476
            return Ok(PackReport::default());
477
        }
478
        let mut refs: Vec<ObjectRef> = refs.to_vec();
479
        refs.sort();
480
        refs.dedup();
481
        let mut nonce = [0u8; 8];
482
        rand::rngs::OsRng.fill_bytes(&mut nonce);
483
        let id = hex::encode(nonce);
484
        let tmp_pack = self.dir.join(format!("tmp-{id}.pack"));
485
        let mut report = PackReport::default();
486
        let mut index: Vec<(ObjectRef, PackEntry)> = Vec::with_capacity(refs.len());
487
        {
488
            let mut f = io::BufWriter::new(fs::File::create(&tmp_pack)?);
489
            f.write_all(PACK_MAGIC)?;
490
            let mut offset = PACK_MAGIC.len() as u64;
491
            for r in &refs {
492
                let (ty, body) = self.get(r)?;
493
                let (flags, stored) = if self.is_nocompress(r) {
494
                    (0, body.clone())
495
                } else {
496
                    let compressed = zstd::bulk::compress(&body, ZSTD_LEVEL)
497
                        .map_err(|e| StoreError::Corrupt(*r, format!("zstd: {e}")))?;
498
                    if compressed.len() < body.len() {
499
                        (FLAG_ZSTD, compressed)
500
                    } else {
501
                        (0, body.clone())
502
                    }
503
                };
504
                let header_len = 1 + 1 + 4 + 4 + 32;
505
                f.write_all(&[type_index(ty), flags])?;
506
                f.write_all(&(stored.len() as u32).to_le_bytes())?;
507
                f.write_all(&(body.len() as u32).to_le_bytes())?;
508
                f.write_all(&r.0)?;
509
                f.write_all(&stored)?;
510
                index.push((
511
                    *r,
512
                    PackEntry {
513
                        offset: offset + header_len,
514
                        stored_len: stored.len() as u32,
515
                        raw_len: body.len() as u32,
516
                        ty,
517
                        flags,
518
                    },
519
                ));
520
                offset += header_len + stored.len() as u64;
521
                report.objects += 1;
522
                report.raw_bytes += body.len() as u64;
523
                report.packed_bytes += stored.len() as u64;
524
            }
525
            f.flush()?;
526
            f.get_ref().sync_all()?;
527
        }
528
        self.finish_pack_files(&id, &tmp_pack, &index)?;
529
        for p in old_packs {
530
            fs::remove_file(p)?;
531
            fs::remove_file(p.with_extension("idx"))?;
532
        }
533
        self.reload_packs()?;
534
        for path in loose_paths {
535
            fs::remove_file(path)?;
536
            if let Some(dir) = path.parent() {
537
                let _ = fs::remove_dir(dir);
538
            }
539
        }
540
        Ok(report)
541
    }
542
543
    pub fn dir(&self) -> &Path {
544
        &self.dir
545
    }
546
547
    pub(crate) fn path_for(&self, r: &ObjectRef) -> PathBuf {
548
        let h = r.to_hex();
549
        self.dir.join("loose").join(&h[..2]).join(&h[2..])
550
    }
551
552
    pub fn has(&self, r: &ObjectRef) -> bool {
553
        self.path_for(r).is_file()
554
            || self.find_packed(r).is_some()
555
            || self
556
                .pending
557
                .borrow()
558
                .as_ref()
559
                .is_some_and(|p| p.lookup.contains_key(r))
560
    }
561
562
    /// Start a write batch: until the matching `finish`, `put` collects
563
    /// objects and writes them out together, with one `fsync` for a large
564
    /// batch (a pack) instead of one per object. Batches nest; only the
565
    /// outermost does the work. Objects are readable while pending.
566
    pub fn begin_batch(&self) -> Result<Batch<'_>> {
567
        let depth = self.batch_depth.get();
568
        if depth == 0 {
569
            *self.pending.borrow_mut() = Some(Pending {
570
                buffered: vec![],
571
                buffered_bytes: 0,
572
                lookup: Default::default(),
573
                pack: None,
574
                report: PackReport::default(),
575
            });
576
        }
577
        self.batch_depth.set(depth + 1);
578
        Ok(Batch {
579
            store: self,
580
            done: false,
581
        })
582
    }
583
584
    fn finish_batch(&self) -> Result<()> {
585
        let depth = self.batch_depth.get();
586
        if depth == 0 {
587
            return Ok(());
588
        }
589
        self.batch_depth.set(depth - 1);
590
        if depth > 1 {
591
            return Ok(());
592
        }
593
        let Some(pending) = self.pending.borrow_mut().take() else {
594
            return Ok(());
595
        };
596
        self.write_pending(pending)
597
    }
598
599
    /// Make every object written so far durable, even inside an open
600
    /// batch, which stays open. Call before a reference that names them
601
    /// is moved (spec §13.6): the operation head, a bookmark pointer.
602
    pub fn flush(&self) -> Result<()> {
603
        let Some(pending) = self.pending.borrow_mut().take() else {
604
            return Ok(());
605
        };
606
        self.write_pending(pending)?;
607
        *self.pending.borrow_mut() = Some(Pending {
608
            buffered: vec![],
609
            buffered_bytes: 0,
610
            lookup: Default::default(),
611
            pack: None,
612
            report: PackReport::default(),
613
        });
614
        Ok(())
615
    }
616
617
    fn write_pending(&self, pending: Pending) -> Result<()> {
618
        match pending.pack {
619
            None => {
620
                for (ty, body) in &pending.buffered {
621
                    self.put_loose(*ty, body)?;
622
                }
623
                Ok(())
624
            }
625
            Some(mut pack) => {
626
                pack.writer.flush()?;
627
                pack.writer.get_ref().sync_all()?;
628
                drop(pack.writer);
629
                pack.index.sort_by_key(|(r, _)| *r);
630
                self.finish_pack_files(&pack.id, &pack.tmp_path, &pack.index)?;
631
                Ok(())
632
            }
633
        }
634
    }
635
636
    /// Whether a batch is open and holds `r`.
637
    fn pending_get(&self, r: &ObjectRef) -> Option<Result<(ObjectType, Vec<u8>)>> {
638
        let mut guard = self.pending.borrow_mut();
639
        let pending = guard.as_mut()?;
640
        let entry = *pending.lookup.get(r)?;
641
        Some(match entry {
642
            PendingEntry::Buffered(i) => {
643
                let (ty, body) = &pending.buffered[i];
644
                Ok((*ty, body.clone()))
645
            }
646
            PendingEntry::Packed(e) => {
647
                let Some(pack) = pending.pack.as_mut() else {
648
                    return Some(Err(StoreError::NotFound(*r)));
649
                };
650
                match pack.writer.flush() {
651
                    Ok(()) => PackIndex {
652
                        path: pack.tmp_path.clone(),
653
                        refs: vec![],
654
                        entries: vec![],
655
                    }
656
                    .read(r, e),
657
                    Err(err) => Err(err.into()),
658
                }
659
            }
660
        })
661
    }
662
663
    /// Append one object to the pending pack file.
664
    fn pending_pack_write(
665
        &self,
666
        pack: &mut PendingPack,
667
        r: &ObjectRef,
668
        ty: ObjectType,
669
        body: &[u8],
670
        report: &mut PackReport,
671
    ) -> Result<PackEntry> {
672
        let (flags, stored) = if self.is_nocompress(r) {
673
            (0, body.to_vec())
674
        } else {
675
            let compressed = zstd::bulk::compress(body, ZSTD_LEVEL)
676
                .map_err(|e| StoreError::Corrupt(*r, format!("zstd: {e}")))?;
677
            if compressed.len() < body.len() {
678
                (FLAG_ZSTD, compressed)
679
            } else {
680
                (0, body.to_vec())
681
            }
682
        };
683
        let header_len = 1 + 1 + 4 + 4 + 32;
684
        pack.writer.write_all(&[type_index(ty), flags])?;
685
        pack.writer
686
            .write_all(&(stored.len() as u32).to_le_bytes())?;
687
        pack.writer.write_all(&(body.len() as u32).to_le_bytes())?;
688
        pack.writer.write_all(&r.0)?;
689
        pack.writer.write_all(&stored)?;
690
        let entry = PackEntry {
691
            offset: pack.offset + header_len,
692
            stored_len: stored.len() as u32,
693
            raw_len: body.len() as u32,
694
            ty,
695
            flags,
696
        };
697
        pack.offset += header_len + stored.len() as u64;
698
        pack.index.push((*r, entry));
699
        report.objects += 1;
700
        report.raw_bytes += body.len() as u64;
701
        report.packed_bytes += stored.len() as u64;
702
        Ok(entry)
703
    }
704
705
    fn put_pending(&self, r: ObjectRef, ty: ObjectType, body: &[u8]) -> Result<()> {
706
        let mut guard = self.pending.borrow_mut();
707
        let pending = guard.as_mut().expect("batch open");
708
        if pending.lookup.contains_key(&r) {
709
            return Ok(());
710
        }
711
        if pending.pack.is_none()
712
            && pending.buffered.len() < SPILL_OBJECTS
713
            && pending.buffered_bytes + body.len() <= SPILL_BYTES
714
        {
715
            pending.buffered.push((ty, body.to_vec()));
716
            pending.buffered_bytes += body.len();
717
            pending
718
                .lookup
719
                .insert(r, PendingEntry::Buffered(pending.buffered.len() - 1));
720
            return Ok(());
721
        }
722
        if pending.pack.is_none() {
723
            // Spill: open the pack and move the buffered objects into it.
724
            let mut nonce = [0u8; 8];
725
            rand::rngs::OsRng.fill_bytes(&mut nonce);
726
            let id = hex::encode(nonce);
727
            let tmp_path = self.dir.join(format!("tmp-{id}.pack"));
728
            let mut writer = io::BufWriter::new(fs::File::create(&tmp_path)?);
729
            writer.write_all(PACK_MAGIC)?;
730
            let mut pack = PendingPack {
731
                id,
732
                tmp_path,
733
                writer,
734
                offset: PACK_MAGIC.len() as u64,
735
                index: vec![],
736
            };
737
            let buffered = std::mem::take(&mut pending.buffered);
738
            pending.buffered_bytes = 0;
739
            let mut report = std::mem::take(&mut pending.report);
740
            for (bty, bbody) in &buffered {
741
                let br = hash_object(*bty, bbody);
742
                let e = self.pending_pack_write(&mut pack, &br, *bty, bbody, &mut report)?;
743
                pending.lookup.insert(br, PendingEntry::Packed(e));
744
            }
745
            pending.report = report;
746
            pending.pack = Some(pack);
747
        }
748
        let mut report = std::mem::take(&mut pending.report);
749
        let pack = pending.pack.as_mut().unwrap();
750
        let e = self.pending_pack_write(pack, &r, ty, body, &mut report)?;
751
        pending.report = report;
752
        pending.lookup.insert(r, PendingEntry::Packed(e));
753
        Ok(())
754
    }
755
756
    /// Store `body` as an object of type `ty`; idempotent.
757
    pub fn put(&self, ty: ObjectType, body: &[u8]) -> Result<ObjectRef> {
758
        // Refuse what the store could not read back (spec §13.2), before
759
        // anything is written: a pack must never hold an entry its index
760
        // would reject.
761
        if body.len() > MAX_OBJECT_LEN as usize {
762
            return Err(StoreError::TooLarge {
763
                kind: ty.name(),
764
                size: body.len(),
765
                limit: MAX_OBJECT_LEN,
766
            });
767
        }
768
        let r = hash_object(ty, body);
769
        let path = self.path_for(&r);
770
        if path.is_file() || self.find_packed(&r).is_some() {
771
            return Ok(r);
772
        }
773
        if self.pending.borrow().is_some() {
774
            self.put_pending(r, ty, body)?;
775
            return Ok(r);
776
        }
777
        self.put_loose(ty, body)?;
778
        Ok(r)
779
    }
780
781
    /// Write one loose object atomically with its own `fsync`.
782
    fn put_loose(&self, ty: ObjectType, body: &[u8]) -> Result<ObjectRef> {
783
        let r = hash_object(ty, body);
784
        let path = self.path_for(&r);
785
        if path.is_file() {
786
            return Ok(r);
787
        }
788
        let fanout = path.parent().unwrap();
789
        if !fanout.is_dir() {
790
            fs::create_dir_all(fanout)?;
791
            crate::durable::sync_dir(&self.dir.join("loose"))?;
792
        }
793
        let mut nonce = [0u8; 8];
794
        rand::rngs::OsRng.fill_bytes(&mut nonce);
795
        let tmp = self
796
            .dir
797
            .join("loose")
798
            .join(format!("tmp-{}", hex::encode(nonce)));
799
        crate::durable::step("write loose object")?;
800
        {
801
            let mut f = fs::File::create(&tmp)?;
802
            f.write_all(&header(ty, body.len()))?;
803
            f.write_all(body)?;
804
            f.sync_all()?;
805
        }
806
        crate::durable::rename(&tmp, &path)?;
807
        Ok(r)
808
    }
809
810
    /// Read and verify an object, loose, packed, or pending in a batch.
811
    pub fn get(&self, r: &ObjectRef) -> Result<(ObjectType, Vec<u8>)> {
812
        match read_loose_at(&self.path_for(r), r) {
813
            Some(found) => found,
814
            None => match self.find_packed(r) {
815
                Some((i, e)) => self.packs.borrow()[i].read(r, e),
816
                None => self.pending_get(r).unwrap_or(Err(StoreError::NotFound(*r))),
817
            },
818
        }
819
    }
820
821
    /// Many objects at once: loose ones are read and verified in parallel,
822
    /// the rest (packed, or pending in a batch) one by one. Results are in
823
    /// the order of `refs`.
824
    pub fn get_many(&self, refs: &[ObjectRef]) -> Vec<Result<Object>> {
825
        use rayon::prelude::*;
826
        let paths: Vec<(ObjectRef, PathBuf)> =
827
            refs.iter().map(|r| (*r, self.path_for(r))).collect();
828
        let mut found: Vec<Option<Result<Object>>> =
829
            paths.par_iter().map(|(r, p)| read_loose_at(p, r)).collect();
830
        // Packed ones: look them up in order, then read them in parallel.
831
        let packs = self.packs.borrow();
832
        let packed: Vec<(usize, usize, PackEntry)> = found
833
            .iter()
834
            .enumerate()
835
            .filter(|(_, f)| f.is_none())
836
            .filter_map(|(i, _)| {
837
                packs
838
                    .iter()
839
                    .enumerate()
840
                    .find_map(|(p, pack)| pack.find(&refs[i]).map(|e| (i, p, e)))
841
            })
842
            .collect();
843
        let indexes: &[PackIndex] = &packs;
844
        let read: Vec<(usize, Result<Object>)> = packed
845
            .par_iter()
846
            .map(|(i, p, e)| (*i, indexes[*p].read(&refs[*i], *e)))
847
            .collect();
848
        drop(packs);
849
        for (i, r) in read {
850
            found[i] = Some(r);
851
        }
852
        found
853
            .into_iter()
854
            .zip(refs)
855
            .map(|(f, r)| match f {
856
                Some(f) => f,
857
                None => self.get(r),
858
            })
859
            .collect()
860
    }
861
862
    pub fn get_typed(&self, r: &ObjectRef, want: ObjectType) -> Result<Vec<u8>> {
863
        let (ty, body) = self.get(r)?;
864
        if ty != want {
865
            return Err(StoreError::WrongType(*r, ty.name(), want.name()));
866
        }
867
        Ok(body)
868
    }
869
870
    pub fn put_change(&self, c: &Change) -> Result<ObjectRef> {
871
        self.put(ObjectType::Change, &c.encode())
872
    }
873
874
    pub fn get_change(&self, r: &ObjectRef) -> Result<Change> {
875
        Ok(Change::decode(&self.get_typed(r, ObjectType::Change)?)?)
876
    }
877
878
    pub fn put_chunk(&self, bytes: &[u8]) -> Result<ObjectRef> {
879
        self.put(ObjectType::Chunk, bytes)
880
    }
881
882
    /// Store a chunk that packs will keep uncompressed (`nocompress`, spec §13.3).
883
    pub fn put_chunk_nocompress(&self, bytes: &[u8]) -> Result<ObjectRef> {
884
        let r = hash_object(ObjectType::Chunk, bytes);
885
        let dir = self.dir.join("nocompress");
886
        fs::create_dir_all(&dir)?;
887
        let _ = fs::File::create(dir.join(r.to_hex()));
888
        self.put(ObjectType::Chunk, bytes)
889
    }
890
891
    fn is_nocompress(&self, r: &ObjectRef) -> bool {
892
        self.dir.join("nocompress").join(r.to_hex()).exists()
893
    }
894
895
    pub fn put_state(&self, st: &StateObject) -> Result<ObjectRef> {
896
        self.put(ObjectType::State, &st.encode())
897
    }
898
899
    pub fn get_state(&self, r: &ObjectRef) -> Result<StateObject> {
900
        Ok(StateObject::decode(&self.get_typed(r, ObjectType::State)?)?)
901
    }
902
903
    pub fn put_attest(&self, a: &Attest) -> Result<ObjectRef> {
904
        self.put(ObjectType::Attest, &a.encode())
905
    }
906
907
    pub fn get_attest(&self, r: &ObjectRef) -> Result<Attest> {
908
        Ok(Attest::decode(&self.get_typed(r, ObjectType::Attest)?)?)
909
    }
910
911
    fn seal_index_path(&self, change: &ObjectRef) -> PathBuf {
912
        let h = change.to_hex();
913
        self.dir.join("seals").join(&h[..2]).join(&h[2..])
914
    }
915
916
    /// Record that seal attestation `attest` covers `change`.
917
    pub fn index_seal(&self, change: &ObjectRef, attest: &ObjectRef) -> Result<()> {
918
        let mut seals = self.seals_for(change)?;
919
        if seals.contains(attest) {
920
            return Ok(());
921
        }
922
        seals.push(*attest);
923
        seals.sort();
924
        let path = self.seal_index_path(change);
925
        fs::create_dir_all(path.parent().unwrap())?;
926
        let text: String = seals.iter().map(|s| format!("{}\n", s.to_hex())).collect();
927
        crate::durable::write_atomic(&path, text.as_bytes())?;
928
        Ok(())
929
    }
930
931
    /// Seal attestations known to cover `change`.
932
    pub fn seals_for(&self, change: &ObjectRef) -> Result<Vec<ObjectRef>> {
933
        match fs::read_to_string(self.seal_index_path(change)) {
934
            Ok(t) => Ok(t.lines().filter_map(ObjectRef::from_hex).collect()),
935
            Err(e) if e.kind() == io::ErrorKind::NotFound => Ok(vec![]),
936
            Err(e) => Err(e.into()),
937
        }
938
    }
939
940
    pub fn put_identity(&self, id: &Identity) -> Result<ObjectRef> {
941
        self.put(ObjectType::Identity, &id.encode())
942
    }
943
944
    pub fn get_identity(&self, r: &ObjectRef) -> Result<Identity> {
945
        Ok(Identity::decode(&self.get_typed(r, ObjectType::Identity)?)?)
946
    }
947
948
    fn sig_path(&self, r: &ObjectRef) -> PathBuf {
949
        let h = r.to_hex();
950
        self.dir.join("sigs").join(&h[..2]).join(&h[2..])
951
    }
952
953
    /// Signatures stored alongside object `r` (spec §5.4). Empty if none.
954
    pub fn signatures(&self, r: &ObjectRef) -> Result<Vec<Signature>> {
955
        match fs::read(self.sig_path(r)) {
956
            Ok(b) => Ok(Signature::decode_list(&b)?),
957
            Err(e) if e.kind() == io::ErrorKind::NotFound => Ok(vec![]),
958
            Err(e) => Err(e.into()),
959
        }
960
    }
961
962
    /// Attach a signature to object `r`, replacing any earlier one by the same key.
963
    pub fn put_signature(&self, r: &ObjectRef, sig: &Signature) -> Result<()> {
964
        let mut sigs = self.signatures(r)?;
965
        sigs.retain(|s| s.key != sig.key);
966
        sigs.push(sig.clone());
967
        sigs.sort_by_key(|s| s.key);
968
        let path = self.sig_path(r);
969
        fs::create_dir_all(path.parent().unwrap())?;
970
        let mut nonce = [0u8; 8];
971
        rand::rngs::OsRng.fill_bytes(&mut nonce);
972
        let tmp = self.dir.join(format!("sigs-tmp-{}", hex::encode(nonce)));
973
        {
974
            let mut f = fs::File::create(&tmp)?;
975
            f.write_all(&Signature::encode_list(&sigs))?;
976
            f.sync_all()?;
977
        }
978
        crate::durable::rename(&tmp, &path)?;
979
        Ok(())
980
    }
981
982
    /// Every object reference in the store, loose and packed (slow; for gc and tests).
983
    pub fn list(&self) -> io::Result<Vec<ObjectRef>> {
984
        let mut out: Vec<ObjectRef> = self.loose_objects()?.into_iter().map(|(r, _)| r).collect();
985
        for p in self.packs.borrow().iter() {
986
            out.extend(p.refs.iter().copied());
987
        }
988
        out.sort();
989
        out.dedup();
990
        Ok(out)
991
    }
992
}
993
994
/// A loose object at `path`, verified against `r`; `None` when there is no
995
/// such file (it may be packed).
996
fn read_loose_at(path: &Path, r: &ObjectRef) -> Option<Result<Object>> {
997
    let data = match fs::read(path) {
998
        Ok(d) => d,
999
        Err(e) if e.kind() == io::ErrorKind::NotFound => return None,
1000
        Err(e) => return Some(Err(e.into())),
1001
    };
1002
    Some((|| {
1003
        let nul = data
1004
            .iter()
1005
            .position(|&b| b == 0)
1006
            .ok_or_else(|| StoreError::Corrupt(*r, "no header".into()))?;
1007
        let head = std::str::from_utf8(&data[..nul])
1008
            .map_err(|_| StoreError::Corrupt(*r, "header not UTF-8".into()))?;
1009
        let mut parts = head.split(' ');
1010
        let (Some("oxy1"), Some(ty), Some(len), None) =
1011
            (parts.next(), parts.next(), parts.next(), parts.next())
1012
        else {
1013
            return Err(StoreError::Corrupt(*r, format!("bad header {head:?}")));
1014
        };
1015
        let ty = ObjectType::from_name(ty)
1016
            .ok_or_else(|| StoreError::Corrupt(*r, format!("unknown type {ty}")))?;
1017
        let len: usize = len
1018
            .parse()
1019
            .map_err(|_| StoreError::Corrupt(*r, "bad length".into()))?;
1020
        let body = &data[nul + 1..];
1021
        if body.len() != len {
1022
            return Err(StoreError::Corrupt(
1023
                *r,
1024
                format!("length {} != header {len}", body.len()),
1025
            ));
1026
        }
1027
        if hash_object(ty, body) != *r {
1028
            return Err(StoreError::Corrupt(*r, "hash mismatch".into()));
1029
        }
1030
        Ok((ty, body.to_vec()))
1031
    })())
1032
}
1033
1034
#[cfg(test)]
1035
mod tests {
1036
    use super::*;
1037
1038
    #[test]
1039
    fn put_get_verify() {
1040
        let dir = tempfile::tempdir().unwrap();
1041
        let s = Store::open(dir.path().join("objects")).unwrap();
1042
        let r = s.put(ObjectType::Chunk, b"hello").unwrap();
1043
        assert!(s.has(&r));
1044
        assert_eq!(s.get(&r).unwrap(), (ObjectType::Chunk, b"hello".to_vec()));
1045
        assert_eq!(s.put(ObjectType::Chunk, b"hello").unwrap(), r);
1046
        assert_eq!(s.list().unwrap(), vec![r]);
1047
        assert!(matches!(
1048
            s.get_typed(&r, ObjectType::Blob),
1049
            Err(StoreError::WrongType(..))
1050
        ));
1051
        assert!(matches!(
1052
            s.get(&ObjectRef([9; 32])),
1053
            Err(StoreError::NotFound(_))
1054
        ));
1055
1056
        // Tamper: flip a byte, read must fail.
1057
        let path = s.path_for(&r);
1058
        let mut data = fs::read(&path).unwrap();
1059
        let last = data.len() - 1;
1060
        data[last] ^= 1;
1061
        fs::write(&path, data).unwrap();
1062
        assert!(matches!(s.get(&r), Err(StoreError::Corrupt(..))));
1063
    }
1064
1065
    #[test]
1066
    fn packing_keeps_every_object_readable() {
1067
        let dir = tempfile::tempdir().unwrap();
1068
        let s = Store::open(dir.path()).unwrap();
1069
        let mut refs = Vec::new();
1070
        for i in 0..300u32 {
1071
            let body = format!("object {i} ").repeat(1 + (i % 7) as usize);
1072
            refs.push((s.put(ObjectType::Chunk, body.as_bytes()).unwrap(), body));
1073
        }
1074
        let c = crate::objects::tests::sample_change();
1075
        let cr = s.put_change(&c).unwrap();
1076
        assert_eq!(s.loose_count().unwrap(), 301);
1077
        let report = s.pack_loose().unwrap();
1078
        assert_eq!(report.objects, 301);
1079
        assert!(report.packed_bytes < report.raw_bytes, "{report:?}");
1080
        assert_eq!(s.loose_count().unwrap(), 0);
1081
        assert_eq!(s.pack_count(), 1);
1082
        for (r, body) in &refs {
1083
            assert!(s.has(r));
1084
            assert_eq!(
1085
                s.get(r).unwrap(),
1086
                (ObjectType::Chunk, body.as_bytes().to_vec())
1087
            );
1088
        }
1089
        assert_eq!(s.get_change(&cr).unwrap(), c);
1090
        assert_eq!(s.list().unwrap().len(), 301);
1091
        // Re-putting a packed object is a no-op; new objects go loose and a
1092
        // second pack coexists with the first.
1093
        assert_eq!(
1094
            s.put(ObjectType::Chunk, refs[0].1.as_bytes()).unwrap(),
1095
            refs[0].0
1096
        );
1097
        assert_eq!(s.loose_count().unwrap(), 0);
1098
        let extra = s.put(ObjectType::Chunk, b"after packing").unwrap();
1099
        s.pack_loose().unwrap();
1100
        assert_eq!(s.pack_count(), 2);
1101
        assert_eq!(s.get(&extra).unwrap().1, b"after packing");
1102
        // A reopened store sees the packs.
1103
        let s2 = Store::open(dir.path()).unwrap();
1104
        assert_eq!(s2.pack_count(), 2);
1105
        assert_eq!(s2.get(&refs[5].0).unwrap().1, refs[5].1.as_bytes());
1106
        assert!(matches!(
1107
            s2.get(&ObjectRef([9; 32])),
1108
            Err(StoreError::NotFound(_))
1109
        ));
1110
        // Corrupting an object's body inside the pack is detected on read.
1111
        let target = refs[5].0;
1112
        let (pack_path, offset) = {
1113
            let packs = s2.packs.borrow();
1114
            let (i, e) = s2.find_packed(&target).unwrap();
1115
            (packs[i].path.clone(), e.offset as usize)
1116
        };
1117
        let mut data = fs::read(&pack_path).unwrap();
1118
        data[offset] ^= 0xff;
1119
        fs::write(&pack_path, data).unwrap();
1120
        let s3 = Store::open(dir.path()).unwrap();
1121
        assert!(matches!(s3.get(&target), Err(StoreError::Corrupt(..))));
1122
        assert_eq!(
1123
            s3.get(&refs[6].0).unwrap().1,
1124
            refs[6].1.as_bytes(),
1125
            "other objects are unaffected"
1126
        );
1127
    }
1128
1129
    #[test]
1130
    fn nocompress_chunks_pack_raw() {
1131
        let dir = tempfile::tempdir().unwrap();
1132
        let s = Store::open(dir.path()).unwrap();
1133
        let body = vec![b'a'; 10_000];
1134
        let raw = s.put_chunk_nocompress(&body).unwrap();
1135
        let compressible = s.put(ObjectType::Chunk, &vec![b'b'; 10_000]).unwrap();
1136
        s.pack_loose().unwrap();
1137
        let packs = s.packs.borrow();
1138
        let (i, e) = s.find_packed(&raw).unwrap();
1139
        assert_eq!((e.flags, e.stored_len as usize), (0, 10_000));
1140
        let (_, c) = s.find_packed(&compressible).unwrap();
1141
        assert_eq!(c.flags, FLAG_ZSTD);
1142
        assert!(c.stored_len < 10_000);
1143
        assert_eq!(packs[i].read(&raw, e).unwrap().1, body);
1144
    }
1145
1146
    #[test]
1147
    fn many_packs_merge_into_one() {
1148
        let dir = tempfile::tempdir().unwrap();
1149
        let s = Store::open(dir.path()).unwrap();
1150
        let mut refs = Vec::new();
1151
        for i in 0..(MAX_PACKS + 2) {
1152
            refs.push(
1153
                s.put(ObjectType::Chunk, format!("pack {i}").as_bytes())
1154
                    .unwrap(),
1155
            );
1156
            s.pack_loose().unwrap();
1157
        }
1158
        assert_eq!(s.pack_count(), MAX_PACKS + 2);
1159
        assert_eq!(s.merge_packs().unwrap(), MAX_PACKS + 2);
1160
        assert_eq!(s.pack_count(), 1);
1161
        assert_eq!(s.merge_packs().unwrap(), 0);
1162
        for (i, r) in refs.iter().enumerate() {
1163
            assert_eq!(s.get(r).unwrap().1, format!("pack {i}").as_bytes());
1164
        }
1165
        assert_eq!(Store::open(dir.path()).unwrap().pack_count(), 1);
1166
    }
1167
1168
    #[test]
1169
    fn prune_removes_unreachable_loose_and_packed_objects() {
1170
        use std::collections::HashSet;
1171
        let dir = tempfile::tempdir().unwrap();
1172
        let s = Store::open(dir.path()).unwrap();
1173
        let keep_a = s.put(ObjectType::Chunk, b"keep a").unwrap();
1174
        let drop_a = s.put(ObjectType::Chunk, b"drop a").unwrap();
1175
        s.pack_loose().unwrap();
1176
        let keep_b = s.put(ObjectType::Chunk, b"keep b").unwrap();
1177
        let drop_b = s.put(ObjectType::Chunk, b"drop b").unwrap();
1178
        let sk = crate::identity::SecretKey {
1179
            ed_seed: [1; 32],
1180
            ml_seed: [2; 32],
1181
        };
1182
        let id = sk.identity("a", "b", 0);
1183
        let key = s.put_identity(&id).unwrap();
1184
        s.put_signature(
1185
            &drop_b,
1186
            &sk.sign(crate::identity::Domain::Change, &key, &drop_b),
1187
        )
1188
        .unwrap();
1189
        s.index_seal(&drop_a, &ObjectRef([3; 32])).unwrap();
1190
        let keep: HashSet<ObjectRef> = [keep_a, keep_b, key].into_iter().collect();
1191
        // Everything is too young under a grace period.
1192
        let r = s
1193
            .prune(&keep, std::time::Duration::from_secs(3600))
1194
            .unwrap();
1195
        assert_eq!((r.objects, r.deferred), (0, 2));
1196
        let r = s.prune(&keep, std::time::Duration::ZERO).unwrap();
1197
        assert_eq!(r.objects, 2, "{r:?}");
1198
        assert!(s.has(&keep_a) && s.has(&keep_b) && s.has(&key));
1199
        assert!(!s.has(&drop_a) && !s.has(&drop_b));
1200
        assert!(s.signatures(&drop_b).unwrap().is_empty());
1201
        assert!(s.seals_for(&drop_a).unwrap().is_empty());
1202
        assert_eq!(s.pack_count(), 1);
1203
        assert_eq!(s.get(&keep_a).unwrap().1, b"keep a");
1204
        assert_eq!(s.list().unwrap().len(), 3);
1205
    }
1206
1207
    #[test]
1208
    fn signatures_are_stored_beside_objects() {
1209
        use crate::identity::{Domain, SecretKey};
1210
        let dir = tempfile::tempdir().unwrap();
1211
        let s = Store::open(dir.path()).unwrap();
1212
        let r = s.put(ObjectType::Chunk, b"x").unwrap();
1213
        assert!(s.signatures(&r).unwrap().is_empty());
1214
        let sk = SecretKey {
1215
            ed_seed: [1; 32],
1216
            ml_seed: [2; 32],
1217
        };
1218
        let id = sk.identity("a", "b", 0);
1219
        let key = s.put_identity(&id).unwrap();
1220
        assert_eq!(s.get_identity(&key).unwrap(), id);
1221
        let sig = sk.sign(Domain::Change, &key, &r);
1222
        s.put_signature(&r, &sig).unwrap();
1223
        s.put_signature(&r, &sig).unwrap();
1224
        assert_eq!(s.signatures(&r).unwrap(), vec![sig]);
1225
        // The object itself is untouched by its signatures.
1226
        assert_eq!(s.get(&r).unwrap().1, b"x");
1227
    }
1228
1229
    #[test]
1230
    fn change_round_trip_through_store() {
1231
        let dir = tempfile::tempdir().unwrap();
1232
        let s = Store::open(dir.path()).unwrap();
1233
        let c = crate::objects::tests::sample_change();
1234
        let r = s.put_change(&c).unwrap();
1235
        assert_eq!(r, c.hash());
1236
        assert_eq!(s.get_change(&r).unwrap(), c);
1237
    }
1238
1239
    fn xorshift(state: &mut u64) -> u64 {
1240
        *state ^= *state << 13;
1241
        *state ^= *state >> 7;
1242
        *state ^= *state << 17;
1243
        *state
1244
    }
1245
1246
    /// Random corruption of pack and index files never panics: reads fail
1247
    /// with an error or the store refuses to open, and untouched objects
1248
    /// stay readable when the index survives.
1249
    #[test]
1250
    fn corrupted_packs_and_indexes_fail_cleanly() {
1251
        let dir = tempfile::tempdir().unwrap();
1252
        let objects = dir.path().join("objects");
1253
        let s = Store::open(&objects).unwrap();
1254
        let mut refs = Vec::new();
1255
        let mut seed = 0x9e3779b97f4a7c15u64;
1256
        for i in 0..40u32 {
1257
            let len = (xorshift(&mut seed) % 5000) as usize;
1258
            let mut body = vec![0u8; len];
1259
            for b in body.iter_mut() {
1260
                *b = if i % 2 == 0 {
1261
                    (xorshift(&mut seed) & 0xff) as u8
1262
                } else {
1263
                    b'a' + (i % 26) as u8
1264
                };
1265
            }
1266
            refs.push(s.put(ObjectType::Chunk, &body).unwrap());
1267
        }
1268
        s.pack_loose().unwrap();
1269
        let pack = fs::read_dir(&objects)
1270
            .unwrap()
1271
            .filter_map(|e| e.ok().map(|e| e.path()))
1272
            .find(|p| p.extension().is_some_and(|x| x == "pack"))
1273
            .unwrap();
1274
        let idx = pack.with_extension("idx");
1275
        let pristine_pack = fs::read(&pack).unwrap();
1276
        let pristine_idx = fs::read(&idx).unwrap();
1277
1278
        for round in 0..300u32 {
1279
            let (path, pristine) = if round % 2 == 0 {
1280
                (&pack, &pristine_pack)
1281
            } else {
1282
                (&idx, &pristine_idx)
1283
            };
1284
            let mut data = pristine.clone();
1285
            match xorshift(&mut seed) % 4 {
1286
                0 => {
1287
                    let at = (xorshift(&mut seed) as usize) % data.len();
1288
                    data[at] ^= 1 << (xorshift(&mut seed) % 8);
1289
                }
1290
                1 => {
1291
                    let at = (xorshift(&mut seed) as usize) % data.len();
1292
                    data[at] = (xorshift(&mut seed) & 0xff) as u8;
1293
                }
1294
                2 => {
1295
                    let keep = (xorshift(&mut seed) as usize) % data.len();
1296
                    data.truncate(keep);
1297
                }
1298
                _ => {
1299
                    let at = (xorshift(&mut seed) as usize) % data.len();
1300
                    let n = ((xorshift(&mut seed) % 8) + 1) as usize;
1301
                    for k in 0..n.min(data.len() - at) {
1302
                        data[at + k] = 0xff;
1303
                    }
1304
                }
1305
            }
1306
            fs::write(path, &data).unwrap();
1307
            // Opening may refuse a broken index; a store that opens must
1308
            // either return each object intact or report it corrupt/missing.
1309
            if let Ok(s2) = Store::open(&objects) {
1310
                for r in &refs {
1311
                    match s2.get(r) {
1312
                        Ok((ty, body)) => assert_eq!(
1313
                            hash_object(ty, &body),
1314
                            *r,
1315
                            "a returned object always matches its reference"
1316
                        ),
1317
                        Err(StoreError::Corrupt(..))
1318
                        | Err(StoreError::NotFound(_))
1319
                        | Err(StoreError::Io(_)) => {}
1320
                        Err(e) => panic!("unexpected error {e}"),
1321
                    }
1322
                }
1323
            }
1324
            fs::write(path, pristine).unwrap();
1325
        }
1326
        // Restored files read back in full.
1327
        let s3 = Store::open(&objects).unwrap();
1328
        for r in &refs {
1329
            let (ty, body) = s3.get(r).unwrap();
1330
            assert_eq!(hash_object(ty, &body), *r);
1331
        }
1332
    }
1333
1334
    /// A small batch becomes loose objects; a large one becomes a single
1335
    /// pack. Objects are readable while pending, and batches nest.
1336
    #[test]
1337
    fn batches_write_loose_or_one_pack() {
1338
        let dir = tempfile::tempdir().unwrap();
1339
        let s = Store::open(dir.path().join("objects")).unwrap();
1340
        let outer = s.begin_batch().unwrap();
1341
        let a = s.put(ObjectType::Chunk, b"small a").unwrap();
1342
        {
1343
            let inner = s.begin_batch().unwrap();
1344
            let b = s.put(ObjectType::Chunk, b"small b").unwrap();
1345
            assert!(s.has(&b));
1346
            assert_eq!(s.get(&b).unwrap().1, b"small b");
1347
            inner.finish().unwrap();
1348
            assert_eq!(
1349
                s.loose_count().unwrap(),
1350
                0,
1351
                "the inner batch defers to the outer one"
1352
            );
1353
        }
1354
        assert_eq!(s.get(&a).unwrap().1, b"small a");
1355
        assert_eq!(
1356
            s.put(ObjectType::Chunk, b"small a").unwrap(),
1357
            a,
1358
            "idempotent while pending"
1359
        );
1360
        outer.finish().unwrap();
1361
        assert_eq!(s.loose_count().unwrap(), 2);
1362
        assert_eq!(s.pack_count(), 0);
1363
        assert_eq!(s.get(&a).unwrap().1, b"small a");
1364
1365
        // Large: spills into one pack, readable during and after.
1366
        let big = s.begin_batch().unwrap();
1367
        let mut refs = Vec::new();
1368
        let mut seed = 7u64;
1369
        for i in 0..300u32 {
1370
            let mut body = vec![b'z'; 40_000];
1371
            if i % 3 == 0 {
1372
                for b in body.iter_mut() {
1373
                    *b = (xorshift(&mut seed) & 0xff) as u8;
1374
                }
1375
            }
1376
            body[..4].copy_from_slice(&i.to_le_bytes());
1377
            let r = if i % 5 == 0 {
1378
                s.put_chunk_nocompress(&body).unwrap()
1379
            } else {
1380
                s.put(ObjectType::Chunk, &body).unwrap()
1381
            };
1382
            assert_eq!(
1383
                s.get(&r).unwrap().1,
1384
                body,
1385
                "readable while pending (object {i})"
1386
            );
1387
            refs.push((r, body));
1388
        }
1389
        assert!(s.has(&refs[0].0));
1390
        big.finish().unwrap();
1391
        assert_eq!(s.pack_count(), 1);
1392
        assert_eq!(
1393
            s.loose_count().unwrap(),
1394
            2,
1395
            "nothing from the big batch is loose"
1396
        );
1397
        assert!(fs::read_dir(dir.path().join("objects"))
1398
            .unwrap()
1399
            .flatten()
1400
            .all(|e| !e.file_name().to_string_lossy().starts_with("tmp-")));
1401
        let s2 = Store::open(dir.path().join("objects")).unwrap();
1402
        for (r, body) in &refs {
1403
            assert_eq!(s2.get(r).unwrap().1, *body);
1404
        }
1405
        // A dropped batch still writes its objects.
1406
        {
1407
            let _b = s2.begin_batch().unwrap();
1408
            s2.put(ObjectType::Chunk, b"dropped").unwrap();
1409
        }
1410
        assert_eq!(
1411
            s2.get(&hash_object(ObjectType::Chunk, b"dropped"))
1412
                .unwrap()
1413
                .1,
1414
            b"dropped"
1415
        );
1416
    }
1417
1418
    #[test]
1419
    fn get_many_reads_loose_and_packed_objects_in_order() {
1420
        let dir = tempfile::tempdir().unwrap();
1421
        let s = Store::open(dir.path().join("objects")).unwrap();
1422
        let packed: Vec<ObjectRef> = (0..20)
1423
            .map(|i| {
1424
                s.put(ObjectType::Chunk, format!("packed {i}").as_bytes())
1425
                    .unwrap()
1426
            })
1427
            .collect();
1428
        s.pack_loose().unwrap();
1429
        let loose: Vec<ObjectRef> = (0..20)
1430
            .map(|i| {
1431
                s.put(ObjectType::Chunk, format!("loose {i}").as_bytes())
1432
                    .unwrap()
1433
            })
1434
            .collect();
1435
        let missing = hash_object(ObjectType::Chunk, b"never stored");
1436
        let mut refs: Vec<ObjectRef> = Vec::new();
1437
        for i in 0..20 {
1438
            refs.push(loose[i]);
1439
            refs.push(packed[i]);
1440
        }
1441
        refs.push(missing);
1442
        let got = s.get_many(&refs);
1443
        assert_eq!(got.len(), refs.len());
1444
        for (i, r) in refs.iter().enumerate().take(40) {
1445
            let want = if i % 2 == 0 {
1446
                format!("loose {}", i / 2)
1447
            } else {
1448
                format!("packed {}", i / 2)
1449
            };
1450
            let (ty, body) = got[i].as_ref().unwrap();
1451
            assert_eq!(
1452
                (*ty, body.as_slice()),
1453
                (ObjectType::Chunk, want.as_bytes()),
1454
                "{r:?}"
1455
            );
1456
        }
1457
        assert!(matches!(got[40], Err(StoreError::NotFound(_))));
1458
    }
1459
}
1460
1461
#[cfg(test)]
1462
mod batch_tests {
1463
    use super::*;
1464
1465
    #[test]
1466
    fn a_large_batch_packs_every_object_and_indexes_them_all() {
1467
        let dir = tempfile::tempdir().unwrap();
1468
        let store = Store::open(dir.path()).unwrap();
1469
        let mut refs = Vec::new();
1470
        {
1471
            let batch = store.begin_batch().unwrap();
1472
            for i in 0..1000u32 {
1473
                let body = format!("object {i} {}", "x".repeat(20_000 + i as usize)).into_bytes();
1474
                refs.push(store.put(ObjectType::Blob, &body).unwrap());
1475
            }
1476
            batch.finish().unwrap();
1477
        }
1478
        let again = Store::open(dir.path()).unwrap();
1479
        for r in &refs {
1480
            again
1481
                .get(r)
1482
                .unwrap_or_else(|e| panic!("{}: {e}", r.short(12)));
1483
        }
1484
    }
1485
}
1486
1487
#[cfg(test)]
1488
mod size_tests {
1489
    use super::*;
1490
1491
    #[test]
1492
    fn an_object_over_the_limit_is_refused_and_nothing_is_written() {
1493
        let dir = tempfile::tempdir().unwrap();
1494
        let store = Store::open(dir.path()).unwrap();
1495
        let big = vec![b'x'; MAX_OBJECT_LEN as usize + 1];
1496
        {
1497
            let batch = store.begin_batch().unwrap();
1498
            store.put(ObjectType::Blob, b"small").unwrap();
1499
            let e = store.put(ObjectType::Change, &big).unwrap_err();
1500
            assert!(
1501
                matches!(e, StoreError::TooLarge { kind: "change", .. }),
1502
                "{e}"
1503
            );
1504
            assert!(e.to_string().contains("commit it in parts"), "{e}");
1505
            batch.finish().unwrap();
1506
        }
1507
        // The store still opens and reads what was written.
1508
        let again = Store::open(dir.path()).unwrap();
1509
        again.get(&hash_object(ObjectType::Blob, b"small")).unwrap();
1510
    }
1511
}