//! Object store (spec §13.1, §13.3, §13.4): loose objects, one per key //! under `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. //! //! Everything is kept on a [`crate::strata::Store`] (docs/stores.md): a //! directory, which is what `.oxy/objects` is, or a bucket, which is how a //! repository lives in its owner's region with the hub as one client of //! it. The keys are the same wherever the store is, so a bucket holds //! exactly what a directory would, and a directory store is as fast as //! the files were before there was a trait: a loose object is one file, a //! packed one is one ranged read of the pack. What must be local, the //! pack being streamed out and the derived `metas` cache, lives in a //! scratch directory. //! //! 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`. //! //! Beside the objects: `nocompress/` marks a chunk packs keep raw; //! `sig///` holds one signature over an object, written //! once per signer; `seal///` and //! `receipt///` index the seals covering a change and //! the receipts on a state. Older stores hold these as one list per //! object (`sigs/`, `seals/`, `receipts/`), which are still read. 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 crate::strata::{self, Error as BackendError, LocalStore, MemoryStore}; use rand::RngCore; use std::cell::RefCell; use std::fs; use std::io::{self, Write}; use std::path::{Path, PathBuf}; use std::sync::Arc; 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); /// Where a long operation reports what it is doing (oxygen#66): a phase, /// or a count within one, as a line for a spinner or a log. pub type Progress<'a> = &'a mut dyn FnMut(&str); /// The storage under a store: objects by key, see [`crate::strata`]. pub type Backend = dyn strata::Store; fn silent() -> impl FnMut(&str) { |_| {} } /// A backend error as a store error: what is not found or already there /// is handled where it matters; the rest is an I/O failure. fn backend_err(e: BackendError) -> StoreError { match e { BackendError::Io(e) => StoreError::Io(e), other => StoreError::Io(io::Error::other(other.to_string())), } } /// `put_new` where an object already there is what we wanted. fn put_new_or_existing(backend: &Backend, key: &str, bytes: &[u8]) -> Result<()> { match backend.put_new(key, bytes) { Ok(()) | Err(BackendError::Exists(_)) => Ok(()), Err(e) => Err(backend_err(e)), } } /// `get` as an option: `None` when there is no such key. fn get_optional(backend: &Backend, key: &str) -> Result>> { match backend.get(key) { Ok(b) => Ok(Some(b)), Err(BackendError::NotFound(_)) => Ok(None), Err(e) => Err(backend_err(e)), } } fn delete(backend: &Backend, key: &str) -> Result<()> { backend.delete(key).map_err(backend_err) } /// `loose//`. fn loose_key(r: &ObjectRef) -> String { let h = r.to_hex(); format!("loose/{}/{}", &h[..2], &h[2..]) } /// The object a loose key names, if it is one. fn loose_ref(key: &str) -> Option { let rest = key.strip_prefix("loose/")?; let (hh, tail) = rest.split_once('/')?; if hh.len() != 2 || tail.contains('/') { return None; } ObjectRef::from_hex(&format!("{hh}{tail}")) } /// `///` for the entries kept under an object. fn fanout_prefix(index: &str, r: &ObjectRef) -> String { let h = r.to_hex(); format!("{index}/{}/{}/", &h[..2], &h[2..]) } /// `//`: the one-file-per-object layout of older stores. fn legacy_key(index: &str, r: &ObjectRef) -> String { let h = r.to_hex(); format!("{index}/{}/{}", &h[..2], &h[2..]) } fn now_ms() -> u64 { std::time::SystemTime::now() .duration_since(std::time::UNIX_EPOCH) .map(|d| d.as_millis() as u64) .unwrap_or(0) } #[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 { /// The pack's key, `pack-.pack`. key: String, 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("key", &self.key) .field("objects", &self.refs.len()) .field("dict", &self.dict.is_some()) .finish() } } /// Decode a stored pack entry and check it against its reference. fn decode_entry( r: &ObjectRef, e: PackEntry, stored: Vec, dict: Option<&Arc>>, ) -> Result { let body = if e.flags & FLAG_DICT != 0 { let dict = dict.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)) } impl PackIndex { fn load(backend: &Backend, idx_key: &str, pack_key: &str) -> Result { let data = backend.get(idx_key).map_err(backend_err)?; let bad = |m: &str| StoreError::Corrupt(ObjectRef::ZERO, format!("{idx_key}: {m}")); 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 = backend .stat(pack_key) .map_err(backend_err)? .ok_or_else(|| bad("the index names a pack that is not there"))? .size; 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!( "{idx_key}: the index claims an object of {} bytes ({} stored), more than the {MAX_OBJECT_LEN} an object may be", 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_key = format!("{}.dict", pack_key.trim_end_matches(".pack")); let dict = get_optional(backend, &dict_key)? .map(|d| Arc::new(zstd::dict::DecoderDictionary::copy(&d))); Ok(PackIndex { key: pack_key.to_string(), refs, entries, dict, }) } fn find(&self, r: &ObjectRef) -> Option { self.refs.binary_search(r).ok().map(|i| self.entries[i]) } /// Read one entry from the pack: one ranged read, then decode. fn fetch(&self, backend: &Backend, r: &ObjectRef, e: PackEntry) -> Result { let stored = backend .get_range(&self.key, e.offset..e.offset + e.stored_len as u64) .map_err(|err| match err { BackendError::NotFound(_) => StoreError::NotFound(*r), other => backend_err(other), })?; decode_entry(r, e, stored, self.dict.as_ref()) } } 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; pub struct Store { backend: Arc, /// Temporary files (a pack being streamed) and derived caches (`metas`). /// For a store in a directory it is that directory. scratch: PathBuf, /// The backend is that directory, so the old file paths still hold. local: bool, 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>>, } impl std::fmt::Debug for Store { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { f.debug_struct("Store") .field("backend", &self.backend.describe()) .field("scratch", &self.scratch) .field("packs", &self.packs.borrow().len()) .finish() } } /// 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) } /// A random name for a temporary file or a pack. fn fresh_id() -> String { let mut nonce = [0u8; 8]; rand::rngs::OsRng.fill_bytes(&mut nonce); hex::encode(nonce) } /// Under `OXY_STORE_BACKEND=memory` every store opens on a store in /// memory shared by directory, so the whole test suite can run without a /// directory store, as it would over a bucket. Objects live for the /// process. fn memory_backend_for(dir: &Path) -> Option> { use std::collections::HashMap; use std::sync::{Mutex, OnceLock}; if std::env::var_os("OXY_STORE_BACKEND").as_deref() != Some(std::ffi::OsStr::new("memory")) { return None; } static STORES: OnceLock>>> = OnceLock::new(); let key = dir.canonicalize().unwrap_or_else(|_| dir.to_path_buf()); let mut map = STORES .get_or_init(Default::default) .lock() .unwrap_or_else(|p| p.into_inner()); let store = map .entry(key) .or_insert_with(|| Arc::new(MemoryStore::new())) .clone(); Some(store) } #[cfg(test)] thread_local! { /// A backend every store opened on this thread uses instead of its /// directory: tests of a repository kept on a store that is not this /// disk's directory, with several handles on one store. pub(crate) static TEST_BACKEND: RefCell>> = const { RefCell::new(None) }; } impl Store { /// Open the store in `dir`, which is also its scratch directory. pub fn open(dir: impl Into) -> io::Result { let dir = dir.into(); fs::create_dir_all(dir.join("loose"))?; #[cfg(test)] if let Some(backend) = TEST_BACKEND.with(|t| t.borrow().clone()) { return Store::open_on(backend, dir); } if let Some(backend) = memory_backend_for(&dir) { return Store::open_on(backend, dir); } let backend: Arc = Arc::new(LocalStore::open(&dir).map_err(|e| io::Error::other(e.to_string()))?); let mut store = Store::open_on(backend, dir)?; store.local = true; Ok(store) } /// Open the store kept on `backend`, with `scratch` for what must be /// on this disk: a pack while it is being written, and caches. pub fn open_on(backend: Arc, scratch: impl Into) -> io::Result { let scratch = scratch.into(); fs::create_dir_all(&scratch)?; let store = Store { backend, scratch, local: false, 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) } /// The storage under this store. pub fn backend(&self) -> &Arc { &self.backend } /// Whether the store is the directory `dir()` names, where every key /// is a file. pub fn is_local(&self) -> bool { self.local } fn reload_packs(&self) -> Result<()> { let listed = self.backend.list("pack-").map_err(backend_err)?; let mut idx_keys: Vec<&str> = listed .iter() .map(|e| e.key.as_str()) .filter(|k| k.ends_with(".idx")) .collect(); idx_keys.sort(); let mut packs = Vec::new(); for idx in idx_keys { let pack = format!("{}.pack", idx.trim_end_matches(".idx")); if listed.iter().any(|e| e.key == pack) { packs.push(PackIndex::load(self.backend.as_ref(), 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, with their keys and ages. fn loose_objects(&self) -> Result> { let mut out: Vec<(ObjectRef, strata::Entry)> = self .backend .list("loose/") .map_err(backend_err)? .into_iter() .filter_map(|e| loose_ref(&e.key).map(|r| (r, e))) .collect(); out.sort_by_key(|(r, _)| *r); Ok(out) } pub fn loose_count(&self) -> io::Result { self.loose_objects() .map(|l| l.len()) .map_err(|e| io::Error::other(e.to_string())) } /// Move every loose object into a new pack (spec §13.4). Loose objects /// 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(|(_, e)| e.key.clone()).collect::>(), &[], ZSTD_LEVEL, &mut silent(), ) } /// 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)> { self.repack_all_with(level, &mut silent()) } /// [`Store::repack_all`], reporting progress. pub fn repack_all_with(&self, level: i32, progress: Progress) -> Result<(PackReport, usize)> { let loose = self.loose_objects()?; let (mut refs, keys, trained): (Vec, Vec, bool) = { let packs = self.packs.borrow(); ( packs.iter().flat_map(|p| p.refs.iter().copied()).collect(), packs.iter().map(|p| p.key.clone()).collect(), packs.first().is_some_and(|p| p.dict.is_some()), ) }; // Nothing loose and one pack already made by a full repack (only // such a pack has a trained dictionary): rewriting it would give // the same pack. A hub's nightly gc rewrote andymccall/oxygen's // 48 000 objects (37 s under its write lock) every time. if loose.is_empty() && keys.len() <= 1 && (level == ZSTD_LEVEL || trained) { return Ok((PackReport::default(), 0)); } refs.extend(loose.iter().map(|(r, _)| *r)); let loose_keys: Vec = loose.into_iter().map(|(_, e)| e.key).collect(); let n = keys.len(); let report = self.write_pack(&refs, &loose_keys, &keys, level, progress)?; 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, keys): (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.key.clone()).collect(), ) }; let n = keys.len(); self.write_pack(&refs, &[], &keys, ZSTD_LEVEL, &mut silent())?; 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. Signatures, seal and receipt entries of /// deleted objects go too. pub fn prune( &self, keep: &std::collections::HashSet, grace: std::time::Duration, ) -> Result { self.prune_with(keep, grace, &mut silent()) } /// [`Store::prune`], reporting progress. pub fn prune_with( &self, keep: &std::collections::HashSet, grace: std::time::Duration, progress: Progress, ) -> Result { progress("pruning loose objects"); let now = now_ms(); let grace_ms = grace.as_millis() as u64; // An object whose age the store does not know is old enough. let old_enough = |modified: Option| modified.is_none_or(|t| now.saturating_sub(t) >= grace_ms); let mut report = PruneReport::default(); // Loose objects. for (r, entry) in self.loose_objects()? { if keep.contains(&r) { continue; } if !old_enough(entry.modified_ms) { report.deferred += 1; continue; } report.bytes += entry.size; report.objects += 1; delete(self.backend.as_ref(), &entry.key)?; self.remove_sidecars(&r)?; } // Packs: rewrite those that hold unreachable objects. let mut keep_refs: Vec = Vec::new(); let mut old_keys: Vec = Vec::new(); let mut rewrite = false; progress("pruning packs"); for p in self.packs.borrow().iter() { let unreachable = p.refs.iter().filter(|r| !keep.contains(r)).count(); if unreachable == 0 { continue; } let modified = self .backend .stat(&p.key) .map_err(backend_err)? .and_then(|e| e.modified_ms); if !old_enough(modified) { report.deferred += unreachable; continue; } rewrite = true; old_keys.push(p.key.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_keys.contains(&p.key)) .flat_map(|p| p.refs.iter().copied()) .filter(|r| !keep.contains(r)) .collect() }; self.write_pack(&keep_refs, &[], &old_keys, ZSTD_LEVEL, progress)?; for r in dropped { self.remove_sidecars(&r)?; } } Ok(report) } /// Everything kept beside an object: its signatures and the seal and /// receipt entries under it, in both layouts. fn remove_sidecars(&self, r: &ObjectRef) -> Result<()> { for index in ["sigs", "seals", "receipts"] { delete(self.backend.as_ref(), &legacy_key(index, r))?; } for index in ["sig", "seal", "receipt"] { for e in self .backend .list(&fanout_prefix(index, r)) .map_err(backend_err)? { delete(self.backend.as_ref(), &e.key)?; } } Ok(()) } /// Put a complete, synced temporary pack file and its index into the /// store; `index` must be sorted by ObjectRef. The pack goes first: an /// index never names a pack that is not there. fn finish_pack_files( &self, id: &str, tmp_pack: &Path, index: &[(ObjectRef, PackEntry)], ) -> Result<()> { let mut idx = Vec::with_capacity(12 + index.len() * IDX_ENTRY); idx.extend_from_slice(IDX_MAGIC); idx.extend_from_slice(&(index.len() as u32).to_le_bytes()); for (r, e) in index { idx.extend_from_slice(&r.0); idx.extend_from_slice(&e.offset.to_le_bytes()); idx.extend_from_slice(&e.stored_len.to_le_bytes()); idx.extend_from_slice(&e.raw_len.to_le_bytes()); idx.extend_from_slice(&[type_index(e.ty), e.flags]); } let put = self .backend .put_new_file(&format!("pack-{id}.pack"), tmp_pack) .map_err(backend_err); // A store on another disk copied it; the one in this directory // moved it and this finds nothing. let _ = fs::remove_file(tmp_pack); put?; self.backend .put_new(&format!("pack-{id}.idx"), &idx) .map_err(backend_err)?; self.reload_packs() } /// Remove a pack with its index and dictionary. fn remove_pack(&self, pack_key: &str) -> Result<()> { let stem = pack_key.trim_end_matches(".pack"); delete(self.backend.as_ref(), pack_key)?; delete(self.backend.as_ref(), &format!("{stem}.idx"))?; delete(self.backend.as_ref(), &format!("{stem}.dict")) } /// Write `refs` into a new pack, then delete `loose_keys` and `old_packs`. fn write_pack( &self, refs: &[ObjectRef], loose_keys: &[String], old_packs: &[String], level: i32, progress: Progress, ) -> Result { if refs.is_empty() { for p in old_packs { self.remove_pack(p)?; } self.reload_packs()?; return Ok(PackReport::default()); } let mut refs: Vec = refs.to_vec(); refs.sort(); refs.dedup(); let id = fresh_id(); let tmp_pack = self.scratch.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"); progress(&format!("training a dictionary on {} objects", refs.len())); self.train_dictionary(&refs) } else { None }; let encoder = dict .as_ref() .map(|d| zstd::dict::EncoderDictionary::copy(d, level)); let nocompress = self.nocompress_set()?; { 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(); let total = refs.len(); let mut reported = usize::MAX; loop { if compressed_batch.is_empty() { let Some(batch) = batches.next() else { break }; // Every tenth, and the first batch. let tenth = report.objects * 10 / total.max(1); if tenth != reported { reported = tenth; progress(&format!("packing {} of {total} objects", report.objects)); } 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, nocompress.contains(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 { self.backend .put_new(&format!("pack-{id}.dict"), d) .map_err(backend_err)?; } progress(&format!("writing the index of {} objects", index.len())); self.finish_pack_files(&id, &tmp_pack, &index)?; for p in old_packs { self.remove_pack(p)?; } self.reload_packs()?; for key in loose_keys { delete(self.backend.as_ref(), key)?; } 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() } /// The scratch directory: for a store in a directory, that directory. pub fn dir(&self) -> &Path { &self.scratch } /// Where a loose object is kept, for a store in a directory (tests /// tamper with and remove objects there). #[cfg(test)] pub(crate) fn path_for(&self, r: &ObjectRef) -> PathBuf { self.scratch.join(loose_key(r)) } /// Remove a loose object, whatever the store is on (tests make an /// object go missing). #[cfg(test)] pub(crate) fn remove_loose_for_test(&self, r: &ObjectRef) -> Result<()> { delete(self.backend.as_ref(), &loose_key(r)) } pub fn has(&self, r: &ObjectRef) -> bool { self.backend.has(&loose_key(r)).unwrap_or(false) || 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 => { let objects: Vec<(ObjectType, &[u8])> = pending .buffered .iter() .map(|(ty, body)| (*ty, body.as_slice())) .collect(); self.put_loose_many(&objects) } 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))); }; // Read from the pack file being written, on this disk. (|| { pack.writer.flush()?; use std::io::{Read, Seek, SeekFrom}; let mut f = fs::File::open(&pack.tmp_path)?; f.seek(SeekFrom::Start(e.offset))?; let mut stored = vec![0u8; e.stored_len as usize]; f.read_exact(&mut stored)?; decode_entry(r, e, stored, None) })() } }) } /// 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 id = fresh_id(); let tmp_path = self.scratch.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.find_packed(r).is_none() && !self.backend.has(&loose_key(r)).unwrap_or(false) { todo.push(i); } } let raw = self.nocompress_set()?; 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); if self.find_packed(&r).is_some() || self.backend.has(&loose_key(&r)).unwrap_or(false) { 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, durable when this returns. fn put_loose(&self, ty: ObjectType, body: &[u8]) -> Result { let r = hash_object(ty, body); let mut bytes = header(ty, body.len()); bytes.extend_from_slice(body); crate::durable::step("write loose object")?; put_new_or_existing(self.backend.as_ref(), &loose_key(&r), &bytes)?; Ok(r) } /// Write several loose objects with as few waits on the disk as the /// guarantees allow (a directory store writes them all, then syncs /// each directory once). Durable when this returns, as `put_loose` /// one at a time would be (spec §13.6). fn put_loose_many(&self, objects: &[(ObjectType, &[u8])]) -> Result<()> { if objects.len() <= 1 { for (ty, body) in objects { self.put_loose(*ty, body)?; } return Ok(()); } let mut seen = std::collections::HashSet::new(); let items: Vec<(String, Vec)> = objects .iter() .filter_map(|(ty, body)| { let r = hash_object(*ty, body); seen.insert(r).then(|| { let mut bytes = header(*ty, body.len()); bytes.extend_from_slice(body); (loose_key(&r), bytes) }) }) .collect(); self.backend.put_many_new(&items).map_err(backend_err) } /// Read and verify an object, packed, loose, or pending in a batch. /// The packs' indexes are in memory, so they are asked first: asking /// for a loose copy of a packed object costs a request that finds /// nothing on a store in a bucket, every time (oxygenhub#228). pub fn get(&self, r: &ObjectRef) -> Result { if let Some((i, e)) = self.find_packed(r) { return self.packs.borrow()[i].fetch(self.backend.as_ref(), r, e); } match get_optional(self.backend.as_ref(), &loose_key(r))? { Some(bytes) => parse_loose(&bytes, r), None => self.pending_get(r).unwrap_or(Err(StoreError::NotFound(*r))), } } /// Many objects at once, read and verified in parallel: packed ones /// from their packs, the others loose (as [`Store::get`], packs first), /// then what is pending in a batch. Results are in the order of `refs`. pub fn get_many(&self, refs: &[ObjectRef]) -> Vec> { use rayon::prelude::*; let backend = self.backend.as_ref(); let packs = self.packs.borrow(); let place: Vec> = refs .iter() .map(|r| { packs .iter() .enumerate() .find_map(|(p, pack)| pack.find(r).map(|e| (p, e))) }) .collect(); let indexes: &[PackIndex] = &packs; let found: Vec>> = refs .par_iter() .zip(place.par_iter()) .map(|(r, at)| match at { Some((p, e)) => Some(indexes[*p].fetch(backend, r, *e)), None => match get_optional(backend, &loose_key(r)) { Ok(Some(bytes)) => Some(parse_loose(&bytes, r)), Ok(None) => None, Err(e) => Some(Err(e)), }, }) .collect(); drop(packs); found .into_iter() .zip(refs) .map(|(f, r)| { f.unwrap_or_else(|| self.pending_get(r).unwrap_or(Err(StoreError::NotFound(*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 encoding = crate::timing::phase("store: encode the change and its texts"); let bytes = c.encode_streaming(|batch| self.put_texts(&batch))?; drop(encoding); let _t = crate::timing::phase("store: put the encoding"); 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. It /// lives in the scratch directory, on this disk. fn read_metas_cache(&self) -> crate::fasthash::FastMap { let mut out = crate::fasthash::FastMap::default(); let Ok(bytes) = fs::read(self.scratch.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.scratch.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.scratch.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); put_new_or_existing( self.backend.as_ref(), &format!("nocompress/{}", r.to_hex()), b"", )?; self.put(ObjectType::Chunk, bytes) } fn is_nocompress(&self, r: &ObjectRef) -> bool { self.backend .has(&format!("nocompress/{}", r.to_hex())) .unwrap_or(false) } /// Every chunk marked to be kept raw. fn nocompress_set(&self) -> Result> { Ok(self .backend .list("nocompress/") .map_err(backend_err)? .into_iter() .filter_map(|e| ObjectRef::from_hex(e.key.trim_start_matches("nocompress/"))) .collect()) } 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)?)?) } /// Record `value` under `key` in `index`: one write-once entry, so two /// writers of a shared store never lose each other's. fn index_add(&self, index: &str, key: &ObjectRef, value: &ObjectRef) -> Result<()> { let entry = format!("{}{}", fanout_prefix(index, key), value.to_hex()); put_new_or_existing(self.backend.as_ref(), &entry, b"") } /// The values recorded under `key` in `index`, with those an older /// store kept as one list in `legacy`, sorted. fn index_get(&self, index: &str, legacy: &str, key: &ObjectRef) -> Result> { let mut values: Vec = self .backend .list(&fanout_prefix(index, key)) .map_err(backend_err)? .into_iter() .filter_map(|e| e.key.rsplit('/').next().and_then(ObjectRef::from_hex)) .collect(); if let Some(bytes) = get_optional(self.backend.as_ref(), &legacy_key(legacy, key))? { values.extend( String::from_utf8_lossy(&bytes) .lines() .filter_map(ObjectRef::from_hex), ); } values.sort(); values.dedup(); Ok(values) } /// Record that seal attestation `attest` covers `change`. pub fn index_seal(&self, change: &ObjectRef, attest: &ObjectRef) -> Result<()> { self.index_add("seal", change, attest) } /// Seal attestations known to cover `change`. pub fn seals_for(&self, change: &ObjectRef) -> Result> { self.index_get("seal", "seals", change) } /// Record receipt attestation `attest` for `state` (spec §14.6). pub fn index_receipt(&self, state: &ObjectRef, attest: &ObjectRef) -> Result<()> { self.index_add("receipt", state, attest) } /// Receipt attestations recorded for `state`. pub fn receipts_for(&self, state: &ObjectRef) -> Result> { self.index_get("receipt", "receipts", state) } pub fn index_approval(&self, state: &ObjectRef, attest: &ObjectRef) -> Result<()> { self.index_add("approval", state, attest) } /// Approval attestations recorded for `state` (OxygenFlow approval /// gates, oxygenhub#235). pub fn approvals_for(&self, state: &ObjectRef) -> Result> { self.index_get("approval", "approvals", state) } /// Every state with an approval, for garbage collection. pub fn approval_states(&self) -> Result> { self.indexed_states("approval", "approvals") } /// Every state with a receipt, for garbage collection. pub fn receipt_states(&self) -> Result> { self.indexed_states("receipt", "receipts") } /// The states an index (`one///` entries, and the /// packed `many//` lists) holds attestations for. fn indexed_states(&self, one: &str, many: &str) -> Result> { let mut out: Vec = Vec::new(); for e in self.backend.list(&format!("{one}/")).map_err(backend_err)? { // receipt/// let mut parts = e.key.splitn(4, '/'); let (Some(_), Some(hh), Some(rest), Some(_)) = (parts.next(), parts.next(), parts.next(), parts.next()) else { continue; }; out.extend(ObjectRef::from_hex(&format!("{hh}{rest}"))); } for e in self .backend .list(&format!("{many}/")) .map_err(backend_err)? { let mut parts = e.key.splitn(3, '/'); let (Some(_), Some(hh), Some(rest)) = (parts.next(), parts.next(), parts.next()) else { continue; }; out.extend(ObjectRef::from_hex(&format!("{hh}{rest}"))); } out.sort(); out.dedup(); 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)?)?) } /// Signatures stored alongside object `r` (spec §5.4), one per signing /// key, sorted by key. Empty if none. pub fn signatures(&self, r: &ObjectRef) -> Result> { let mut sigs: Vec = Vec::new(); for e in self .backend .list(&fanout_prefix("sig", r)) .map_err(backend_err)? { let bytes = self.backend.get(&e.key).map_err(backend_err)?; sigs.extend(Signature::decode_list(&bytes)?); } // An older store's one list per object; a signer kept there and // here counts once, as written here. if let Some(bytes) = get_optional(self.backend.as_ref(), &legacy_key("sigs", r))? { for sig in Signature::decode_list(&bytes)? { if !sigs.iter().any(|s| s.key == sig.key) { sigs.push(sig); } } } sigs.sort_by_key(|s| s.key); sigs.dedup_by_key(|s| s.key); Ok(sigs) } /// Attach a signature to object `r`, replacing any earlier one by the /// same key: one entry per signer, so the latest by a key is the one /// kept, and two signers never touch each other's. pub fn put_signature(&self, r: &ObjectRef, sig: &Signature) -> Result<()> { let key = format!("{}{}", fanout_prefix("sig", r), sig.key.to_hex()); let bytes = Signature::encode_list(std::slice::from_ref(sig)); if get_optional(self.backend.as_ref(), &key)?.as_deref() == Some(bytes.as_slice()) { return Ok(()); } delete(self.backend.as_ref(), &key)?; put_new_or_existing(self.backend.as_ref(), &key, &bytes) } /// Finish packs a writer left unfinished: a batch writes its objects to /// `.tmp-ID.pack` and only indexes it at the end, so a process that /// stopped part way (an import killed after hours) leaves every object /// in a pack nothing can read. Each such pack untouched for `min_age` /// is read through: every whole entry checked against its reference, /// a torn last one cut off, then indexed as a pack of the store. /// Returns the packs and objects adopted. pub fn adopt_unfinished_packs( &self, min_age: std::time::Duration, progress: &dyn Fn(&str), ) -> Result<(usize, usize)> { use std::io::{BufReader, Read}; let mut packs = 0; let mut objects = 0; let Ok(dir) = fs::read_dir(&self.scratch) else { return Ok((0, 0)); }; for entry in dir.flatten() { let name = entry.file_name().to_string_lossy().to_string(); let Some(id) = name .strip_prefix(".tmp-") .and_then(|n| n.strip_suffix(".pack")) else { continue; }; let path = entry.path(); let meta = fs::metadata(&path)?; let age = meta .modified() .ok() .and_then(|m| m.elapsed().ok()) .unwrap_or_default(); if age < min_age { progress(&format!( "{name}: written to {}s ago, left alone", age.as_secs() )); continue; } let mut f = BufReader::with_capacity(1 << 20, fs::File::open(&path)?); let mut magic = [0u8; 8]; if f.read_exact(&mut magic).is_err() || &magic != PACK_MAGIC { progress(&format!("{name}: not a pack, left alone")); continue; } let mut offset = PACK_MAGIC.len() as u64; let mut index: Vec<(ObjectRef, PackEntry)> = Vec::new(); loop { let mut head = [0u8; 42]; if f.read_exact(&mut head).is_err() { break; } let Some(ty) = ObjectType::ALL.get(head[0] as usize).copied() else { break; }; let flags = head[1]; let stored_len = u32::from_le_bytes(head[2..6].try_into().unwrap()); let raw_len = u32::from_le_bytes(head[6..10].try_into().unwrap()); let r = ObjectRef(head[10..42].try_into().unwrap()); let mut stored = vec![0u8; stored_len as usize]; if flags & FLAG_DICT != 0 || f.read_exact(&mut stored).is_err() { break; } let e = PackEntry { offset: offset + 42, stored_len, raw_len, ty, flags, }; if decode_entry(&r, e, stored, None).is_err() { break; } index.push((r, e)); offset += 42 + stored_len as u64; if index.len().is_multiple_of(200_000) { progress(&format!( "{name}: {} objects checked, {} MiB", index.len(), offset >> 20 )); } } if index.is_empty() { progress(&format!("{name}: no whole object, left alone")); continue; } if offset < meta.len() { progress(&format!( "{name}: {} torn bytes at the end cut off", meta.len() - offset )); fs::OpenOptions::new() .write(true) .open(&path)? .set_len(offset)?; } index.sort_by_key(|(r, _)| *r); index.dedup_by_key(|(r, _)| *r); objects += index.len(); packs += 1; progress(&format!("{name}: {} objects indexed", index.len())); self.finish_pack_files(id, &path, &index)?; } Ok((packs, objects)) } /// 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() .map_err(|e| io::Error::other(e.to_string()))? .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's bytes, verified against `r`. fn parse_loose(data: &[u8], r: &ObjectRef) -> Result { 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())) } /// Whether the objects are files that a test may tamper with: not under /// `OXY_STORE_BACKEND=memory`. #[cfg(test)] fn files_on_disk() -> bool { std::env::var_os("OXY_STORE_BACKEND").is_none() } #[cfg(test)] mod tests { use super::*; /// The store under test twice: in a directory, and on a store in /// memory with a scratch directory, as over a bucket. fn each_store(dir: &Path, f: impl Fn(&str, Store)) { f("directory", Store::open(dir.join("objects")).unwrap()); f( "memory", Store::open_on(Arc::new(MemoryStore::new()), dir.join("scratch")).unwrap(), ); } #[test] fn put_get_verify() { if !super::files_on_disk() { return; } 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() { if !super::files_on_disk() { return; } 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(); (dir.path().join(&packs[i].key), 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() { if !super::files_on_disk() { return; } 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); } let packs = |dir: &Path| { let mut v: Vec = fs::read_dir(dir) .unwrap() .flatten() .map(|e| e.file_name().to_string_lossy().into_owned()) .filter(|n| n.starts_with("pack-")) .collect(); v.sort(); v }; // With nothing new, a repack leaves that pack as it is: it would // only write the same again. let before = packs(dir.path()); let (report, merged) = again.repack_all(REPACK_ZSTD_LEVEL).unwrap(); assert_eq!((report.objects, merged), (0, 0)); assert_eq!(packs(dir.path()), before); // One new object: a repack replaces the pack and its dictionary. let extra = again.put(ObjectType::Chunk, b"one more\n").unwrap(); again.repack_all(REPACK_ZSTD_LEVEL).unwrap(); let files = packs(dir.path()); assert_eq!(files.len(), 3, "one pack, index and dictionary: {files:?}"); assert_ne!(files, before); assert_eq!(&again.get(&refs[7]).unwrap().1, &bodies[7]); assert_eq!(again.get(&extra).unwrap().1, b"one more\n"); } #[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].fetch(s.backend.as_ref(), &raw, e).unwrap().1, body); } #[test] fn many_packs_merge_into_one() { let dir = tempfile::tempdir().unwrap(); each_store(dir.path(), |which, s| { 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, "{which}"); 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()); } let again = Store::open_on(s.backend().clone(), s.dir()).unwrap(); assert_eq!(again.pack_count(), 1, "{which}: reopened"); }); } #[test] fn prune_removes_unreachable_loose_and_packed_objects() { use std::collections::HashSet; let dir = tempfile::tempdir().unwrap(); each_store(dir.path(), |which, s| { 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), "{which}"); let r = s.prune(&keep, std::time::Duration::ZERO).unwrap(); assert_eq!(r.objects, 2, "{which}: {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(); each_store(dir.path(), |which, s| { 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.clone()], "{which}"); // A second signer is kept beside the first, sorted by key. let sk2 = SecretKey { ed_seed: [3; 32], ml_seed: [4; 32], }; let key2 = s.put_identity(&sk2.identity("c", "d", 0)).unwrap(); let sig2 = sk2.sign(Domain::Change, &key2, &r); s.put_signature(&r, &sig2).unwrap(); let mut both = vec![sig.clone(), sig2.clone()]; both.sort_by_key(|s| s.key); assert_eq!(s.signatures(&r).unwrap(), both); // The object itself is untouched by its signatures. assert_eq!(s.get(&r).unwrap().1, b"x"); // Seal and receipt entries, and the states with receipts. s.index_seal(&r, &ObjectRef([5; 32])).unwrap(); s.index_seal(&r, &ObjectRef([4; 32])).unwrap(); s.index_seal(&r, &ObjectRef([5; 32])).unwrap(); assert_eq!( s.seals_for(&r).unwrap(), vec![ObjectRef([4; 32]), ObjectRef([5; 32])] ); let st = ObjectRef([6; 32]); s.index_receipt(&st, &ObjectRef([7; 32])).unwrap(); assert_eq!(s.receipts_for(&st).unwrap(), vec![ObjectRef([7; 32])]); assert_eq!(s.receipt_states().unwrap(), vec![st]); }); } /// A store written before signatures, seals and receipts were one /// entry each (`sigs/`, `seals/`, `receipts/` lists) still reads, and /// what is written now sits beside what was there. #[test] fn an_older_store_s_sidecar_lists_are_still_read() { if !super::files_on_disk() { return; } use crate::identity::{Domain, SecretKey}; let dir = tempfile::tempdir().unwrap(); let s = Store::open(dir.path()).unwrap(); let r = s.put(ObjectType::Chunk, b"old").unwrap(); let h = r.to_hex(); let sk = SecretKey { ed_seed: [1; 32], ml_seed: [2; 32], }; let key = s.put_identity(&sk.identity("a", "b", 0)).unwrap(); let sig = sk.sign(Domain::Change, &key, &r); let legacy = dir.path().join("sigs").join(&h[..2]); fs::create_dir_all(&legacy).unwrap(); fs::write( legacy.join(&h[2..]), Signature::encode_list(std::slice::from_ref(&sig)), ) .unwrap(); let seals = dir.path().join("seals").join(&h[..2]); fs::create_dir_all(&seals).unwrap(); fs::write( seals.join(&h[2..]), format!("{}\n", ObjectRef([9; 32]).to_hex()), ) .unwrap(); let receipts = dir.path().join("receipts").join(&h[..2]); fs::create_dir_all(&receipts).unwrap(); fs::write( receipts.join(&h[2..]), format!("{}\n", ObjectRef([8; 32]).to_hex()), ) .unwrap(); assert_eq!(s.signatures(&r).unwrap(), vec![sig.clone()]); assert_eq!(s.seals_for(&r).unwrap(), vec![ObjectRef([9; 32])]); assert_eq!(s.receipts_for(&r).unwrap(), vec![ObjectRef([8; 32])]); assert_eq!(s.receipt_states().unwrap(), vec![r]); // New entries join the old ones; the same signer is not doubled. s.put_signature(&r, &sig).unwrap(); s.index_seal(&r, &ObjectRef([1; 32])).unwrap(); assert_eq!(s.signatures(&r).unwrap().len(), 1); assert_eq!( s.seals_for(&r).unwrap(), vec![ObjectRef([1; 32]), ObjectRef([9; 32])] ); // Pruning the object takes both layouts with it. let r2 = s.put(ObjectType::Chunk, b"kept").unwrap(); let keep = [r2, key].into_iter().collect(); s.prune(&keep, std::time::Duration::ZERO).unwrap(); assert!(s.signatures(&r).unwrap().is_empty()); assert!(s.seals_for(&r).unwrap().is_empty()); assert!(s.receipt_states().unwrap().is_empty()); } #[test] fn change_round_trip_through_store() { let dir = tempfile::tempdir().unwrap(); each_store(dir.path(), |which, s| { let c = crate::objects::tests::sample_change(); let r = s.put_change(&c).unwrap(); assert_eq!(r, c.hash(), "{which}"); assert_eq!(s.get_change(&r).unwrap(), c); assert_eq!(s.get_change_meta(&r).unwrap().message, c.message); }); } 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() { if !super::files_on_disk() { return; } 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(); each_store(dir.path(), |which, s| { 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, "{which}: 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(s.dir()) .unwrap() .flatten() .all(|e| !e.file_name().to_string_lossy().contains("tmp-"))); let s2 = Store::open_on(s.backend().clone(), s.dir()).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(); each_store(dir.path(), |which, s| { 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()), "{which}: {r:?}" ); } assert!(matches!(got[40], Err(StoreError::NotFound(_)))); }); } /// A packed object is read from its pack without first asking the /// backend for a loose copy (oxygenhub#228: on a bucket each such ask /// was a request that found nothing, every time). #[test] fn packed_objects_cost_no_loose_lookups() { use std::sync::atomic::{AtomicUsize, Ordering}; struct Counting(MemoryStore, AtomicUsize); impl crate::strata::Store for Counting { fn describe(&self) -> String { "counting".into() } fn has(&self, k: &str) -> crate::strata::Result { self.0.has(k) } fn get(&self, k: &str) -> crate::strata::Result> { if k.starts_with("loose/") { self.1.fetch_add(1, Ordering::Relaxed); } self.0.get(k) } fn get_range( &self, k: &str, r: std::ops::Range, ) -> crate::strata::Result> { self.0.get_range(k, r) } fn put_new(&self, k: &str, b: &[u8]) -> crate::strata::Result<()> { self.0.put_new(k, b) } fn list(&self, p: &str) -> crate::strata::Result> { self.0.list(p) } fn delete(&self, k: &str) -> crate::strata::Result<()> { self.0.delete(k) } fn read_ref( &self, k: &str, ) -> crate::strata::Result, crate::strata::Version)>> { self.0.read_ref(k) } fn write_ref( &self, k: &str, b: &[u8], e: Option<&crate::strata::Version>, ) -> crate::strata::Result { self.0.write_ref(k, b, e) } } let dir = tempfile::tempdir().unwrap(); let backend = Arc::new(Counting(MemoryStore::new(), AtomicUsize::new(0))); let s = Store::open_on(backend.clone(), dir.path().join("scratch")).unwrap(); let refs: Vec = (0..10) .map(|i| { s.put(ObjectType::Chunk, format!("p {i}").as_bytes()) .unwrap() }) .collect(); s.pack_loose().unwrap(); let s = Store::open_on(backend.clone(), dir.path().join("scratch")).unwrap(); backend.1.store(0, Ordering::Relaxed); for r in &refs { assert!(s.get(r).is_ok()); } assert!(s.get_many(&refs).iter().all(|g| g.is_ok())); assert_eq!(backend.1.load(Ordering::Relaxed), 0, "no loose lookups"); // An object no pack holds is still looked for loose. let missing = hash_object(ObjectType::Chunk, b"never stored"); assert!(matches!(s.get(&missing), Err(StoreError::NotFound(_)))); assert!(backend.1.load(Ordering::Relaxed) > 0); } } #[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))); } } #[test] fn a_pack_a_stopped_batch_left_is_adopted_and_its_torn_end_cut() { // oxygen#79: a batch never finished (a process killed part way) // leaves its objects in an unindexed .tmp pack. let dir = tempfile::tempdir().unwrap(); let mut refs = Vec::new(); { let store = Store::open(dir.path()).unwrap(); let batch = store.begin_batch().unwrap(); for i in 0..300u32 { let body = format!("object {i} {}", "y".repeat(20_000 + i as usize)).into_bytes(); refs.push(store.put(ObjectType::Blob, &body).unwrap()); } std::mem::forget(batch); } let tmp: Vec<_> = fs::read_dir(dir.path()) .unwrap() .flatten() .filter(|e| e.file_name().to_string_lossy().starts_with(".tmp-")) .map(|e| e.path()) .collect(); assert_eq!(tmp.len(), 1, "one unfinished pack"); // A torn last entry: a header and part of its body. let mut f = fs::OpenOptions::new().append(true).open(&tmp[0]).unwrap(); f.write_all(&[0, 1, 0xff, 0, 0, 0, 9, 9, 0, 0]).unwrap(); f.write_all(&[7; 40]).unwrap(); drop(f); let store = Store::open(dir.path()).unwrap(); assert!(store.get(&refs[0]).is_err(), "unreadable before"); let young = store .adopt_unfinished_packs(std::time::Duration::from_secs(3600), &|_| {}) .unwrap(); assert_eq!(young, (0, 0), "a pack written to lately is left alone"); let (packs, objects) = store .adopt_unfinished_packs(std::time::Duration::ZERO, &|_| {}) .unwrap(); assert_eq!((packs, objects), (1, 300)); for r in &refs { store .get(r) .unwrap_or_else(|e| panic!("{}: {e}", r.short(12))); } let again = Store::open(dir.path()).unwrap(); assert!(again.get(&refs[299]).is_ok(), "indexed for good"); } } #[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() { if !super::files_on_disk() { return; } 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); } } }