OXYGENHUB
# 2b872d074ddb oxygen / crates / oxy-core / src / store.rs
33.1 KiB · Rust · last changed Hardening: CBOR and pack corruption tests, index bounds validation, concurrent HTTP push test, 1 GiB benchmark (spec 0.6q) 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 = |m: &str| StoreError::Corrupt(ObjectRef::ZERO, format!("{}: {m}", idx_path.display()));
51
        if data.len() < 12 || &data[..8] != IDX_MAGIC {
52
            return Err(bad("bad index header"));
53
        }
54
        let count = u32::from_le_bytes(data[8..12].try_into().unwrap()) as usize;
55
        if data.len() != 12 + count * IDX_ENTRY {
56
            return Err(bad("index length mismatch"));
57
        }
58
        let pack_len = fs::metadata(pack_path)?.len();
59
        let mut refs = Vec::with_capacity(count);
60
        let mut entries = Vec::with_capacity(count);
61
        for i in 0..count {
62
            let e = &data[12 + i * IDX_ENTRY..12 + (i + 1) * IDX_ENTRY];
63
            let r = ObjectRef::from_slice(&e[..32]).unwrap();
64
            if let Some(prev) = refs.last() {
65
                if r <= *prev {
66
                    return Err(bad("index not sorted"));
67
                }
68
            }
69
            refs.push(r);
70
            let entry = PackEntry {
71
                offset: u64::from_le_bytes(e[32..40].try_into().unwrap()),
72
                stored_len: u32::from_le_bytes(e[40..44].try_into().unwrap()),
73
                raw_len: u32::from_le_bytes(e[44..48].try_into().unwrap()),
74
                ty: ObjectType::ALL.get(e[48] as usize).copied().ok_or_else(|| bad("unknown object type"))?,
75
                flags: e[49],
76
            };
77
            if entry.stored_len > MAX_OBJECT_LEN || entry.raw_len > MAX_OBJECT_LEN || entry.offset.saturating_add(entry.stored_len as u64) > pack_len {
78
                return Err(bad("index entry points outside the pack"));
79
            }
80
            entries.push(entry);
81
        }
82
        Ok(PackIndex { path: pack_path.to_path_buf(), refs, entries })
83
    }
84
85
    fn find(&self, r: &ObjectRef) -> Option<PackEntry> {
86
        self.refs.binary_search(r).ok().map(|i| self.entries[i])
87
    }
88
89
    fn read(&self, r: &ObjectRef, e: PackEntry) -> Result<(ObjectType, Vec<u8>)> {
90
        let mut f = fs::File::open(&self.path)?;
91
        f.seek(SeekFrom::Start(e.offset))?;
92
        let mut stored = vec![0u8; e.stored_len as usize];
93
        f.read_exact(&mut stored)?;
94
        let body = if e.flags & FLAG_ZSTD != 0 {
95
            zstd::bulk::decompress(&stored, e.raw_len as usize).map_err(|err| StoreError::Corrupt(*r, format!("zstd: {err}")))?
96
        } else {
97
            stored
98
        };
99
        if body.len() != e.raw_len as usize || hash_object(e.ty, &body) != *r {
100
            return Err(StoreError::Corrupt(*r, "packed object does not match its reference".into()));
101
        }
102
        Ok((e.ty, body))
103
    }
