//! Object store (spec §13.1, §13.3, §13.4): loose objects, one file per //! object under `objects/loose//`, and packs, `pack-.pack` //! with a sorted `pack-.idx`, holding zstd-compressed objects. Reads //! check loose objects first, then every pack; every read re-hashes. //! //! Pack file: `OXYPACK1` then entries of //! `type u8 | flags u8 (bit 0: zstd) | stored_len u32 | raw_len u32 | ref 32 | body`. //! Index file: `OXYIDX01` then `count u32` then sorted entries of //! `ref 32 | offset u64 | stored_len u32 | raw_len u32 | type u8 | flags u8`. use crate::attest::Attest; use crate::hash::{hash_object, header, ObjectRef, ObjectType}; use crate::identity::{Identity, Signature}; use crate::objects::{Blob, Change, ObjectError, StateObject}; use crate::receipt::Receipt; use rand::RngCore; use std::cell::RefCell; use std::fs; use std::io::{self, Read, Seek, SeekFrom, Write}; use std::path::{Path, PathBuf}; const PACK_MAGIC: &[u8; 8] = b"OXYPACK1"; const IDX_MAGIC: &[u8; 8] = b"OXYIDX01"; const IDX_ENTRY: usize = 32 + 8 + 4 + 4 + 1 + 1; const FLAG_ZSTD: u8 = 1; /// Compressed with the pack's trained dictionary (`pack-ID.dict`), which a /// full repack makes from a sample of the objects it packs: small files /// share most of their vocabulary, which each one alone cannot use. const FLAG_DICT: u8 = 2; /// The size of a pack's dictionary, and the least a repack must pack for /// one to pay. const DICT_BYTES: usize = 110 << 10; const DICT_MIN_OBJECTS: usize = 1000; /// Objects up to this size are compressed with the dictionary. const DICT_OBJECT_MAX: usize = 64 << 10; const ZSTD_LEVEL: i32 = 3; /// The level of a full repack (`oxy gc`): slower, once, for a smaller store. pub const REPACK_ZSTD_LEVEL: i32 = 9; /// Largest object a pack entry may claim; anything bigger is corruption, /// and refusing it keeps a bad index from driving huge allocations. const MAX_OBJECT_LEN: u32 = 1 << 30; /// An object read from the store: its type and body. pub type Object = (ObjectType, Vec); #[derive(Clone, Copy, Debug)] struct PackEntry { offset: u64, stored_len: u32, raw_len: u32, ty: ObjectType, flags: u8, } /// One pack's index, fully in memory and sorted by ObjectRef. struct PackIndex { path: PathBuf, refs: Vec, entries: Vec, /// The pack's dictionary, prepared once, when it has one. dict: Option>>, } impl std::fmt::Debug for PackIndex { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { f.debug_struct("PackIndex") .field("path", &self.path) .field("objects", &self.refs.len()) .field("dict", &self.dict.is_some()) .finish() } } impl PackIndex { fn load(idx_path: &Path, pack_path: &Path) -> Result { let data = fs::read(idx_path)?; let bad = |m: &str| StoreError::Corrupt(ObjectRef::ZERO, format!("{}: {m}", idx_path.display())); if data.len() < 12 || &data[..8] != IDX_MAGIC { return Err(bad("bad index header")); } let count = u32::from_le_bytes(data[8..12].try_into().unwrap()) as usize; if data.len() != 12 + count * IDX_ENTRY { return Err(bad("index length mismatch")); } let pack_len = fs::metadata(pack_path)?.len(); let mut refs = Vec::with_capacity(count); let mut entries = Vec::with_capacity(count); for i in 0..count { let e = &data[12 + i * IDX_ENTRY..12 + (i + 1) * IDX_ENTRY]; let r = ObjectRef::from_slice(&e[..32]).unwrap(); if let Some(prev) = refs.last() { if r <= *prev { return Err(bad("index not sorted")); } } refs.push(r); let entry = PackEntry { offset: u64::from_le_bytes(e[32..40].try_into().unwrap()), stored_len: u32::from_le_bytes(e[40..44].try_into().unwrap()), raw_len: u32::from_le_bytes(e[44..48].try_into().unwrap()), ty: ObjectType::ALL .get(e[48] as usize) .copied() .ok_or_else(|| bad("unknown object type"))?, flags: e[49], }; if entry.stored_len > MAX_OBJECT_LEN || entry.raw_len > MAX_OBJECT_LEN { return Err(StoreError::Corrupt( r, format!( "{}: the index claims an object of {} bytes ({} stored), more than the {MAX_OBJECT_LEN} an object may be", idx_path.display(), entry.raw_len, entry.stored_len ), )); } if entry.offset.saturating_add(entry.stored_len as u64) > pack_len { return Err(bad("index entry points outside the pack")); } entries.push(entry); } let dict = match fs::read(pack_path.with_extension("dict")) { Ok(d) => Some(std::sync::Arc::new(zstd::dict::DecoderDictionary::copy(&d))), Err(e) if e.kind() == io::ErrorKind::NotFound => None, Err(e) => return Err(e.into()), }; Ok(PackIndex { path: pack_path.to_path_buf(), refs, entries, dict, }) } fn find(&self, r: &ObjectRef) -> Option { self.refs.binary_search(r).ok().map(|i| self.entries[i]) } fn read(&self, r: &ObjectRef, e: PackEntry) -> Result<(ObjectType, Vec)> { let mut f = fs::File::open(&self.path)?; f.seek(SeekFrom::Start(e.offset))?; let mut stored = vec![0u8; e.stored_len as usize]; f.read_exact(&mut stored)?; let body = if e.flags & FLAG_DICT != 0 { let dict = self.dict.as_ref().ok_or_else(|| { StoreError::Corrupt(*r, "packed with a dictionary the pack lacks".into()) })?; zstd::bulk::Decompressor::with_prepared_dictionary(dict) .and_then(|mut d| d.decompress(&stored, e.raw_len as usize)) .map_err(|err| StoreError::Corrupt(*r, format!("zstd: {err}")))? } else if e.flags & FLAG_ZSTD != 0 { zstd::bulk::decompress(&stored, e.raw_len as usize) .map_err(|err| StoreError::Corrupt(*r, format!("zstd: {err}")))? } else { stored }; if body.len() != e.raw_len as usize || hash_object(e.ty, &body) != *r { return Err(StoreError::Corrupt( *r, "packed object does not match its reference".into(), )); } Ok((e.ty, body)) } } fn type_index(ty: ObjectType) -> u8 { ObjectType::ALL.iter().position(|t| *t == ty).unwrap() as u8 } /// What `pack_loose` did. #[derive(Debug, Clone, Default, PartialEq, Eq)] pub struct PackReport { pub objects: usize, pub raw_bytes: u64, pub packed_bytes: u64, } /// Packs tolerated before `merge_packs` folds them into one (spec §13.4). pub const MAX_PACKS: usize = 16; /// What `prune` did. #[derive(Debug, Clone, Default, PartialEq, Eq)] pub struct PruneReport { pub objects: usize, pub bytes: u64, /// Unreachable objects left alone because they are younger than the grace period. pub deferred: usize, } #[derive(Debug, thiserror::Error)] pub enum StoreError { #[error("io: {0}")] Io(#[from] io::Error), #[error("corrupt object {0}: {1}")] Corrupt(ObjectRef, String), #[error("object {0}")] Object(#[from] ObjectError), #[error("object {0} is a {1}, expected {2}")] WrongType(ObjectRef, &'static str, &'static str), #[error("object {0} not found")] NotFound(ObjectRef), #[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)")] TooLarge { kind: &'static str, size: usize, limit: u32, }, } pub type Result = std::result::Result; #[derive(Debug)] pub struct Store { dir: PathBuf, packs: RefCell>, /// An open write batch (spec §13.4): objects go here instead of to loose /// files, and are written out with one `fsync` when the batch finishes. pending: RefCell>, batch_depth: std::cell::Cell, /// Change metadata already read, from the `metas` cache file. metas: RefCell>>, } /// Objects written since `begin_batch`. Small batches stay in memory and /// become loose objects; once `SPILL_OBJECTS` or `SPILL_BYTES` is exceeded /// the batch streams into a pack file instead. #[derive(Debug)] struct Pending { buffered: Vec<(ObjectType, Vec)>, buffered_bytes: usize, lookup: std::collections::HashMap, pack: Option, report: PackReport, } #[derive(Debug, Clone, Copy)] enum PendingEntry { Buffered(usize), Packed(PackEntry), } #[derive(Debug)] struct PendingPack { id: String, tmp_path: PathBuf, writer: io::BufWriter, offset: u64, index: Vec<(ObjectRef, PackEntry)>, } const SPILL_OBJECTS: usize = 256; const SPILL_BYTES: usize = 8 << 20; /// Finishes the enclosing batch when dropped; call `finish` to see errors. pub struct Batch<'a> { store: &'a Store, done: bool, } impl Batch<'_> { pub fn finish(mut self) -> Result<()> { self.done = true; self.store.finish_batch() } } impl Drop for Batch<'_> { fn drop(&mut self) { if !self.done { let _ = self.store.finish_batch(); } } } /// Append one stored object to a pending pack and index it. fn write_pack_entry( pack: &mut PendingPack, r: &ObjectRef, ty: ObjectType, raw_len: usize, flags: u8, stored: &[u8], report: &mut PackReport, ) -> Result { let header_len = 1 + 1 + 4 + 4 + 32; pack.writer.write_all(&[type_index(ty), flags])?; pack.writer .write_all(&(stored.len() as u32).to_le_bytes())?; pack.writer.write_all(&(raw_len as u32).to_le_bytes())?; pack.writer.write_all(&r.0)?; pack.writer.write_all(stored)?; let entry = PackEntry { offset: pack.offset + header_len, stored_len: stored.len() as u32, raw_len: raw_len as u32, ty, flags, }; pack.offset += header_len + stored.len() as u64; pack.index.push((*r, entry)); report.objects += 1; report.raw_bytes += raw_len as u64; report.packed_bytes += stored.len() as u64; Ok(entry) } impl Store { pub fn open(dir: impl Into) -> io::Result { let dir = dir.into(); fs::create_dir_all(dir.join("loose"))?; let store = Store { dir, packs: RefCell::new(Vec::new()), pending: RefCell::new(None), batch_depth: std::cell::Cell::new(0), metas: RefCell::new(None), }; store .reload_packs() .map_err(|e| io::Error::other(e.to_string()))?; Ok(store) } fn reload_packs(&self) -> Result<()> { let mut packs = Vec::new(); let mut names: Vec = fs::read_dir(&self.dir)? .filter_map(|e| e.ok().map(|e| e.path())) .filter(|p| { p.extension().is_some_and(|x| x == "idx") && p.file_name() .is_some_and(|n| n.to_string_lossy().starts_with("pack-")) }) .collect(); names.sort(); for idx in names { let pack = idx.with_extension("pack"); if pack.is_file() { packs.push(PackIndex::load(&idx, &pack)?); } } *self.packs.borrow_mut() = packs; Ok(()) } pub fn pack_count(&self) -> usize { self.packs.borrow().len() } fn find_packed(&self, r: &ObjectRef) -> Option<(usize, PackEntry)> { self.packs .borrow() .iter() .enumerate() .find_map(|(i, p)| p.find(r).map(|e| (i, e))) } /// Loose objects on disk, with their paths. fn loose_objects(&self) -> io::Result> { let mut out = Vec::new(); let loose = self.dir.join("loose"); for d in fs::read_dir(&loose)? { let d = d?; if !d.file_type()?.is_dir() { continue; } let prefix = d.file_name().to_string_lossy().to_string(); for f in fs::read_dir(d.path())? { let f = f?; let name = f.file_name().to_string_lossy().to_string(); if let Some(r) = ObjectRef::from_hex(&format!("{prefix}{name}")) { out.push((r, f.path())); } } } out.sort(); Ok(out) } pub fn loose_count(&self) -> io::Result { Ok(self.loose_objects()?.len()) } /// Move every loose object into a new pack (spec §13.4). Loose files are /// deleted only after the pack and its index are durable. pub fn pack_loose(&self) -> Result { let loose = self.loose_objects()?; self.write_pack( &loose.iter().map(|(r, _)| *r).collect::>(), &loose.iter().map(|(_, p)| p.clone()).collect::>(), &[], ZSTD_LEVEL, ) } /// Fold every pack and loose object into one pack compressed at /// `level` (spec §13.4): what an explicit `oxy gc` does, as `git gc` /// repacks. Returns the report and how many packs were folded in. pub fn repack_all(&self, level: i32) -> Result<(PackReport, usize)> { let loose = self.loose_objects()?; let (mut refs, paths): (Vec, Vec) = { let packs = self.packs.borrow(); ( packs.iter().flat_map(|p| p.refs.iter().copied()).collect(), packs.iter().map(|p| p.path.clone()).collect(), ) }; if loose.is_empty() && paths.len() <= 1 && level == ZSTD_LEVEL { return Ok((PackReport::default(), 0)); } refs.extend(loose.iter().map(|(r, _)| *r)); let loose_paths: Vec = loose.into_iter().map(|(_, p)| p).collect(); let n = paths.len(); let report = self.write_pack(&refs, &loose_paths, &paths, level)?; Ok((report, n)) } /// Fold every pack into one when there are more than `MAX_PACKS` /// (spec §13.4). Returns how many packs were merged. pub fn merge_packs(&self) -> Result { let (refs, paths): (Vec, Vec) = { let packs = self.packs.borrow(); if packs.len() <= MAX_PACKS { return Ok(0); } ( packs.iter().flat_map(|p| p.refs.iter().copied()).collect(), packs.iter().map(|p| p.path.clone()).collect(), ) }; let n = paths.len(); self.write_pack(&refs, &[], &paths, ZSTD_LEVEL)?; Ok(n) } /// Delete every object not in `keep` (spec §21). Loose objects and packs /// younger than `grace` are left alone, since a concurrent writer may not /// have recorded its references yet. Packs holding unreachable objects /// are rewritten without them. Signature sidecars and seal-index entries /// of deleted objects go too. pub fn prune( &self, keep: &std::collections::HashSet, grace: std::time::Duration, ) -> Result { let now = std::time::SystemTime::now(); let old_enough = |m: &fs::Metadata| { m.modified() .map(|t| now.duration_since(t).unwrap_or_default() >= grace) .unwrap_or(true) }; let mut report = PruneReport::default(); // Loose objects. for (r, path) in self.loose_objects()? { if keep.contains(&r) { continue; } let meta = fs::metadata(&path)?; if !old_enough(&meta) { report.deferred += 1; continue; } report.bytes += meta.len(); report.objects += 1; fs::remove_file(&path)?; let _ = fs::remove_dir(path.parent().unwrap()); self.remove_sidecars(&r)?; } // Packs: rewrite those that hold unreachable objects. let mut keep_refs: Vec = Vec::new(); let mut old_paths: Vec = Vec::new(); let mut rewrite = false; for p in self.packs.borrow().iter() { let meta = fs::metadata(&p.path)?; let unreachable = p.refs.iter().filter(|r| !keep.contains(r)).count(); if unreachable == 0 { continue; } if !old_enough(&meta) { report.deferred += unreachable; continue; } rewrite = true; old_paths.push(p.path.clone()); for (r, e) in p.refs.iter().zip(p.entries.iter()) { if keep.contains(r) { keep_refs.push(*r); } else { report.objects += 1; report.bytes += e.stored_len as u64; } } } if rewrite { let dropped: Vec = { let packs = self.packs.borrow(); packs .iter() .filter(|p| old_paths.contains(&p.path)) .flat_map(|p| p.refs.iter().copied()) .filter(|r| !keep.contains(r)) .collect() }; self.write_pack(&keep_refs, &[], &old_paths, ZSTD_LEVEL)?; for r in dropped { self.remove_sidecars(&r)?; } } Ok(report) } fn remove_sidecars(&self, r: &ObjectRef) -> Result<()> { let sidecars = [ self.sig_path(r), self.index_path("seals", r), self.index_path("receipts", r), ]; for path in sidecars { match fs::remove_file(&path) { Ok(()) => { let _ = fs::remove_dir(path.parent().unwrap()); } Err(e) if e.kind() == io::ErrorKind::NotFound => {} Err(e) => return Err(e.into()), } } Ok(()) } /// Write `refs` into a new pack, then delete `loose_paths` and `old_packs`. /// Write the index for a complete, synced temporary pack file and move /// both into place; `index` must be sorted by ObjectRef. fn finish_pack_files( &self, id: &str, tmp_pack: &Path, index: &[(ObjectRef, PackEntry)], ) -> Result<()> { let pack_path = self.dir.join(format!("pack-{id}.pack")); let idx_path = self.dir.join(format!("pack-{id}.idx")); let tmp_idx = self.dir.join(format!("tmp-{id}.idx")); { let mut f = io::BufWriter::new(fs::File::create(&tmp_idx)?); f.write_all(IDX_MAGIC)?; f.write_all(&(index.len() as u32).to_le_bytes())?; for (r, e) in index { f.write_all(&r.0)?; f.write_all(&e.offset.to_le_bytes())?; f.write_all(&e.stored_len.to_le_bytes())?; f.write_all(&e.raw_len.to_le_bytes())?; f.write_all(&[type_index(e.ty), e.flags])?; } f.flush()?; f.get_ref().sync_all()?; } // The pack first: an index never names a pack that is not there. crate::durable::rename(tmp_pack, &pack_path)?; crate::durable::rename(&tmp_idx, &idx_path)?; self.reload_packs() } fn write_pack( &self, refs: &[ObjectRef], loose_paths: &[PathBuf], old_packs: &[PathBuf], level: i32, ) -> Result { if refs.is_empty() { for p in old_packs { remove_pack_files(p)?; } self.reload_packs()?; return Ok(PackReport::default()); } let mut refs: Vec = refs.to_vec(); refs.sort(); refs.dedup(); let mut nonce = [0u8; 8]; rand::rngs::OsRng.fill_bytes(&mut nonce); let id = hex::encode(nonce); let tmp_pack = self.dir.join(format!("tmp-{id}.pack")); let mut report = PackReport::default(); let mut index: Vec<(ObjectRef, PackEntry)> = Vec::with_capacity(refs.len()); // A full repack of enough objects trains a dictionary for them. let dict = if level == REPACK_ZSTD_LEVEL && refs.len() >= DICT_MIN_OBJECTS { let _t = crate::timing::phase("pack: train a dictionary"); self.train_dictionary(&refs) } else { None }; let encoder = dict .as_ref() .map(|d| zstd::dict::EncoderDictionary::copy(d, level)); { let mut f = io::BufWriter::new(fs::File::create(&tmp_pack)?); f.write_all(PACK_MAGIC)?; let mut offset = PACK_MAGIC.len() as u64; // Read a batch (the store is not shared between threads), // compress it on every core, write it in order. let mut batches = refs.chunks(1024); let mut compressed_batch = Vec::new(); loop { if compressed_batch.is_empty() { let Some(batch) = batches.next() else { break }; let mut read: Vec<(ObjectRef, ObjectType, Vec, bool)> = Vec::with_capacity(batch.len()); for r in batch { let (ty, body) = self.get(r)?; read.push((*r, ty, body, self.is_nocompress(r))); } use rayon::prelude::*; compressed_batch = read .into_par_iter() .map_init( || { encoder.as_ref().and_then(|e| { zstd::bulk::Compressor::with_prepared_dictionary(e).ok() }) }, |with_dict, (r, ty, body, raw)| { let packed = if raw { None } else { // Small objects with the dictionary; // a large one has context enough. let dicted = with_dict .as_mut() .filter(|_| body.len() <= DICT_OBJECT_MAX) .and_then(|c| c.compress(&body).ok()) .map(|c| (FLAG_ZSTD | FLAG_DICT, c)); dicted .or_else(|| { zstd::bulk::compress(&body, level) .ok() .map(|c| (FLAG_ZSTD, c)) }) .filter(|(_, c)| c.len() < body.len()) }; (r, ty, body, packed) }, ) .collect::>(); compressed_batch.reverse(); } let Some((r, ty, body, packed)) = compressed_batch.pop() else { continue; }; let r = &r; let (flags, stored) = match packed { Some(c) => c, None => (0, body.clone()), }; let header_len = 1 + 1 + 4 + 4 + 32; f.write_all(&[type_index(ty), flags])?; f.write_all(&(stored.len() as u32).to_le_bytes())?; f.write_all(&(body.len() as u32).to_le_bytes())?; f.write_all(&r.0)?; f.write_all(&stored)?; index.push(( *r, PackEntry { offset: offset + header_len, stored_len: stored.len() as u32, raw_len: body.len() as u32, ty, flags, }, )); offset += header_len + stored.len() as u64; report.objects += 1; report.raw_bytes += body.len() as u64; report.packed_bytes += stored.len() as u64; } f.flush()?; f.get_ref().sync_all()?; } // The dictionary before the pack that needs it becomes visible. if let Some(d) = &dict { crate::durable::write_atomic(&self.dir.join(format!("pack-{id}.dict")), d)?; } self.finish_pack_files(&id, &tmp_pack, &index)?; for p in old_packs { remove_pack_files(p)?; } self.reload_packs()?; for path in loose_paths { fs::remove_file(path)?; if let Some(dir) = path.parent() { let _ = fs::remove_dir(dir); } } Ok(report) } /// A dictionary for these objects, trained on a sample of them (at /// most 10 000, 32 MB); `None` when training fails. fn train_dictionary(&self, refs: &[ObjectRef]) -> Option> { let step = (refs.len() / 10_000).max(1); let mut samples: Vec> = Vec::new(); let mut total = 0usize; for r in refs.iter().step_by(step) { if total > 32 << 20 { break; } let Ok((ty, body)) = self.get(r) else { continue; }; if matches!( ty, ObjectType::Chunk | ObjectType::Blob | ObjectType::Change ) && !body.is_empty() && body.len() < 1 << 20 { total += body.len(); samples.push(body); } } zstd::dict::from_samples(&samples, DICT_BYTES).ok() } pub fn dir(&self) -> &Path { &self.dir } pub(crate) fn path_for(&self, r: &ObjectRef) -> PathBuf { let h = r.to_hex(); self.dir.join("loose").join(&h[..2]).join(&h[2..]) } pub fn has(&self, r: &ObjectRef) -> bool { self.path_for(r).is_file() || self.find_packed(r).is_some() || self .pending .borrow() .as_ref() .is_some_and(|p| p.lookup.contains_key(r)) } /// Start a write batch: until the matching `finish`, `put` collects /// objects and writes them out together, with one `fsync` for a large /// batch (a pack) instead of one per object. Batches nest; only the /// outermost does the work. Objects are readable while pending. pub fn begin_batch(&self) -> Result> { let depth = self.batch_depth.get(); if depth == 0 { *self.pending.borrow_mut() = Some(Pending { buffered: vec![], buffered_bytes: 0, lookup: Default::default(), pack: None, report: PackReport::default(), }); } self.batch_depth.set(depth + 1); Ok(Batch { store: self, done: false, }) } fn finish_batch(&self) -> Result<()> { let depth = self.batch_depth.get(); if depth == 0 { return Ok(()); } self.batch_depth.set(depth - 1); if depth > 1 { return Ok(()); } let Some(pending) = self.pending.borrow_mut().take() else { return Ok(()); }; self.write_pending(pending) } /// Make every object written so far durable, even inside an open /// batch, which stays open. Call before a reference that names them /// is moved (spec §13.6): the operation head, a bookmark pointer. pub fn flush(&self) -> Result<()> { let Some(pending) = self.pending.borrow_mut().take() else { return Ok(()); }; self.write_pending(pending)?; *self.pending.borrow_mut() = Some(Pending { buffered: vec![], buffered_bytes: 0, lookup: Default::default(), pack: None, report: PackReport::default(), }); Ok(()) } fn write_pending(&self, pending: Pending) -> Result<()> { match pending.pack { None => { for (ty, body) in &pending.buffered { self.put_loose(*ty, body)?; } Ok(()) } Some(mut pack) => { pack.writer.flush()?; pack.writer.get_ref().sync_all()?; drop(pack.writer); pack.index.sort_by_key(|(r, _)| *r); self.finish_pack_files(&pack.id, &pack.tmp_path, &pack.index)?; Ok(()) } } } /// Whether a batch is open and holds `r`. fn pending_get(&self, r: &ObjectRef) -> Option)>> { let mut guard = self.pending.borrow_mut(); let pending = guard.as_mut()?; let entry = *pending.lookup.get(r)?; Some(match entry { PendingEntry::Buffered(i) => { let (ty, body) = &pending.buffered[i]; Ok((*ty, body.clone())) } PendingEntry::Packed(e) => { let Some(pack) = pending.pack.as_mut() else { return Some(Err(StoreError::NotFound(*r))); }; match pack.writer.flush() { Ok(()) => PackIndex { path: pack.tmp_path.clone(), refs: vec![], entries: vec![], dict: None, } .read(r, e), Err(err) => Err(err.into()), } } }) } /// Append one object to the pending pack file. fn pending_pack_write( &self, pack: &mut PendingPack, r: &ObjectRef, ty: ObjectType, body: &[u8], report: &mut PackReport, ) -> Result { let (flags, stored) = if self.is_nocompress(r) { (0, body.to_vec()) } else { let compressed = zstd::bulk::compress(body, ZSTD_LEVEL) .map_err(|e| StoreError::Corrupt(*r, format!("zstd: {e}")))?; if compressed.len() < body.len() { (FLAG_ZSTD, compressed) } else { (0, body.to_vec()) } }; write_pack_entry(pack, r, ty, body.len(), flags, &stored, report) } fn put_pending(&self, r: ObjectRef, ty: ObjectType, body: &[u8]) -> Result<()> { let mut guard = self.pending.borrow_mut(); let pending = guard.as_mut().expect("batch open"); if pending.lookup.contains_key(&r) { return Ok(()); } if pending.pack.is_none() && pending.buffered.len() < SPILL_OBJECTS && pending.buffered_bytes + body.len() <= SPILL_BYTES { pending.buffered.push((ty, body.to_vec())); pending.buffered_bytes += body.len(); pending .lookup .insert(r, PendingEntry::Buffered(pending.buffered.len() - 1)); return Ok(()); } self.ensure_pack(pending)?; let mut report = std::mem::take(&mut pending.report); let pack = pending.pack.as_mut().unwrap(); let e = self.pending_pack_write(pack, &r, ty, body, &mut report)?; pending.report = report; pending.lookup.insert(r, PendingEntry::Packed(e)); Ok(()) } /// Open the pending batch's pack, moving the objects buffered so far /// into it (a spill). fn ensure_pack(&self, pending: &mut Pending) -> Result<()> { if pending.pack.is_none() { // Spill: open the pack and move the buffered objects into it. let mut nonce = [0u8; 8]; rand::rngs::OsRng.fill_bytes(&mut nonce); let id = hex::encode(nonce); let tmp_path = self.dir.join(format!("tmp-{id}.pack")); let mut writer = io::BufWriter::new(fs::File::create(&tmp_path)?); writer.write_all(PACK_MAGIC)?; let mut pack = PendingPack { id, tmp_path, writer, offset: PACK_MAGIC.len() as u64, index: vec![], }; let buffered = std::mem::take(&mut pending.buffered); pending.buffered_bytes = 0; let mut report = std::mem::take(&mut pending.report); for (bty, bbody) in &buffered { let br = hash_object(*bty, bbody); let e = self.pending_pack_write(&mut pack, &br, *bty, bbody, &mut report)?; pending.lookup.insert(br, PendingEntry::Packed(e)); } pending.report = report; pending.pack = Some(pack); } Ok(()) } /// Store many objects at once. Inside a large write batch they are /// compressed on every core and appended in order; otherwise each is /// stored as [`Store::put`] does. pub fn put_many(&self, objects: &[(ObjectType, &[u8])]) -> Result> { use rayon::prelude::*; let total: usize = objects.iter().map(|(_, b)| b.len()).sum(); if self.pending.borrow().is_none() || total < SPILL_BYTES { return objects.iter().map(|(ty, b)| self.put(*ty, b)).collect(); } for (ty, b) in objects { if b.len() > MAX_OBJECT_LEN as usize { return Err(StoreError::TooLarge { kind: ty.name(), size: b.len(), limit: MAX_OBJECT_LEN, }); } } let refs: Vec = objects .par_iter() .map(|(ty, b)| hash_object(*ty, b)) .collect(); // What is new: not stored already, not earlier in this call. let mut seen: std::collections::HashSet = std::collections::HashSet::new(); let mut todo: Vec = Vec::new(); for (i, r) in refs.iter().enumerate() { let known = self .pending .borrow() .as_ref() .is_some_and(|p| p.lookup.contains_key(r)); if seen.insert(*r) && !known && !self.path_for(r).is_file() && self.find_packed(r).is_none() { todo.push(i); } } let raw: std::collections::HashSet = match fs::read_dir(self.dir.join("nocompress")) { Ok(rd) => rd .flatten() .filter_map(|e| ObjectRef::from_hex(&e.file_name().to_string_lossy())) .collect(), Err(_) => std::collections::HashSet::new(), }; let packed: Vec<(u8, Vec)> = todo .par_iter() .map(|i| { let body = objects[*i].1; if raw.contains(&refs[*i]) { return (0, body.to_vec()); } match zstd::bulk::compress(body, ZSTD_LEVEL) { Ok(c) if c.len() < body.len() => (FLAG_ZSTD, c), _ => (0, body.to_vec()), } }) .collect(); let mut guard = self.pending.borrow_mut(); let pending = guard.as_mut().expect("batch open"); self.ensure_pack(pending)?; let mut report = std::mem::take(&mut pending.report); let pack = pending.pack.as_mut().unwrap(); for (i, (flags, stored)) in todo.iter().zip(packed) { let (ty, body) = objects[*i]; let e = write_pack_entry(pack, &refs[*i], ty, body.len(), flags, &stored, &mut report)?; pending.lookup.insert(refs[*i], PendingEntry::Packed(e)); } pending.report = report; Ok(refs) } /// Store `body` as an object of type `ty`; idempotent. pub fn put(&self, ty: ObjectType, body: &[u8]) -> Result { // Refuse what the store could not read back (spec §13.2), before // anything is written: a pack must never hold an entry its index // would reject. if body.len() > MAX_OBJECT_LEN as usize { return Err(StoreError::TooLarge { kind: ty.name(), size: body.len(), limit: MAX_OBJECT_LEN, }); } let r = hash_object(ty, body); let path = self.path_for(&r); if path.is_file() || self.find_packed(&r).is_some() { return Ok(r); } if self.pending.borrow().is_some() { self.put_pending(r, ty, body)?; return Ok(r); } self.put_loose(ty, body)?; Ok(r) } /// Write one loose object atomically with its own `fsync`. fn put_loose(&self, ty: ObjectType, body: &[u8]) -> Result { let r = hash_object(ty, body); let path = self.path_for(&r); if path.is_file() { return Ok(r); } let fanout = path.parent().unwrap(); if !fanout.is_dir() { fs::create_dir_all(fanout)?; crate::durable::sync_dir(&self.dir.join("loose"))?; } let mut nonce = [0u8; 8]; rand::rngs::OsRng.fill_bytes(&mut nonce); let tmp = self .dir .join("loose") .join(format!("tmp-{}", hex::encode(nonce))); crate::durable::step("write loose object")?; { let mut f = fs::File::create(&tmp)?; f.write_all(&header(ty, body.len()))?; f.write_all(body)?; f.sync_all()?; } crate::durable::rename(&tmp, &path)?; Ok(r) } /// Read and verify an object, loose, packed, or pending in a batch. pub fn get(&self, r: &ObjectRef) -> Result<(ObjectType, Vec)> { match read_loose_at(&self.path_for(r), r) { Some(found) => found, None => match self.find_packed(r) { Some((i, e)) => self.packs.borrow()[i].read(r, e), None => self.pending_get(r).unwrap_or(Err(StoreError::NotFound(*r))), }, } } /// Many objects at once: loose ones are read and verified in parallel, /// the rest (packed, or pending in a batch) one by one. Results are in /// the order of `refs`. pub fn get_many(&self, refs: &[ObjectRef]) -> Vec> { use rayon::prelude::*; let paths: Vec<(ObjectRef, PathBuf)> = refs.iter().map(|r| (*r, self.path_for(r))).collect(); let mut found: Vec>> = paths.par_iter().map(|(r, p)| read_loose_at(p, r)).collect(); // Packed ones: look them up in order, then read them in parallel. let packs = self.packs.borrow(); let packed: Vec<(usize, usize, PackEntry)> = found .iter() .enumerate() .filter(|(_, f)| f.is_none()) .filter_map(|(i, _)| { packs .iter() .enumerate() .find_map(|(p, pack)| pack.find(&refs[i]).map(|e| (i, p, e))) }) .collect(); let indexes: &[PackIndex] = &packs; let read: Vec<(usize, Result)> = packed .par_iter() .map(|(i, p, e)| (*i, indexes[*p].read(&refs[*i], *e))) .collect(); drop(packs); for (i, r) in read { found[i] = Some(r); } found .into_iter() .zip(refs) .map(|(f, r)| match f { Some(f) => f, None => self.get(r), }) .collect() } pub fn get_typed(&self, r: &ObjectRef, want: ObjectType) -> Result> { let (ty, body) = self.get(r)?; if ty != want { return Err(StoreError::WrongType(*r, ty.name(), want.name())); } Ok(body) } /// Store a change and the text blobs (and their chunks) its large /// inserts are stored as (spec §6.1). pub fn put_change(&self, c: &Change) -> Result { let bytes = c.encode_streaming(|batch| self.put_texts(&batch))?; self.put(ObjectType::Change, &bytes) } /// Store text blobs and their chunks: every chunk at once (compressed /// on every core in a large batch), then the blobs. pub fn put_texts(&self, texts: &[crate::objects::StoredText]) -> Result<()> { let mut chunks: Vec<(ObjectType, &[u8])> = Vec::new(); for t in texts { if !self.has(&t.blob.hash()) { chunks.extend( crate::chunk::chunks(&t.content) .into_iter() .map(|c| (ObjectType::Chunk, c)), ); } } self.put_many(&chunks)?; for t in texts { self.put(ObjectType::Blob, &t.blob.encode())?; } Ok(()) } pub fn get_change(&self, r: &ObjectRef) -> Result { Ok(self.get_change_parts(r)?.0) } /// A change and the objects its stored inserts are made of: each text /// blob and its chunks, which travel and are kept with it. pub fn get_change_parts(&self, r: &ObjectRef) -> Result<(Change, Vec)> { let bytes = self.get_typed(r, ObjectType::Change)?; let mut parts = Vec::new(); let mut failed = None; let decoded = Change::decode_with(&bytes, &mut |text, count, form| match self .read_text(text) { Ok((blob, content)) => { parts.push(*text); parts.extend(blob.chunks); crate::objects::text_lines(&content, count, form) } Err(e) => { let why = e.to_string(); failed = Some(e); Err(ObjectError::Schema(format!( "text blob {}: {why}", text.short(12) ))) } }); match (decoded, failed) { (Ok(c), _) => Ok((c, parts)), (Err(_), Some(StoreError::NotFound(t))) => Err(StoreError::Corrupt( *r, format!("its text blob {} is missing", t.short(12)), )), (Err(e), _) => Err(StoreError::Corrupt(*r, e.to_string())), } } /// A change with every insert left without its lines (empty /// placeholders), and where each one's lines are, in order: for /// history-only states, which read text only for the graphs they build /// (see [`Change::decode_skeleton`]). pub fn get_change_lazy(&self, r: &ObjectRef) -> Result<(Change, Vec)> { let bytes = self.get_typed(r, ObjectType::Change)?; Change::decode_skeleton(&bytes, *r).map_err(|e| StoreError::Corrupt(*r, e.to_string())) } /// A change's metadata (id, dependencies, author, clock, message, kind) /// without its contents: read from its skeleton, so neither its text /// blobs nor its inline lines are read (spec §11). /// /// Served from the `metas` cache when this store read it before (a /// change object can be as large as the tree it adds), as long as the /// store still has the change. pub fn get_change_meta(&self, r: &ObjectRef) -> Result { if self.metas.borrow().is_none() { *self.metas.borrow_mut() = Some(self.read_metas_cache()); } let cached = self.metas.borrow().as_ref().and_then(|m| m.get(r).cloned()); if let Some(meta) = cached { if self.has(r) { #[cfg(test)] assert_eq!( meta, Change::decode_meta(&self.get_typed(r, ObjectType::Change)?).unwrap(), "the metadata cache holds a change's metadata" ); return Ok(meta); } } let bytes = self.get_typed(r, ObjectType::Change)?; let meta = Change::decode_meta(&bytes).map_err(|e| StoreError::Corrupt(*r, e.to_string()))?; self.remember_meta(r, &meta); Ok(meta) } /// The metadata cache: records of a change hash, its metadata in the /// flat layout, and the first 8 bytes of BLAKE3 over both, each /// appended in one write. Derived only: a record that does not check /// ends the reading (a torn write), and the cache can be deleted. fn read_metas_cache(&self) -> crate::fasthash::FastMap { let mut out = crate::fasthash::FastMap::default(); let Ok(bytes) = fs::read(self.dir.join("metas")) else { return out; }; let mut r = crate::flat::Reader::new(&bytes); let mut good = 0; while !r.at_end() { let mut record = || -> std::result::Result<_, ObjectError> { let h = r.object()?; let body = r.bytes()?; let check = r.fixed::<8>()?; Ok((h, body, check)) }; let Ok((h, body, check)) = record() else { break; }; let mut sum = blake3::Hasher::new(); sum.update(&h.0).update(body); if sum.finalize().as_bytes()[..8] != check { break; } match crate::state::ChangeMeta::read_flat(&mut crate::flat::Reader::new(body)) { Ok(m) => out.insert(h, m), Err(_) => break, }; good += 32 + 4 + body.len() + 8; } // Cut a bad tail, so records appended later are read. if good < bytes.len() { let _ = fs::OpenOptions::new() .write(true) .open(self.dir.join("metas")) .and_then(|f| f.set_len(good as u64)); } out } /// Add a change's metadata to the cache; quietly nothing when it /// cannot be written (it is only a cache). fn remember_meta(&self, r: &ObjectRef, meta: &crate::state::ChangeMeta) { let mut body = crate::flat::Writer::default(); meta.write_flat(&mut body); let mut w = crate::flat::Writer::default(); w.object(r); w.bytes(&body.0); let mut sum = blake3::Hasher::new(); sum.update(&r.0).update(&body.0); w.fixed(&sum.finalize().as_bytes()[..8]); let _ = fs::OpenOptions::new() .create(true) .append(true) .open(self.dir.join("metas")) .and_then(|mut f| f.write_all(&w.0)); if let Some(m) = self.metas.borrow_mut().as_mut() { m.insert(*r, meta.clone()); } } fn read_text(&self, r: &ObjectRef) -> Result<(Blob, Vec)> { let blob = Blob::decode(&self.get_typed(r, ObjectType::Blob)?)?; let mut content = Vec::with_capacity(blob.length as usize); for c in &blob.chunks { content.extend(self.get_typed(c, ObjectType::Chunk)?); } Ok((blob, content)) } pub fn put_chunk(&self, bytes: &[u8]) -> Result { self.put(ObjectType::Chunk, bytes) } /// Store a chunk that packs will keep uncompressed (`nocompress`, spec §13.3). pub fn put_chunk_nocompress(&self, bytes: &[u8]) -> Result { let r = hash_object(ObjectType::Chunk, bytes); let dir = self.dir.join("nocompress"); fs::create_dir_all(&dir)?; let _ = fs::File::create(dir.join(r.to_hex())); self.put(ObjectType::Chunk, bytes) } fn is_nocompress(&self, r: &ObjectRef) -> bool { self.dir.join("nocompress").join(r.to_hex()).exists() } pub fn put_state(&self, st: &StateObject) -> Result { self.put(ObjectType::State, &st.encode()) } pub fn get_state(&self, r: &ObjectRef) -> Result { Ok(StateObject::decode(&self.get_typed(r, ObjectType::State)?)?) } pub fn put_attest(&self, a: &Attest) -> Result { self.put(ObjectType::Attest, &a.encode()) } pub fn get_attest(&self, r: &ObjectRef) -> Result { Ok(Attest::decode(&self.get_typed(r, ObjectType::Attest)?)?) } pub fn put_receipt(&self, r: &Receipt) -> Result { self.put(ObjectType::Receipt, &r.encode()) } pub fn get_receipt(&self, r: &ObjectRef) -> Result { Ok(Receipt::decode(&self.get_typed(r, ObjectType::Receipt)?)?) } fn index_path(&self, index: &str, key: &ObjectRef) -> PathBuf { let h = key.to_hex(); self.dir.join(index).join(&h[..2]).join(&h[2..]) } /// Add `value` to the sorted list kept for `key` in `index`. fn index_add(&self, index: &str, key: &ObjectRef, value: &ObjectRef) -> Result<()> { let mut values = self.index_get(index, key)?; if values.contains(value) { return Ok(()); } values.push(*value); values.sort(); let path = self.index_path(index, key); fs::create_dir_all(path.parent().unwrap())?; let text: String = values.iter().map(|s| format!("{}\n", s.to_hex())).collect(); crate::durable::write_atomic(&path, text.as_bytes())?; Ok(()) } fn index_get(&self, index: &str, key: &ObjectRef) -> Result> { match fs::read_to_string(self.index_path(index, key)) { Ok(t) => Ok(t.lines().filter_map(ObjectRef::from_hex).collect()), Err(e) if e.kind() == io::ErrorKind::NotFound => Ok(vec![]), Err(e) => Err(e.into()), } } /// Record that seal attestation `attest` covers `change`. pub fn index_seal(&self, change: &ObjectRef, attest: &ObjectRef) -> Result<()> { self.index_add("seals", change, attest) } /// Seal attestations known to cover `change`. pub fn seals_for(&self, change: &ObjectRef) -> Result> { self.index_get("seals", change) } /// Record receipt attestation `attest` for `state` (spec §14.6). pub fn index_receipt(&self, state: &ObjectRef, attest: &ObjectRef) -> Result<()> { self.index_add("receipts", state, attest) } /// Receipt attestations recorded for `state`. pub fn receipts_for(&self, state: &ObjectRef) -> Result> { self.index_get("receipts", state) } /// Every state with a receipt, for garbage collection. pub fn receipt_states(&self) -> Result> { let mut out = Vec::new(); let dir = self.dir.join("receipts"); let top = match fs::read_dir(&dir) { Ok(d) => d, Err(e) if e.kind() == io::ErrorKind::NotFound => return Ok(out), Err(e) => return Err(e.into()), }; for d in top { let d = d?; let prefix = d.file_name().to_string_lossy().into_owned(); for f in fs::read_dir(d.path())? { let name = f?.file_name().to_string_lossy().into_owned(); if let Some(r) = ObjectRef::from_hex(&format!("{prefix}{name}")) { out.push(r); } } } out.sort(); Ok(out) } pub fn put_identity(&self, id: &Identity) -> Result { self.put(ObjectType::Identity, &id.encode()) } pub fn get_identity(&self, r: &ObjectRef) -> Result { Ok(Identity::decode(&self.get_typed(r, ObjectType::Identity)?)?) } fn sig_path(&self, r: &ObjectRef) -> PathBuf { let h = r.to_hex(); self.dir.join("sigs").join(&h[..2]).join(&h[2..]) } /// Signatures stored alongside object `r` (spec §5.4). Empty if none. pub fn signatures(&self, r: &ObjectRef) -> Result> { match fs::read(self.sig_path(r)) { Ok(b) => Ok(Signature::decode_list(&b)?), Err(e) if e.kind() == io::ErrorKind::NotFound => Ok(vec![]), Err(e) => Err(e.into()), } } /// Attach a signature to object `r`, replacing any earlier one by the same key. pub fn put_signature(&self, r: &ObjectRef, sig: &Signature) -> Result<()> { let mut sigs = self.signatures(r)?; sigs.retain(|s| s.key != sig.key); sigs.push(sig.clone()); sigs.sort_by_key(|s| s.key); let path = self.sig_path(r); fs::create_dir_all(path.parent().unwrap())?; let mut nonce = [0u8; 8]; rand::rngs::OsRng.fill_bytes(&mut nonce); let tmp = self.dir.join(format!("sigs-tmp-{}", hex::encode(nonce))); { let mut f = fs::File::create(&tmp)?; f.write_all(&Signature::encode_list(&sigs))?; f.sync_all()?; } crate::durable::rename(&tmp, &path)?; Ok(()) } /// Every object reference in the store, loose and packed (slow; for gc and tests). pub fn list(&self) -> io::Result> { let mut out: Vec = self.loose_objects()?.into_iter().map(|(r, _)| r).collect(); for p in self.packs.borrow().iter() { out.extend(p.refs.iter().copied()); } out.sort(); out.dedup(); Ok(out) } } /// A loose object at `path`, verified against `r`; `None` when there is no /// such file (it may be packed). fn read_loose_at(path: &Path, r: &ObjectRef) -> Option> { let data = match fs::read(path) { Ok(d) => d, Err(e) if e.kind() == io::ErrorKind::NotFound => return None, Err(e) => return Some(Err(e.into())), }; Some((|| { let nul = data .iter() .position(|&b| b == 0) .ok_or_else(|| StoreError::Corrupt(*r, "no header".into()))?; let head = std::str::from_utf8(&data[..nul]) .map_err(|_| StoreError::Corrupt(*r, "header not UTF-8".into()))?; let mut parts = head.split(' '); let (Some("oxy1"), Some(ty), Some(len), None) = (parts.next(), parts.next(), parts.next(), parts.next()) else { return Err(StoreError::Corrupt(*r, format!("bad header {head:?}"))); }; let ty = ObjectType::from_name(ty) .ok_or_else(|| StoreError::Corrupt(*r, format!("unknown type {ty}")))?; let len: usize = len .parse() .map_err(|_| StoreError::Corrupt(*r, "bad length".into()))?; let body = &data[nul + 1..]; if body.len() != len { return Err(StoreError::Corrupt( *r, format!("length {} != header {len}", body.len()), )); } if hash_object(ty, body) != *r { return Err(StoreError::Corrupt(*r, "hash mismatch".into())); } Ok((ty, body.to_vec())) })()) } #[cfg(test)] mod tests { use super::*; #[test] fn put_get_verify() { let dir = tempfile::tempdir().unwrap(); let s = Store::open(dir.path().join("objects")).unwrap(); let r = s.put(ObjectType::Chunk, b"hello").unwrap(); assert!(s.has(&r)); assert_eq!(s.get(&r).unwrap(), (ObjectType::Chunk, b"hello".to_vec())); assert_eq!(s.put(ObjectType::Chunk, b"hello").unwrap(), r); assert_eq!(s.list().unwrap(), vec![r]); assert!(matches!( s.get_typed(&r, ObjectType::Blob), Err(StoreError::WrongType(..)) )); assert!(matches!( s.get(&ObjectRef([9; 32])), Err(StoreError::NotFound(_)) )); // Tamper: flip a byte, read must fail. let path = s.path_for(&r); let mut data = fs::read(&path).unwrap(); let last = data.len() - 1; data[last] ^= 1; fs::write(&path, data).unwrap(); assert!(matches!(s.get(&r), Err(StoreError::Corrupt(..)))); } #[test] fn packing_keeps_every_object_readable() { let dir = tempfile::tempdir().unwrap(); let s = Store::open(dir.path()).unwrap(); let mut refs = Vec::new(); for i in 0..300u32 { let body = format!("object {i} ").repeat(1 + (i % 7) as usize); refs.push((s.put(ObjectType::Chunk, body.as_bytes()).unwrap(), body)); } let c = crate::objects::tests::sample_change(); let cr = s.put_change(&c).unwrap(); assert_eq!(s.loose_count().unwrap(), 301); let report = s.pack_loose().unwrap(); assert_eq!(report.objects, 301); assert!(report.packed_bytes < report.raw_bytes, "{report:?}"); assert_eq!(s.loose_count().unwrap(), 0); assert_eq!(s.pack_count(), 1); for (r, body) in &refs { assert!(s.has(r)); assert_eq!( s.get(r).unwrap(), (ObjectType::Chunk, body.as_bytes().to_vec()) ); } assert_eq!(s.get_change(&cr).unwrap(), c); assert_eq!(s.list().unwrap().len(), 301); // Re-putting a packed object is a no-op; new objects go loose and a // second pack coexists with the first. assert_eq!( s.put(ObjectType::Chunk, refs[0].1.as_bytes()).unwrap(), refs[0].0 ); assert_eq!(s.loose_count().unwrap(), 0); let extra = s.put(ObjectType::Chunk, b"after packing").unwrap(); s.pack_loose().unwrap(); assert_eq!(s.pack_count(), 2); assert_eq!(s.get(&extra).unwrap().1, b"after packing"); // A reopened store sees the packs. let s2 = Store::open(dir.path()).unwrap(); assert_eq!(s2.pack_count(), 2); assert_eq!(s2.get(&refs[5].0).unwrap().1, refs[5].1.as_bytes()); assert!(matches!( s2.get(&ObjectRef([9; 32])), Err(StoreError::NotFound(_)) )); // Corrupting an object's body inside the pack is detected on read. let target = refs[5].0; let (pack_path, offset) = { let packs = s2.packs.borrow(); let (i, e) = s2.find_packed(&target).unwrap(); (packs[i].path.clone(), e.offset as usize) }; let mut data = fs::read(&pack_path).unwrap(); data[offset] ^= 0xff; fs::write(&pack_path, data).unwrap(); let s3 = Store::open(dir.path()).unwrap(); assert!(matches!(s3.get(&target), Err(StoreError::Corrupt(..)))); assert_eq!( s3.get(&refs[6].0).unwrap().1, refs[6].1.as_bytes(), "other objects are unaffected" ); } #[test] fn a_full_repack_trains_a_dictionary_and_reads_back() { let dir = tempfile::tempdir().unwrap(); let s = Store::open(dir.path()).unwrap(); // Many small, alike files: what a dictionary is for. let bodies: Vec> = (0..1500) .map(|i| { format!( "// SPDX-License-Identifier: GPL-2.0\n#include \n\ static int probe_{i}(struct device *dev)\n{{\n\treturn {i};\n}}\n" ) .into_bytes() }) .collect(); let refs: Vec = bodies .iter() .map(|b| s.put(ObjectType::Chunk, b).unwrap()) .collect(); s.repack_all(REPACK_ZSTD_LEVEL).unwrap(); let dicts = fs::read_dir(dir.path()) .unwrap() .flatten() .filter(|e| e.path().extension().is_some_and(|x| x == "dict")) .count(); assert_eq!(dicts, 1, "the pack has a dictionary"); let with_dict = refs .iter() .filter(|r| s.find_packed(r).unwrap().1.flags & FLAG_DICT != 0) .count(); assert!(with_dict > 1000, "most use it: {with_dict}"); let again = Store::open(dir.path()).unwrap(); for (r, b) in refs.iter().zip(&bodies) { assert_eq!(&again.get(r).unwrap().1, b); } // A repack after that replaces the pack and its dictionary. again.repack_all(REPACK_ZSTD_LEVEL).unwrap(); let files: Vec = fs::read_dir(dir.path()) .unwrap() .flatten() .map(|e| e.file_name().to_string_lossy().into_owned()) .filter(|n| n.starts_with("pack-")) .collect(); assert_eq!(files.len(), 3, "one pack, index and dictionary: {files:?}"); assert_eq!(&again.get(&refs[7]).unwrap().1, &bodies[7]); } #[test] fn nocompress_chunks_pack_raw() { let dir = tempfile::tempdir().unwrap(); let s = Store::open(dir.path()).unwrap(); let body = vec![b'a'; 10_000]; let raw = s.put_chunk_nocompress(&body).unwrap(); let compressible = s.put(ObjectType::Chunk, &vec![b'b'; 10_000]).unwrap(); s.pack_loose().unwrap(); let packs = s.packs.borrow(); let (i, e) = s.find_packed(&raw).unwrap(); assert_eq!((e.flags, e.stored_len as usize), (0, 10_000)); let (_, c) = s.find_packed(&compressible).unwrap(); assert_eq!(c.flags, FLAG_ZSTD); assert!(c.stored_len < 10_000); assert_eq!(packs[i].read(&raw, e).unwrap().1, body); } #[test] fn many_packs_merge_into_one() { let dir = tempfile::tempdir().unwrap(); let s = Store::open(dir.path()).unwrap(); let mut refs = Vec::new(); for i in 0..(MAX_PACKS + 2) { refs.push( s.put(ObjectType::Chunk, format!("pack {i}").as_bytes()) .unwrap(), ); s.pack_loose().unwrap(); } assert_eq!(s.pack_count(), MAX_PACKS + 2); assert_eq!(s.merge_packs().unwrap(), MAX_PACKS + 2); assert_eq!(s.pack_count(), 1); assert_eq!(s.merge_packs().unwrap(), 0); for (i, r) in refs.iter().enumerate() { assert_eq!(s.get(r).unwrap().1, format!("pack {i}").as_bytes()); } assert_eq!(Store::open(dir.path()).unwrap().pack_count(), 1); } #[test] fn prune_removes_unreachable_loose_and_packed_objects() { use std::collections::HashSet; let dir = tempfile::tempdir().unwrap(); let s = Store::open(dir.path()).unwrap(); let keep_a = s.put(ObjectType::Chunk, b"keep a").unwrap(); let drop_a = s.put(ObjectType::Chunk, b"drop a").unwrap(); s.pack_loose().unwrap(); let keep_b = s.put(ObjectType::Chunk, b"keep b").unwrap(); let drop_b = s.put(ObjectType::Chunk, b"drop b").unwrap(); let sk = crate::identity::SecretKey { ed_seed: [1; 32], ml_seed: [2; 32], }; let id = sk.identity("a", "b", 0); let key = s.put_identity(&id).unwrap(); s.put_signature( &drop_b, &sk.sign(crate::identity::Domain::Change, &key, &drop_b), ) .unwrap(); s.index_seal(&drop_a, &ObjectRef([3; 32])).unwrap(); let keep: HashSet = [keep_a, keep_b, key].into_iter().collect(); // Everything is too young under a grace period. let r = s .prune(&keep, std::time::Duration::from_secs(3600)) .unwrap(); assert_eq!((r.objects, r.deferred), (0, 2)); let r = s.prune(&keep, std::time::Duration::ZERO).unwrap(); assert_eq!(r.objects, 2, "{r:?}"); assert!(s.has(&keep_a) && s.has(&keep_b) && s.has(&key)); assert!(!s.has(&drop_a) && !s.has(&drop_b)); assert!(s.signatures(&drop_b).unwrap().is_empty()); assert!(s.seals_for(&drop_a).unwrap().is_empty()); assert_eq!(s.pack_count(), 1); assert_eq!(s.get(&keep_a).unwrap().1, b"keep a"); assert_eq!(s.list().unwrap().len(), 3); } #[test] fn signatures_are_stored_beside_objects() { use crate::identity::{Domain, SecretKey}; let dir = tempfile::tempdir().unwrap(); let s = Store::open(dir.path()).unwrap(); let r = s.put(ObjectType::Chunk, b"x").unwrap(); assert!(s.signatures(&r).unwrap().is_empty()); let sk = SecretKey { ed_seed: [1; 32], ml_seed: [2; 32], }; let id = sk.identity("a", "b", 0); let key = s.put_identity(&id).unwrap(); assert_eq!(s.get_identity(&key).unwrap(), id); let sig = sk.sign(Domain::Change, &key, &r); s.put_signature(&r, &sig).unwrap(); s.put_signature(&r, &sig).unwrap(); assert_eq!(s.signatures(&r).unwrap(), vec![sig]); // The object itself is untouched by its signatures. assert_eq!(s.get(&r).unwrap().1, b"x"); } #[test] fn change_round_trip_through_store() { let dir = tempfile::tempdir().unwrap(); let s = Store::open(dir.path()).unwrap(); let c = crate::objects::tests::sample_change(); let r = s.put_change(&c).unwrap(); assert_eq!(r, c.hash()); assert_eq!(s.get_change(&r).unwrap(), c); } fn xorshift(state: &mut u64) -> u64 { *state ^= *state << 13; *state ^= *state >> 7; *state ^= *state << 17; *state } /// Random corruption of pack and index files never panics: reads fail /// with an error or the store refuses to open, and untouched objects /// stay readable when the index survives. #[test] fn corrupted_packs_and_indexes_fail_cleanly() { let dir = tempfile::tempdir().unwrap(); let objects = dir.path().join("objects"); let s = Store::open(&objects).unwrap(); let mut refs = Vec::new(); let mut seed = 0x9e3779b97f4a7c15u64; for i in 0..40u32 { let len = (xorshift(&mut seed) % 5000) as usize; let mut body = vec![0u8; len]; for b in body.iter_mut() { *b = if i % 2 == 0 { (xorshift(&mut seed) & 0xff) as u8 } else { b'a' + (i % 26) as u8 }; } refs.push(s.put(ObjectType::Chunk, &body).unwrap()); } s.pack_loose().unwrap(); let pack = fs::read_dir(&objects) .unwrap() .filter_map(|e| e.ok().map(|e| e.path())) .find(|p| p.extension().is_some_and(|x| x == "pack")) .unwrap(); let idx = pack.with_extension("idx"); let pristine_pack = fs::read(&pack).unwrap(); let pristine_idx = fs::read(&idx).unwrap(); for round in 0..300u32 { let (path, pristine) = if round % 2 == 0 { (&pack, &pristine_pack) } else { (&idx, &pristine_idx) }; let mut data = pristine.clone(); match xorshift(&mut seed) % 4 { 0 => { let at = (xorshift(&mut seed) as usize) % data.len(); data[at] ^= 1 << (xorshift(&mut seed) % 8); } 1 => { let at = (xorshift(&mut seed) as usize) % data.len(); data[at] = (xorshift(&mut seed) & 0xff) as u8; } 2 => { let keep = (xorshift(&mut seed) as usize) % data.len(); data.truncate(keep); } _ => { let at = (xorshift(&mut seed) as usize) % data.len(); let n = ((xorshift(&mut seed) % 8) + 1) as usize; for k in 0..n.min(data.len() - at) { data[at + k] = 0xff; } } } fs::write(path, &data).unwrap(); // Opening may refuse a broken index; a store that opens must // either return each object intact or report it corrupt/missing. if let Ok(s2) = Store::open(&objects) { for r in &refs { match s2.get(r) { Ok((ty, body)) => assert_eq!( hash_object(ty, &body), *r, "a returned object always matches its reference" ), Err(StoreError::Corrupt(..)) | Err(StoreError::NotFound(_)) | Err(StoreError::Io(_)) => {} Err(e) => panic!("unexpected error {e}"), } } } fs::write(path, pristine).unwrap(); } // Restored files read back in full. let s3 = Store::open(&objects).unwrap(); for r in &refs { let (ty, body) = s3.get(r).unwrap(); assert_eq!(hash_object(ty, &body), *r); } } /// A small batch becomes loose objects; a large one becomes a single /// pack. Objects are readable while pending, and batches nest. #[test] fn batches_write_loose_or_one_pack() { let dir = tempfile::tempdir().unwrap(); let s = Store::open(dir.path().join("objects")).unwrap(); let outer = s.begin_batch().unwrap(); let a = s.put(ObjectType::Chunk, b"small a").unwrap(); { let inner = s.begin_batch().unwrap(); let b = s.put(ObjectType::Chunk, b"small b").unwrap(); assert!(s.has(&b)); assert_eq!(s.get(&b).unwrap().1, b"small b"); inner.finish().unwrap(); assert_eq!( s.loose_count().unwrap(), 0, "the inner batch defers to the outer one" ); } assert_eq!(s.get(&a).unwrap().1, b"small a"); assert_eq!( s.put(ObjectType::Chunk, b"small a").unwrap(), a, "idempotent while pending" ); outer.finish().unwrap(); assert_eq!(s.loose_count().unwrap(), 2); assert_eq!(s.pack_count(), 0); assert_eq!(s.get(&a).unwrap().1, b"small a"); // Large: spills into one pack, readable during and after. let big = s.begin_batch().unwrap(); let mut refs = Vec::new(); let mut seed = 7u64; for i in 0..300u32 { let mut body = vec![b'z'; 40_000]; if i % 3 == 0 { for b in body.iter_mut() { *b = (xorshift(&mut seed) & 0xff) as u8; } } body[..4].copy_from_slice(&i.to_le_bytes()); let r = if i % 5 == 0 { s.put_chunk_nocompress(&body).unwrap() } else { s.put(ObjectType::Chunk, &body).unwrap() }; assert_eq!( s.get(&r).unwrap().1, body, "readable while pending (object {i})" ); refs.push((r, body)); } assert!(s.has(&refs[0].0)); big.finish().unwrap(); assert_eq!(s.pack_count(), 1); assert_eq!( s.loose_count().unwrap(), 2, "nothing from the big batch is loose" ); assert!(fs::read_dir(dir.path().join("objects")) .unwrap() .flatten() .all(|e| !e.file_name().to_string_lossy().starts_with("tmp-"))); let s2 = Store::open(dir.path().join("objects")).unwrap(); for (r, body) in &refs { assert_eq!(s2.get(r).unwrap().1, *body); } // A dropped batch still writes its objects. { let _b = s2.begin_batch().unwrap(); s2.put(ObjectType::Chunk, b"dropped").unwrap(); } assert_eq!( s2.get(&hash_object(ObjectType::Chunk, b"dropped")) .unwrap() .1, b"dropped" ); } #[test] fn get_many_reads_loose_and_packed_objects_in_order() { let dir = tempfile::tempdir().unwrap(); let s = Store::open(dir.path().join("objects")).unwrap(); let packed: Vec = (0..20) .map(|i| { s.put(ObjectType::Chunk, format!("packed {i}").as_bytes()) .unwrap() }) .collect(); s.pack_loose().unwrap(); let loose: Vec = (0..20) .map(|i| { s.put(ObjectType::Chunk, format!("loose {i}").as_bytes()) .unwrap() }) .collect(); let missing = hash_object(ObjectType::Chunk, b"never stored"); let mut refs: Vec = Vec::new(); for i in 0..20 { refs.push(loose[i]); refs.push(packed[i]); } refs.push(missing); let got = s.get_many(&refs); assert_eq!(got.len(), refs.len()); for (i, r) in refs.iter().enumerate().take(40) { let want = if i % 2 == 0 { format!("loose {}", i / 2) } else { format!("packed {}", i / 2) }; let (ty, body) = got[i].as_ref().unwrap(); assert_eq!( (*ty, body.as_slice()), (ObjectType::Chunk, want.as_bytes()), "{r:?}" ); } assert!(matches!(got[40], Err(StoreError::NotFound(_)))); } } #[cfg(test)] mod batch_tests { use super::*; #[test] fn a_large_batch_packs_every_object_and_indexes_them_all() { let dir = tempfile::tempdir().unwrap(); let store = Store::open(dir.path()).unwrap(); let mut refs = Vec::new(); { let batch = store.begin_batch().unwrap(); for i in 0..1000u32 { let body = format!("object {i} {}", "x".repeat(20_000 + i as usize)).into_bytes(); refs.push(store.put(ObjectType::Blob, &body).unwrap()); } batch.finish().unwrap(); } let again = Store::open(dir.path()).unwrap(); for r in &refs { again .get(r) .unwrap_or_else(|e| panic!("{}: {e}", r.short(12))); } } } #[cfg(test)] mod size_tests { use super::*; #[test] fn an_object_over_the_limit_is_refused_and_nothing_is_written() { let dir = tempfile::tempdir().unwrap(); let store = Store::open(dir.path()).unwrap(); let big = vec![b'x'; MAX_OBJECT_LEN as usize + 1]; { let batch = store.begin_batch().unwrap(); store.put(ObjectType::Blob, b"small").unwrap(); let e = store.put(ObjectType::Change, &big).unwrap_err(); assert!( matches!(e, StoreError::TooLarge { kind: "change", .. }), "{e}" ); assert!(e.to_string().contains("commit it in parts"), "{e}"); batch.finish().unwrap(); } // The store still opens and reads what was written. let again = Store::open(dir.path()).unwrap(); again.get(&hash_object(ObjectType::Blob, b"small")).unwrap(); } } #[cfg(test)] mod text_tests { use super::*; use crate::objects::{AtomRef, Edit, CHANGE_VERSION}; fn large_change() -> Change { let mut c = crate::objects::tests::sample_change(); assert_eq!(c.version, CHANGE_VERSION); c.edits[0] = Edit::Insert { node: crate::ids::NodeId([7; 16]), after: AtomRef::Start, before: AtomRef::End, lines: (0..2000) .map(|i| format!("line {i}\n").into_bytes()) .collect(), }; c } #[test] fn a_large_insert_is_stored_beside_its_change_and_read_back() { let dir = tempfile::tempdir().unwrap(); let s = Store::open(dir.path()).unwrap(); let c = large_change(); let r = s.put_change(&c).unwrap(); assert_eq!(r, c.hash()); let (_, body) = s.get(&r).unwrap(); assert!( body.len() < 1024, "the change holds a reference, not {} bytes of lines", body.len() ); let (back, parts) = s.get_change_parts(&r).unwrap(); assert_eq!(back, c); // The text blob and its chunk. assert_eq!(parts.len(), 2); for p in &parts { assert!(s.has(p)); } // Storing it again writes nothing new. let n = s.list().unwrap().len(); s.put_change(&c).unwrap(); assert_eq!(s.list().unwrap().len(), n); } #[test] fn a_change_whose_text_blob_is_missing_is_reported_not_misread() { let dir = tempfile::tempdir().unwrap(); let s = Store::open(dir.path()).unwrap(); let c = large_change(); let r = s.put_change(&c).unwrap(); let (_, parts) = s.get_change_parts(&r).unwrap(); fs::remove_file(s.path_for(&parts[0])).unwrap(); let e = s.get_change(&r).unwrap_err(); assert!( matches!(&e, StoreError::Corrupt(h, why) if *h == r && why.contains("missing")), "{e}" ); } } #[cfg(test)] mod put_many_tests { use super::*; #[test] fn many_objects_in_a_batch_are_compressed_together_and_all_read_back() { let dir = tempfile::tempdir().unwrap(); let s = Store::open(dir.path()).unwrap(); let before = s.put(ObjectType::Chunk, b"already here").unwrap(); let bodies: Vec> = (0..200u32) .map(|i| format!("chunk {i} ").repeat(6000).into_bytes()) .collect(); let mut objects: Vec<(ObjectType, &[u8])> = bodies .iter() .map(|b| (ObjectType::Chunk, b.as_slice())) .collect(); objects.push((ObjectType::Chunk, bodies[0].as_slice())); objects.push((ObjectType::Chunk, b"already here")); let refs = { let batch = s.begin_batch().unwrap(); let refs = s.put_many(&objects).unwrap(); batch.finish().unwrap(); refs }; assert_eq!(refs.len(), objects.len()); assert_eq!(refs[200], refs[0], "a repeat is the same object"); assert_eq!(refs[201], before); let again = Store::open(dir.path()).unwrap(); for (r, b) in refs.iter().zip(&objects) { assert_eq!(again.get(r).unwrap().1, b.1); } } } /// Remove a pack with its index and dictionary. fn remove_pack_files(pack: &Path) -> io::Result<()> { fs::remove_file(pack)?; fs::remove_file(pack.with_extension("idx"))?; match fs::remove_file(pack.with_extension("dict")) { Err(e) if e.kind() != io::ErrorKind::NotFound => Err(e), _ => Ok(()), } }