//! The object store on disk: one file per object, named by its ID, shared by //! every project and workspace on the machine. Files are written once and //! never change, so readers never need a lock. use std::fs; use std::io::Write; use std::path::{Path, PathBuf}; use anyhow::{Context, Result, anyhow}; use vcs_core::bundle::{self, Frame, RAW, ZSTD}; use vcs_core::object::{Kind, Object}; use vcs_core::{Id, Source, chunk}; /// Objects at least this big are compressed on disk and on the wire. const COMPRESS_OVER: usize = 512; #[derive(Clone)] pub struct Store { root: PathBuf, } impl Store { pub fn open(root: &Path) -> Result { fs::create_dir_all(root).with_context(|| format!("creating {}", root.display()))?; Ok(Store { root: root.to_path_buf() }) } fn path(&self, id: &Id) -> PathBuf { let hex = id.hex(); self.root.join(&hex[..2]).join(&hex[2..]) } pub fn has(&self, id: &Id) -> bool { self.path(id).is_file() } /// How many bytes the object takes here (0 when it's missing). pub fn stored_len(&self, id: &Id) -> u64 { fs::metadata(self.path(id)).map(|m| m.len()).unwrap_or(0) } /// Stores an object (no-op if present). Returns its ID. pub fn put(&self, object: &Object) -> Result { let id = object.id(); if self.has(&id) { return Ok(id); } self.write_frame(&encode(object, id))?; Ok(id) } fn write_frame(&self, frame: &Frame) -> Result<()> { let path = self.path(&frame.id); if path.is_file() { return Ok(()); } let dir = path.parent().unwrap(); fs::create_dir_all(dir)?; let mut bytes = Vec::with_capacity(frame.payload.len() + 10); bytes.push(frame.kind as u8); bytes.push(frame.codec); bytes.extend_from_slice(&frame.raw_len.to_le_bytes()); bytes.extend_from_slice(&frame.payload); let tmp = dir.join(format!(".tmp-{}-{}", std::process::id(), rand::random::())); { let mut f = fs::File::create(&tmp)?; f.write_all(&bytes)?; } fs::rename(&tmp, &path)?; Ok(()) } /// The stored frame (possibly compressed), for bundles. pub fn frame(&self, id: &Id) -> Result> { let bytes = match fs::read(self.path(id)) { Ok(b) => b, Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(None), Err(e) => return Err(e.into()), }; if bytes.len() < 10 { return Err(anyhow!("object {} is damaged", id.short())); } let kind = Kind::from_u8(bytes[0]).ok_or_else(|| anyhow!("object {} is damaged", id.short()))?; let codec = bytes[1]; let raw_len = u64::from_le_bytes(bytes[2..10].try_into().unwrap()); Ok(Some(Frame { id: *id, kind, codec, raw_len, payload: bytes[10..].to_vec() })) } pub fn read(&self, id: &Id) -> Result> { match self.frame(id)? { Some(frame) => Ok(Some(decode(&frame)?)), None => Ok(None), } } /// Stores a file's content (chunked when big) and returns its ID. pub fn put_file(&self, content: &[u8]) -> Result { let objects = chunk::file_objects(content); let id = objects[0].id(); if self.has(&id) { return Ok(id); } // Chunks first, so the list never points at missing chunks. for o in objects.iter().skip(1) { self.put(o)?; } self.put(&objects[0])?; Ok(id) } pub fn read_file(&self, id: &Id) -> Result> { Ok(chunk::read_file(self, id)?) } /// Every object a file ID needs (itself plus chunks). pub fn file_parts(&self, id: &Id) -> Result> { let mut out = vec![*id]; // The first byte says whether it's a chunk list; blobs aren't read. let mut kind = [0u8; 1]; match fs::File::open(self.path(id)).and_then(|mut f| std::io::Read::read_exact(&mut f, &mut kind)) { Ok(()) if kind[0] == Kind::Chunks as u8 => {} Ok(()) => return Ok(out), Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(out), Err(e) => return Err(e.into()), } if let Some(frame) = self.frame(id)? && frame.kind == Kind::Chunks { let o = decode(&frame)?; out.extend(vcs_core::object::decode_chunks(&o.data)?.into_iter().map(|(c, _)| c)); } Ok(out) } /// A bundle of the given objects (those present). pub fn bundle(&self, ids: &[Id]) -> Result> { let mut out = Vec::new(); bundle::write_header(&mut out); for id in ids { if let Some(frame) = self.frame(id)? { bundle::write_frame(&mut out, &frame); } } bundle::write_end(&mut out); Ok(out) } /// Stores every object in a bundle after checking it. Returns how many. pub fn ingest(&self, bytes: &[u8]) -> Result { let frames = bundle::read(bytes)?; let n = frames.len(); for frame in frames { if self.has(&frame.id) { continue; } let object = decode(&frame)?; bundle::verify(&frame, &object.data)?; self.write_frame(&frame)?; } Ok(n) } } impl Source for Store { fn get(&self, id: &Id) -> vcs_core::Result> { self.read(id).map_err(|e| vcs_core::Error::Corrupt(format!("{e:#}"))) } } pub fn encode(object: &Object, id: Id) -> Frame { if object.data.len() >= COMPRESS_OVER && let Ok(packed) = zstd::bulk::compress(&object.data, 3) && packed.len() < object.data.len() { return Frame { id, kind: object.kind, codec: ZSTD, raw_len: object.data.len() as u64, payload: packed }; } Frame { id, kind: object.kind, codec: RAW, raw_len: object.data.len() as u64, payload: object.data.clone() } } pub fn decode(frame: &Frame) -> Result { let data = match frame.codec { RAW => frame.payload.clone(), ZSTD => zstd::bulk::decompress(&frame.payload, frame.raw_len as usize).context("damaged object")?, _ => return Err(anyhow!("unknown codec")), }; Ok(Object { kind: frame.kind, data }) } #[cfg(test)] mod tests { use super::*; #[test] fn put_read_bundle() { let dir = tempfile::tempdir().unwrap(); let store = Store::open(dir.path()).unwrap(); let big = "x".repeat(10_000).into_bytes(); let id = store.put_file(&big).unwrap(); assert_eq!(store.read_file(&id).unwrap(), big); let bundle = store.bundle(&[id]).unwrap(); let other = tempfile::tempdir().unwrap(); let store2 = Store::open(other.path()).unwrap(); assert_eq!(store2.ingest(&bundle).unwrap(), 1); assert_eq!(store2.read_file(&id).unwrap(), big); } }