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