//! 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::{Change, ObjectError, StateObject}; 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; const ZSTD_LEVEL: i32 = 3; /// 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; #[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. #[derive(Debug)] struct PackIndex { path: PathBuf, refs: Vec, entries: Vec, } 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 || entry.offset.saturating_add(entry.stored_len as u64) > pack_len { return Err(bad("index entry points outside the pack")); } entries.push(entry); } Ok(PackIndex { path: pack_path.to_path_buf(), refs, entries }) } 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_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), } 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, } /// 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(); } } } 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) }; 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::>(), &[]) } /// 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)?; 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)?; for r in dropped { self.remove_sidecars(&r)?; } } Ok(report) } fn remove_sidecars(&self, r: &ObjectRef) -> Result<()> { for path in [self.sig_path(r), self.seal_index_path(r)] { 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()?; } fs::rename(tmp_pack, &pack_path)?; fs::rename(&tmp_idx, &idx_path)?; self.reload_packs() } fn write_pack(&self, refs: &[ObjectRef], loose_paths: &[PathBuf], old_packs: &[PathBuf]) -> Result { if refs.is_empty() { for p in old_packs { fs::remove_file(p)?; fs::remove_file(p.with_extension("idx"))?; } 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()); { let mut f = io::BufWriter::new(fs::File::create(&tmp_pack)?); f.write_all(PACK_MAGIC)?; let mut offset = PACK_MAGIC.len() as u64; for r in &refs { let (ty, body) = self.get(r)?; let (flags, stored) = if self.is_nocompress(r) { (0, body.clone()) } 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.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()?; } self.finish_pack_files(&id, &tmp_pack, &index)?; for p in old_packs { fs::remove_file(p)?; fs::remove_file(p.with_extension("idx"))?; } 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) } pub fn dir(&self) -> &Path { &self.dir } 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(()) }; 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![] }.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()) } }; 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(&(body.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: body.len() as u32, ty, flags }; pack.offset += header_len + stored.len() as u64; pack.index.push((*r, entry)); report.objects += 1; report.raw_bytes += body.len() as u64; report.packed_bytes += stored.len() as u64; Ok(entry) } 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(()); } 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); } 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(()) } /// Store `body` as an object of type `ty`; idempotent. pub fn put(&self, ty: ObjectType, body: &[u8]) -> Result { 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); } fs::create_dir_all(path.parent().unwrap())?; let mut nonce = [0u8; 8]; rand::rngs::OsRng.fill_bytes(&mut nonce); let tmp = self.dir.join("loose").join(format!("tmp-{}", hex::encode(nonce))); { let mut f = fs::File::create(&tmp)?; f.write_all(&header(ty, body.len()))?; f.write_all(body)?; f.sync_all()?; } fs::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)> { let path = self.path_for(r); let data = match fs::read(&path) { Ok(d) => d, Err(e) if e.kind() == io::ErrorKind::NotFound => { return 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))), }; } Err(e) => return Err(e.into()), }; 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())) } 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) } pub fn put_change(&self, c: &Change) -> Result { self.put(ObjectType::Change, &c.encode()) } pub fn get_change(&self, r: &ObjectRef) -> Result { Ok(Change::decode(&self.get_typed(r, ObjectType::Change)?)?) } 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)?)?) } fn seal_index_path(&self, change: &ObjectRef) -> PathBuf { let h = change.to_hex(); self.dir.join("seals").join(&h[..2]).join(&h[2..]) } /// Record that seal attestation `attest` covers `change`. pub fn index_seal(&self, change: &ObjectRef, attest: &ObjectRef) -> Result<()> { let mut seals = self.seals_for(change)?; if seals.contains(attest) { return Ok(()); } seals.push(*attest); seals.sort(); let path = self.seal_index_path(change); fs::create_dir_all(path.parent().unwrap())?; let text: String = seals.iter().map(|s| format!("{}\n", s.to_hex())).collect(); let tmp = path.with_extension("tmp"); fs::write(&tmp, text)?; fs::rename(&tmp, &path)?; Ok(()) } /// Seal attestations known to cover `change`. pub fn seals_for(&self, change: &ObjectRef) -> Result> { match fs::read_to_string(self.seal_index_path(change)) { 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()), } } 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()?; } fs::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) } } #[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 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"); } }