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