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