104
}
105
106
fn type_index(ty: ObjectType) -> u8 {
107
    ObjectType::ALL.iter().position(|t| *t == ty).unwrap() as u8
108
}
109
110
/// What `pack_loose` did.
111
#[derive(Debug, Clone, Default, PartialEq, Eq)]
112
pub struct PackReport {
113
    pub objects: usize,
114
    pub raw_bytes: u64,
115
    pub packed_bytes: u64,
116
}
117
118
/// Packs tolerated before `merge_packs` folds them into one (spec §13.4).
119
pub const MAX_PACKS: usize = 16;
120
121
/// What `prune` did.
122
#[derive(Debug, Clone, Default, PartialEq, Eq)]
123
pub struct PruneReport {
124
    pub objects: usize,
125
    pub bytes: u64,
126
    /// Unreachable objects left alone because they are younger than the grace period.
127
    pub deferred: usize,
128
}
129
130
#[derive(Debug, thiserror::Error)]
131
pub enum StoreError {
132
    #[error("io: {0}")]
133
    Io(#[from] io::Error),
134
    #[error("corrupt object {0}: {1}")]
135
    Corrupt(ObjectRef, String),
136
    #[error("object {0}")]
137
    Object(#[from] ObjectError),
138
    #[error("object {0} is a {1}, expected {2}")]
139
    WrongType(ObjectRef, &'static str, &'static str),
140
    #[error("object {0} not found")]
141
    NotFound(ObjectRef),
142
}
143
144
pub type Result<T> = std::result::Result<T, StoreError>;
145
146
#[derive(Debug)]
147
pub struct Store {
148
    dir: PathBuf,
149
    packs: RefCell<Vec<PackIndex>>,
150
}
151
152
impl Store {
153
    pub fn open(dir: impl Into<PathBuf>) -> io::Result<Store> {
154
        let dir = dir.into();
155
        fs::create_dir_all(dir.join("loose"))?;
156
        let store = Store { dir, packs: RefCell::new(Vec::new()) };
157
        store.reload_packs().map_err(|e| io::Error::other(e.to_string()))?;
158
        Ok(store)
159
    }
160
161
    fn reload_packs(&self) -> Result<()> {
162
        let mut packs = Vec::new();
163
        let mut names: Vec<PathBuf> = fs::read_dir(&self.dir)?
164
            .filter_map(|e| e.ok().map(|e| e.path()))
165
            .filter(|p| p.extension().is_some_and(|x| x == "idx") && p.file_name().is_some_and(|n| n.to_string_lossy().starts_with("pack-")))
166
            .collect();
167
        names.sort();
168
        for idx in names {
169
            let pack = idx.with_extension("pack");
170
            if pack.is_file() {
171
                packs.push(PackIndex::load(&idx, &pack)?);
172
            }
173
        }
174
        *self.packs.borrow_mut() = packs;
175
        Ok(())
176
    }
177
178
    pub fn pack_count(&self) -> usize {
179
        self.packs.borrow().len()
180
    }
181
182
    fn find_packed(&self, r: &ObjectRef) -> Option<(usize, PackEntry)> {
183
        self.packs.borrow().iter().enumerate().find_map(|(i, p)| p.find(r).map(|e| (i, e)))
184
    }
185
186
    /// Loose objects on disk, with their paths.
187
    fn loose_objects(&self) -> io::Result<Vec<(ObjectRef, PathBuf)>> {
188
        let mut out = Vec::new();
189
        let loose = self.dir.join("loose");
190
        for d in fs::read_dir(&loose)? {
191
            let d = d?;
192
            if !d.file_type()?.is_dir() {
193
                continue;
194
            }
195
            let prefix = d.file_name().to_string_lossy().to_string();
196
            for f in fs::read_dir(d.path())? {
197
                let f = f?;
198
                let name = f.file_name().to_string_lossy().to_string();
199
                if let Some(r) = ObjectRef::from_hex(&format!("{prefix}{name}")) {
200
                    out.push((r, f.path()));
201
                }
202
            }
203
        }
204
        out.sort();
205
        Ok(out)
206
    }
207
208
    pub fn loose_count(&self) -> io::Result<usize> {
209
        Ok(self.loose_objects()?.len())
210
    }
211
212
    /// Move every loose object into a new pack (spec §13.4). Loose files are
213
    /// deleted only after the pack and its index are durable.
214
    pub fn pack_loose(&self) -> Result<PackReport> {
215
        let loose = self.loose_objects()?;
216
        self.write_pack(&loose.iter().map(|(r, _)| *r).collect::<Vec<_>>(), &loose.iter().map(|(_, p)| p.clone()).collect::<Vec<_>>(), &[])
217
    }
218
219
    /// Fold every pack into one when there are more than `MAX_PACKS`
220
    /// (spec §13.4). Returns how many packs were merged.
221
    pub fn merge_packs(&self) -> Result<usize> {
222
        let (refs, paths): (Vec<ObjectRef>, Vec<PathBuf>) = {
223
            let packs = self.packs.borrow();
224
            if packs.len() <= MAX_PACKS {
225
                return Ok(0);
226
            }
227
            (packs.iter().flat_map(|p| p.refs.iter().copied()).collect(), packs.iter().map(|p| p.path.clone()).collect())
228
        };
229
        let n = paths.len();
230
        self.write_pack(&refs, &[], &paths)?;
231
        Ok(n)
232
    }
233
234
    /// Delete every object not in `keep` (spec §21). Loose objects and packs
235
    /// younger than `grace` are left alone, since a concurrent writer may not
236
    /// have recorded its references yet. Packs holding unreachable objects
237
    /// are rewritten without them. Signature sidecars and seal-index entries
238
    /// of deleted objects go too.
239
    pub fn prune(&self, keep: &std::collections::HashSet<ObjectRef>, grace: std::time::Duration) -> Result<PruneReport> {
240
        let now = std::time::SystemTime::now();
241
        let old_enough = |m: &fs::Metadata| m.modified().map(|t| now.duration_since(t).unwrap_or_default() >= grace).unwrap_or(true);
242
        let mut report = PruneReport::default();
243
        // Loose objects.
244
        for (r, path) in self.loose_objects()? {
245
            if keep.contains(&r) {
246
                continue;
247
            }
248
            let meta = fs::metadata(&path)?;
249
            if !old_enough(&meta) {
250
                report.deferred += 1;
251
                continue;
252
            }
253
            report.bytes += meta.len();
254
            report.objects += 1;
255
            fs::remove_file(&path)?;
256
            let _ = fs::remove_dir(path.parent().unwrap());
257
            self.remove_sidecars(&r)?;
258
        }
259
        // Packs: rewrite those that hold unreachable objects.
260
        let mut keep_refs: Vec<ObjectRef> = Vec::new();
261
        let mut old_paths: Vec<PathBuf> = Vec::new();
262
        let mut rewrite = false;
263
        for p in self.packs.borrow().iter() {
264
            let meta = fs::metadata(&p.path)?;
265
            let unreachable = p.refs.iter().filter(|r| !keep.contains(r)).count();
266
            if unreachable == 0 {
267
                continue;
268
            }
269
            if !old_enough(&meta) {
270
                report.deferred += unreachable;
271
                continue;
272
            }
273
            rewrite = true;
274
            old_paths.push(p.path.clone());
275
            for (r, e) in p.refs.iter().zip(p.entries.iter()) {
276
                if keep.contains(r) {
277
                    keep_refs.push(*r);
278
                } else {
279
                    report.objects += 1;
280
                    report.bytes += e.stored_len as u64;
281
                }
282
            }
283
        }
284
        if rewrite {
285
            let dropped: Vec<ObjectRef> = {
286
                let packs = self.packs.borrow();
287
                packs.iter().filter(|p| old_paths.contains(&p.path)).flat_map(|p| p.refs.iter().copied()).filter(|r| !keep.contains(r)).collect()
288
            };
289
            self.write_pack(&keep_refs, &[], &old_paths)?;
290
            for r in dropped {
291
                self.remove_sidecars(&r)?;
292
            }
293
        }
294
        Ok(report)
295
    }
296
297
    fn remove_sidecars(&self, r: &ObjectRef) -> Result<()> {
298
        for path in [self.sig_path(r), self.seal_index_path(r)] {
299
            match fs::remove_file(&path) {
300
                Ok(()) => {
301
                    let _ = fs::remove_dir(path.parent().unwrap());
302
                }
303
                Err(e) if e.kind() == io::ErrorKind::NotFound => {}
304
                Err(e) => return Err(e.into()),
305
            }
306
        }
307
        Ok(())
308
    }
309
310
    /// Write `refs` into a new pack, then delete `loose_paths` and `old_packs`.
311
    fn write_pack(&self, refs: &[ObjectRef], loose_paths: &[PathBuf], old_packs: &[PathBuf]) -> Result<PackReport> {
312
        if refs.is_empty() {
313
            for p in old_packs {
314
                fs::remove_file(p)?;
315
                fs::remove_file(p.with_extension("idx"))?;
316
            }
317
            self.reload_packs()?;
318
            return Ok(PackReport::default());
319
        }
320
        let mut refs: Vec<ObjectRef> = refs.to_vec();
321
        refs.sort();
322
        refs.dedup();
323
        let mut nonce = [0u8; 8];
324
        rand::rngs::OsRng.fill_bytes(&mut nonce);
325
        let id = hex::encode(nonce);
326
        let pack_path = self.dir.join(format!("pack-{id}.pack"));
327
        let idx_path = self.dir.join(format!("pack-{id}.idx"));
328
        let tmp_pack = self.dir.join(format!("tmp-{id}.pack"));
329
        let tmp_idx = self.dir.join(format!("tmp-{id}.idx"));
330
        let mut report = PackReport::default();
331
        let mut index: Vec<(ObjectRef, PackEntry)> = Vec::with_capacity(refs.len());
332
        {
333
            let mut f = io::BufWriter::new(fs::File::create(&tmp_pack)?);
334
            f.write_all(PACK_MAGIC)?;
335
            let mut offset = PACK_MAGIC.len() as u64;
336
            for r in &refs {
337
                let (ty, body) = self.get(r)?;
338
                let (flags, stored) = if self.is_nocompress(r) {
339
                    (0, body.clone())
340
                } else {
341
                    let compressed = zstd::bulk::compress(&body, ZSTD_LEVEL).map_err(|e| StoreError::Corrupt(*r, format!("zstd: {e}")))?;
342
                    if compressed.len() < body.len() { (FLAG_ZSTD, compressed) } else { (0, body.clone()) }
343
                };
344
                let header_len = 1 + 1 + 4 + 4 + 32;
345
                f.write_all(&[type_index(ty), flags])?;
346
                f.write_all(&(stored.len() as u32).to_le_bytes())?;
347
                f.write_all(&(body.len() as u32).to_le_bytes())?;
348
                f.write_all(&r.0)?;
349
                f.write_all(&stored)?;
350
                index.push((*r, PackEntry { offset: offset + header_len, stored_len: stored.len() as u32, raw_len: body.len() as u32, ty, flags }));
351
                offset += header_len + stored.len() as u64;
352
                report.objects += 1;
353
                report.raw_bytes += body.len() as u64;
354
                report.packed_bytes += stored.len() as u64;
355
            }
356
            f.flush()?;
357
            f.get_ref().sync_all()?;
358
        }
359
        {
360
            let mut f = io::BufWriter::new(fs::File::create(&tmp_idx)?);
361
            f.write_all(IDX_MAGIC)?;
362
            f.write_all(&(index.len() as u32).to_le_bytes())?;
363
            for (r, e) in &index {
364
                f.write_all(&r.0)?;
365
                f.write_all(&e.offset.to_le_bytes())?;
366
                f.write_all(&e.stored_len.to_le_bytes())?;
367
                f.write_all(&e.raw_len.to_le_bytes())?;
368
                f.write_all(&[type_index(e.ty), e.flags])?;
369
            }
370
            f.flush()?;
371
            f.get_ref().sync_all()?;
372
        }
373
        fs::rename(&tmp_pack, &pack_path)?;
374
        fs::rename(&tmp_idx, &idx_path)?;
375
        for p in old_packs {
376
            fs::remove_file(p)?;
377
            fs::remove_file(p.with_extension("idx"))?;
378
        }
379
        self.reload_packs()?;
380
        for path in loose_paths {
381
            fs::remove_file(path)?;
382
            if let Some(dir) = path.parent() {
383
                let _ = fs::remove_dir(dir);
384
            }
385
        }
386
        Ok(report)
387
    }
388
389
    pub fn dir(&self) -> &Path {
390
        &self.dir
391
    }
392
393
    fn path_for(&self, r: &ObjectRef) -> PathBuf {
394
        let h = r.to_hex();
395
        self.dir.join("loose").join(&h[..2]).join(&h[2..])
396
    }
397
398
    pub fn has(&self, r: &ObjectRef) -> bool {
399
        self.path_for(r).is_file() || self.find_packed(r).is_some()
400
    }
401
402
    /// Store `body` as an object of type `ty`; idempotent.
403
    pub fn put(&self, ty: ObjectType, body: &[u8]) -> Result<ObjectRef> {
404
        let r = hash_object(ty, body);
405
        let path = self.path_for(&r);
406
        if path.is_file() || self.find_packed(&r).is_some() {
407
            return Ok(r);
408
        }
409
        fs::create_dir_all(path.parent().unwrap())?;
410
        let mut nonce = [0u8; 8];
411
        rand::rngs::OsRng.fill_bytes(&mut nonce);
412
        let tmp = self.dir.join("loose").join(format!("tmp-{}", hex::encode(nonce)));
413
        {
414
            let mut f = fs::File::create(&tmp)?;
415
            f.write_all(&header(ty, body.len()))?;
416
            f.write_all(body)?;
417
            f.sync_all()?;
418
        }
419
        fs::rename(&tmp, &path)?;
420
        Ok(r)
421
    }
422
423
    /// Read and verify an object, loose or packed.
424
    pub fn get(&self, r: &ObjectRef) -> Result<(ObjectType, Vec<u8>)> {
425
        let path = self.path_for(r);
426
        let data = match fs::read(&path) {
427
            Ok(d) => d,
428
            Err(e) if e.kind() == io::ErrorKind::NotFound => {
429
                return match self.find_packed(r) {
430
                    Some((i, e)) => self.packs.borrow()[i].read(r, e),
431
                    None => Err(StoreError::NotFound(*r)),
432
                };
433
            }
434
            Err(e) => return Err(e.into()),
435
        };
436
        let nul = data.iter().position(|&b| b == 0).ok_or_else(|| StoreError::Corrupt(*r, "no header".into()))?;
437
        let head = std::str::from_utf8(&data[..nul]).map_err(|_| StoreError::Corrupt(*r, "header not UTF-8".into()))?;
438
        let mut parts = head.split(' ');
439
        let (Some("oxy1"), Some(ty), Some(len), None) = (parts.next(), parts.next(), parts.next(), parts.next()) else {
440
            return Err(StoreError::Corrupt(*r, format!("bad header {head:?}")));
441
        };
442
        let ty = ObjectType::from_name(ty).ok_or_else(|| StoreError::Corrupt(*r, format!("unknown type {ty}")))?;
443
        let len: usize = len.parse().map_err(|_| StoreError::Corrupt(*r, "bad length".into()))?;
444
        let body = &data[nul + 1..];
445
        if body.len() != len {
446
            return Err(StoreError::Corrupt(*r, format!("length {} != header {len}", body.len())));
447
        }
448
        if hash_object(ty, body) != *r {
449
            return Err(StoreError::Corrupt(*r, "hash mismatch".into()));
450
        }
451
        Ok((ty, body.to_vec()))
452
    }
453
454
    pub fn get_typed(&self, r: &ObjectRef, want: ObjectType) -> Result<Vec<u8>> {
455
        let (ty, body) = self.get(r)?;
456
        if ty != want {
457
            return Err(StoreError::WrongType(*r, ty.name(), want.name()));
458
        }
459
        Ok(body)
460
    }
461
462
    pub fn put_change(&self, c: &Change) -> Result<ObjectRef> {
463
        self.put(ObjectType::Change, &c.encode())
464
    }
465
466
    pub fn get_change(&self, r: &ObjectRef) -> Result<Change> {
467
        Ok(Change::decode(&self.get_typed(r, ObjectType::Change)?)?)
468
    }
469
470
    pub fn put_chunk(&self, bytes: &[u8]) -> Result<ObjectRef> {
471
        self.put(ObjectType::Chunk, bytes)
472
    }
473
474
    /// Store a chunk that packs will keep uncompressed (`nocompress`, spec §13.3).
475
    pub fn put_chunk_nocompress(&self, bytes: &[u8]) -> Result<ObjectRef> {
476
        let r = self.put(ObjectType::Chunk, bytes)?;
477
        let dir = self.dir.join("nocompress");
478
        fs::create_dir_all(&dir)?;
479
        let _ = fs::File::create(dir.join(r.to_hex()));
480
        Ok(r)
481
    }
482
483
    fn is_nocompress(&self, r: &ObjectRef) -> bool {
484
        self.dir.join("nocompress").join(r.to_hex()).exists()
485
    }
486
487
    pub fn put_state(&self, st: &StateObject) -> Result<ObjectRef> {
488
        self.put(ObjectType::State, &st.encode())
489
    }
490
491
    pub fn get_state(&self, r: &ObjectRef) -> Result<StateObject> {
492
        Ok(StateObject::decode(&self.get_typed(r, ObjectType::State)?)?)
493
    }
494
495
    pub fn put_attest(&self, a: &Attest) -> Result<ObjectRef> {
496
        self.put(ObjectType::Attest, &a.encode())
497
    }
498
499
    pub fn get_attest(&self, r: &ObjectRef) -> Result<Attest> {
500
        Ok(Attest::decode(&self.get_typed(r, ObjectType::Attest)?)?)
501
    }
502
503
    fn seal_index_path(&self, change: &ObjectRef) -> PathBuf {
504
        let h = change.to_hex();
505
        self.dir.join("seals").join(&h[..2]).join(&h[2..])
506
    }
507
508
    /// Record that seal attestation `attest` covers `change`.
509
    pub fn index_seal(&self, change: &ObjectRef, attest: &ObjectRef) -> Result<()> {
510
        let mut seals = self.seals_for(change)?;
511
        if seals.contains(attest) {
512
            return Ok(());
513
        }
514
        seals.push(*attest);
515
        seals.sort();
516
        let path = self.seal_index_path(change);
517
        fs::create_dir_all(path.parent().unwrap())?;
518
        let text: String = seals.iter().map(|s| format!("{}\n", s.to_hex())).collect();
519
        let tmp = path.with_extension("tmp");
520
        fs::write(&tmp, text)?;
521
        fs::rename(&tmp, &path)?;
522
        Ok(())
523
    }
524
525
    /// Seal attestations known to cover `change`.
526
    pub fn seals_for(&self, change: &ObjectRef) -> Result<Vec<ObjectRef>> {
527
        match fs::read_to_string(self.seal_index_path(change)) {
528
            Ok(t) => Ok(t.lines().filter_map(ObjectRef::from_hex).collect()),
529
            Err(e) if e.kind() == io::ErrorKind::NotFound => Ok(vec![]),
530
            Err(e) => Err(e.into()),
531
        }
532
    }
533
534
    pub fn put_identity(&self, id: &Identity) -> Result<ObjectRef> {
535
        self.put(ObjectType::Identity, &id.encode())
536
    }
537
538
    pub fn get_identity(&self, r: &ObjectRef) -> Result<Identity> {
539
        Ok(Identity::decode(&self.get_typed(r, ObjectType::Identity)?)?)
540
    }
541
542
    fn sig_path(&self, r: &ObjectRef) -> PathBuf {
543
        let h = r.to_hex();
544
        self.dir.join("sigs").join(&h[..2]).join(&h[2..])
545
    }
546
547
    /// Signatures stored alongside object `r` (spec §5.4). Empty if none.
548
    pub fn signatures(&self, r: &ObjectRef) -> Result<Vec<Signature>> {
549
        match fs::read(self.sig_path(r)) {
550
            Ok(b) => Ok(Signature::decode_list(&b)?),
551
            Err(e) if e.kind() == io::ErrorKind::NotFound => Ok(vec![]),
552
            Err(e) => Err(e.into()),
553
        }
554
    }
555
556
    /// Attach a signature to object `r`, replacing any earlier one by the same key.
557
    pub fn put_signature(&self, r: &ObjectRef, sig: &Signature) -> Result<()> {
558
        let mut sigs = self.signatures(r)?;
559
        sigs.retain(|s| s.key != sig.key);
560
        sigs.push(sig.clone());
561
        sigs.sort_by_key(|s| s.key);
562
        let path = self.sig_path(r);
563
        fs::create_dir_all(path.parent().unwrap())?;
564
        let mut nonce = [0u8; 8];
565
        rand::rngs::OsRng.fill_bytes(&mut nonce);
566
        let tmp = self.dir.join(format!("sigs-tmp-{}", hex::encode(nonce)));
567
        {
568
            let mut f = fs::File::create(&tmp)?;
569
            f.write_all(&Signature::encode_list(&sigs))?;
570
            f.sync_all()?;
571
        }
572
        fs::rename(&tmp, &path)?;
573
        Ok(())
574
    }
575
576
    /// Every object reference in the store, loose and packed (slow; for gc and tests).
577
    pub fn list(&self) -> io::Result<Vec<ObjectRef>> {
578
        let mut out: Vec<ObjectRef> = self.loose_objects()?.into_iter().map(|(r, _)| r).collect();
579
        for p in self.packs.borrow().iter() {
580
            out.extend(p.refs.iter().copied());
581
        }
582
        out.sort();
583
        out.dedup();
584
        Ok(out)
585
    }
586
}
587
588
#[cfg(test)]
589
mod tests {
590
    use super::*;
591
592
    #[test]
593
    fn put_get_verify() {
594
        let dir = tempfile::tempdir().unwrap();
595
        let s = Store::open(dir.path().join("objects")).unwrap();
596
        let r = s.put(ObjectType::Chunk, b"hello").unwrap();
597
        assert!(s.has(&r));
598
        assert_eq!(s.get(&r).unwrap(), (ObjectType::Chunk, b"hello".to_vec()));
599
        assert_eq!(s.put(ObjectType::Chunk, b"hello").unwrap(), r);
600
        assert_eq!(s.list().unwrap(), vec![r]);
601
        assert!(matches!(s.get_typed(&r, ObjectType::Blob), Err(StoreError::WrongType(..))));
602
        assert!(matches!(s.get(&ObjectRef([9; 32])), Err(StoreError::NotFound(_))));
603
604
        // Tamper: flip a byte, read must fail.
605
        let path = s.path_for(&r);
606
        let mut data = fs::read(&path).unwrap();
607
        let last = data.len() - 1;
608
        data[last] ^= 1;
609
        fs::write(&path, data).unwrap();
610
        assert!(matches!(s.get(&r), Err(StoreError::Corrupt(..))));
611
    }
612
613
    #[test]
614
    fn packing_keeps_every_object_readable() {
615
        let dir = tempfile::tempdir().unwrap();
616
        let s = Store::open(dir.path()).unwrap();
617
        let mut refs = Vec::new();
618
        for i in 0..300u32 {
619
            let body = format!("object {i} ").repeat(1 + (i % 7) as usize);
620
            refs.push((s.put(ObjectType::Chunk, body.as_bytes()).unwrap(), body));
621
        }
622
        let c = crate::objects::tests::sample_change();
623
        let cr = s.put_change(&c).unwrap();
624
        assert_eq!(s.loose_count().unwrap(), 301);
625
        let report = s.pack_loose().unwrap();
626
        assert_eq!(report.objects, 301);
627
        assert!(report.packed_bytes < report.raw_bytes, "{report:?}");
628
        assert_eq!(s.loose_count().unwrap(), 0);
629
        assert_eq!(s.pack_count(), 1);
630
        for (r, body) in &refs {
631
            assert!(s.has(r));
632
            assert_eq!(s.get(r).unwrap(), (ObjectType::Chunk, body.as_bytes().to_vec()));
633
        }
634
        assert_eq!(s.get_change(&cr).unwrap(), c);
635
        assert_eq!(s.list().unwrap().len(), 301);
636
        // Re-putting a packed object is a no-op; new objects go loose and a
637
        // second pack coexists with the first.
638
        assert_eq!(s.put(ObjectType::Chunk, refs[0].1.as_bytes()).unwrap(), refs[0].0);
639
        assert_eq!(s.loose_count().unwrap(), 0);
640
        let extra = s.put(ObjectType::Chunk, b"after packing").unwrap();
641
        s.pack_loose().unwrap();
642
        assert_eq!(s.pack_count(), 2);
643
        assert_eq!(s.get(&extra).unwrap().1, b"after packing");
644
        // A reopened store sees the packs.
645
        let s2 = Store::open(dir.path()).unwrap();
646
        assert_eq!(s2.pack_count(), 2);
647
        assert_eq!(s2.get(&refs[5].0).unwrap().1, refs[5].1.as_bytes());
648
        assert!(matches!(s2.get(&ObjectRef([9; 32])), Err(StoreError::NotFound(_))));
649
        // Corrupting an object's body inside the pack is detected on read.
650
        let target = refs[5].0;
651
        let (pack_path, offset) = {
652
            let packs = s2.packs.borrow();
653
            let (i, e) = s2.find_packed(&target).unwrap();
654
            (packs[i].path.clone(), e.offset as usize)
655
        };
656
        let mut data = fs::read(&pack_path).unwrap();
657
        data[offset] ^= 0xff;
658
        fs::write(&pack_path, data).unwrap();
659
        let s3 = Store::open(dir.path()).unwrap();
660
        assert!(matches!(s3.get(&target), Err(StoreError::Corrupt(..))));
661
        assert_eq!(s3.get(&refs[6].0).unwrap().1, refs[6].1.as_bytes(), "other objects are unaffected");
662
    }
663
664
    #[test]
665
    fn nocompress_chunks_pack_raw() {
666
        let dir = tempfile::tempdir().unwrap();
667
        let s = Store::open(dir.path()).unwrap();
668
        let body = vec![b'a'; 10_000];
669
        let raw = s.put_chunk_nocompress(&body).unwrap();
670
        let compressible = s.put(ObjectType::Chunk, &vec![b'b'; 10_000]).unwrap();
671
        s.pack_loose().unwrap();
672
        let packs = s.packs.borrow();
673
        let (i, e) = s.find_packed(&raw).unwrap();
674
        assert_eq!((e.flags, e.stored_len as usize), (0, 10_000));
675
        let (_, c) = s.find_packed(&compressible).unwrap();
676
        assert_eq!(c.flags, FLAG_ZSTD);
677
        assert!(c.stored_len < 10_000);
678
        assert_eq!(packs[i].read(&raw, e).unwrap().1, body);
679
    }
680
681
    #[test]
682
    fn many_packs_merge_into_one() {
683
        let dir = tempfile::tempdir().unwrap();
684
        let s = Store::open(dir.path()).unwrap();
685
        let mut refs = Vec::new();
686
        for i in 0..(MAX_PACKS + 2) {
687
            refs.push(s.put(ObjectType::Chunk, format!("pack {i}").as_bytes()).unwrap());
688
            s.pack_loose().unwrap();
689
        }
690
        assert_eq!(s.pack_count(), MAX_PACKS + 2);
691
        assert_eq!(s.merge_packs().unwrap(), MAX_PACKS + 2);
692
        assert_eq!(s.pack_count(), 1);
693
        assert_eq!(s.merge_packs().unwrap(), 0);
694
        for (i, r) in refs.iter().enumerate() {
695
            assert_eq!(s.get(r).unwrap().1, format!("pack {i}").as_bytes());
696
        }
697
        assert_eq!(Store::open(dir.path()).unwrap().pack_count(), 1);
698
    }
699
700
    #[test]
701
    fn prune_removes_unreachable_loose_and_packed_objects() {
702
        use std::collections::HashSet;
703
        let dir = tempfile::tempdir().unwrap();
704
        let s = Store::open(dir.path()).unwrap();
705
        let keep_a = s.put(ObjectType::Chunk, b"keep a").unwrap();
706
        let drop_a = s.put(ObjectType::Chunk, b"drop a").unwrap();
707
        s.pack_loose().unwrap();
708
        let keep_b = s.put(ObjectType::Chunk, b"keep b").unwrap();
709
        let drop_b = s.put(ObjectType::Chunk, b"drop b").unwrap();
710
        let sk = crate::identity::SecretKey { ed_seed: [1; 32], ml_seed: [2; 32] };
711
        let id = sk.identity("a", "b", 0);
712
        let key = s.put_identity(&id).unwrap();
713
        s.put_signature(&drop_b, &sk.sign(crate::identity::Domain::Change, &key, &drop_b)).unwrap();
714
        s.index_seal(&drop_a, &ObjectRef([3; 32])).unwrap();
715
        let keep: HashSet<ObjectRef> = [keep_a, keep_b, key].into_iter().collect();
716
        // Everything is too young under a grace period.
717
        let r = s.prune(&keep, std::time::Duration::from_secs(3600)).unwrap();
718
        assert_eq!((r.objects, r.deferred), (0, 2));
719
        let r = s.prune(&keep, std::time::Duration::ZERO).unwrap();
720
        assert_eq!(r.objects, 2, "{r:?}");
721
        assert!(s.has(&keep_a) && s.has(&keep_b) && s.has(&key));
722
        assert!(!s.has(&drop_a) && !s.has(&drop_b));
723
        assert!(s.signatures(&drop_b).unwrap().is_empty());
724
        assert!(s.seals_for(&drop_a).unwrap().is_empty());
725
        assert_eq!(s.pack_count(), 1);
726
        assert_eq!(s.get(&keep_a).unwrap().1, b"keep a");
727
        assert_eq!(s.list().unwrap().len(), 3);
728
    }
729
730
    #[test]
731
    fn signatures_are_stored_beside_objects() {
732
        use crate::identity::{Domain, SecretKey};
733
        let dir = tempfile::tempdir().unwrap();
734
        let s = Store::open(dir.path()).unwrap();
735
        let r = s.put(ObjectType::Chunk, b"x").unwrap();
736
        assert!(s.signatures(&r).unwrap().is_empty());
737
        let sk = SecretKey { ed_seed: [1; 32], ml_seed: [2; 32] };
738
        let id = sk.identity("a", "b", 0);
739
        let key = s.put_identity(&id).unwrap();
740
        assert_eq!(s.get_identity(&key).unwrap(), id);
741
        let sig = sk.sign(Domain::Change, &key, &r);
742
        s.put_signature(&r, &sig).unwrap();
743
        s.put_signature(&r, &sig).unwrap();
744
        assert_eq!(s.signatures(&r).unwrap(), vec![sig]);
745
        // The object itself is untouched by its signatures.
746
        assert_eq!(s.get(&r).unwrap().1, b"x");
747
    }
748
749
    #[test]
750
    fn change_round_trip_through_store() {
751
        let dir = tempfile::tempdir().unwrap();
752
        let s = Store::open(dir.path()).unwrap();
753
        let c = crate::objects::tests::sample_change();
754
        let r = s.put_change(&c).unwrap();
755
        assert_eq!(r, c.hash());
756
        assert_eq!(s.get_change(&r).unwrap(), c);
757
    }
758
759
    fn xorshift(state: &mut u64) -> u64 {
760
        *state ^= *state << 13;
761
        *state ^= *state >> 7;
762
        *state ^= *state << 17;
763
        *state
764
    }
765
766
    /// Random corruption of pack and index files never panics: reads fail
767
    /// with an error or the store refuses to open, and untouched objects
768
    /// stay readable when the index survives.
769
    #[test]
770
    fn corrupted_packs_and_indexes_fail_cleanly() {
771
        let dir = tempfile::tempdir().unwrap();
772
        let objects = dir.path().join("objects");
773
        let s = Store::open(&objects).unwrap();
774
        let mut refs = Vec::new();
775
        let mut seed = 0x9e3779b97f4a7c15u64;
776
        for i in 0..40u32 {
777
            let len = (xorshift(&mut seed) % 5000) as usize;
778
            let mut body = vec![0u8; len];
779
            for b in body.iter_mut() {
780
                *b = if i % 2 == 0 { (xorshift(&mut seed) & 0xff) as u8 } else { b'a' + (i % 26) as u8 };
781
            }
782
            refs.push(s.put(ObjectType::Chunk, &body).unwrap());
783
        }
784
        s.pack_loose().unwrap();
785
        let pack = fs::read_dir(&objects).unwrap().filter_map(|e| e.ok().map(|e| e.path())).find(|p| p.extension().is_some_and(|x| x == "pack")).unwrap();
786
        let idx = pack.with_extension("idx");
787
        let pristine_pack = fs::read(&pack).unwrap();
788
        let pristine_idx = fs::read(&idx).unwrap();
789
790
        for round in 0..300u32 {
791
            let (path, pristine) = if round % 2 == 0 { (&pack, &pristine_pack) } else { (&idx, &pristine_idx) };
792
            let mut data = pristine.clone();
793
            match xorshift(&mut seed) % 4 {
794
                0 => {
795
                    let at = (xorshift(&mut seed) as usize) % data.len();
796
                    data[at] ^= 1 << (xorshift(&mut seed) % 8);
797
                }
798
                1 => {
799
                    let at = (xorshift(&mut seed) as usize) % data.len();
800
                    data[at] = (xorshift(&mut seed) & 0xff) as u8;
801
                }
802
                2 => {
803
                    let keep = (xorshift(&mut seed) as usize) % data.len();
804
                    data.truncate(keep);
805
                }
806
                _ => {
807
                    let at = (xorshift(&mut seed) as usize) % data.len();
808
                    let n = ((xorshift(&mut seed) % 8) + 1) as usize;
809
                    for k in 0..n.min(data.len() - at) {
810
                        data[at + k] = 0xff;
811
                    }
812
                }
813
            }
814
            fs::write(path, &data).unwrap();
815
            // Opening may refuse a broken index; a store that opens must
816
            // either return each object intact or report it corrupt/missing.
817
            if let Ok(s2) = Store::open(&objects) {
818
                for r in &refs {
819
                    match s2.get(r) {
820
                        Ok((ty, body)) => assert_eq!(hash_object(ty, &body), *r, "a returned object always matches its reference"),
821
                        Err(StoreError::Corrupt(..)) | Err(StoreError::NotFound(_)) | Err(StoreError::Io(_)) => {}
822
                        Err(e) => panic!("unexpected error {e}"),
823
                    }
824
                }
825
            }
826
            fs::write(path, pristine).unwrap();
827
        }
828
        // Restored files read back in full.
829
        let s3 = Store::open(&objects).unwrap();
830
        for r in &refs {
831
            let (ty, body) = s3.get(r).unwrap();
832
            assert_eq!(hash_object(ty, &body), *r);
833
        }
834
    }
835
}