use std::fs::{self, File};
use std::io::{self, Read, Write};
use std::path::{Path, PathBuf};
use std::process;
use std::sync::Arc;
use std::sync::atomic::{AtomicU64, Ordering};
use tempfile::NamedTempFile;
use crate::batch::{RealSyncer, Syncer};
pub use crate::batch::{SyncPolicy, WriteBatch};
use crate::hash::{self, Hash, object_path, to_hex};
use crate::layout::RepoLayout;
use crate::object::{MkitError, Object, object_id_from_bytes, verified_id_and_object};
use crate::serialize;
mod memory;
mod source;
pub use memory::MemorySource;
pub use source::{DisplaySource, EphemeralSink, ObjectSource};
pub const MKIT_DIR: &str = ".mkit";
pub const OBJECTS_DIR: &str = "objects";
pub const FORMAT_FILE: &str = "format";
pub const FORMAT_VALUE: &str = "bmt-v1";
pub const MAX_RAW_OBJECT_SIZE: usize = 1024 * 1024 * 1024;
pub const MAX_TREE_DEPTH: usize = 128;
#[derive(Debug, thiserror::Error)]
pub enum StoreError {
#[error("path is not an mkit repository (missing .mkit/objects)")]
NotAMkitRepository,
#[error(".mkit already exists in this directory")]
AlreadyInitialized,
#[error(
"repository object-addressing format is {found:?}, expected \"{}\" — this repository predates merkle object addressing and is not readable by this mkit (pre-1.0: no migration). See docs/specs/SPEC-MERKLE-OBJECTS.md.",
FORMAT_VALUE
)]
IncompatibleRepoFormat { found: Option<String> },
#[error("object {0} not found")]
ObjectNotFound(String),
#[error("object exceeds {} byte cap", MAX_RAW_OBJECT_SIZE)]
ObjectTooLarge,
#[error("tree nesting exceeds {} levels", MAX_TREE_DEPTH)]
TreeTooDeep,
#[error("on-disk bytes hash to {actual}, expected {expected}")]
HashMismatch { expected: String, actual: String },
#[error(transparent)]
Io(#[from] io::Error),
#[error(transparent)]
Decode(#[from] MkitError),
#[error("encode_deltas callback returned {actual} results for {expected} candidates")]
DeltaBatchLengthMismatch { expected: usize, actual: usize },
}
#[derive(Debug)]
pub struct BulkWriter<'a> {
store: &'a ObjectStore,
dirs: std::collections::HashSet<PathBuf>,
files: std::collections::HashSet<PathBuf>,
}
impl BulkWriter<'_> {
pub fn write(&mut self, bytes: &[u8]) -> StoreResult<Hash> {
if bytes.len() > MAX_RAW_OBJECT_SIZE {
return Err(StoreError::ObjectTooLarge);
}
let h = object_id_from_bytes(bytes);
let final_path = self.store.path_for(&h);
let shard_dir = final_path
.parent()
.expect("object path always has a 2-hex parent");
if let Ok(existing) = fs::read(&final_path)
&& existing == bytes
&& refresh_mtime(&final_path).is_ok()
{
self.dirs.insert(shard_dir.to_path_buf());
self.files.insert(final_path);
return Ok(h);
}
if !self.dirs.contains(shard_dir) {
fs::create_dir_all(shard_dir)?;
}
crate::atomic::write_content_synced(&final_path, bytes)?;
self.dirs.insert(shard_dir.to_path_buf());
Ok(h)
}
pub fn commit(self) -> StoreResult<()> {
for file in &self.files {
fs::OpenOptions::new().write(true).open(file)?.sync_all()?;
}
for dir in &self.dirs {
crate::atomic::sync_dir(dir)?;
}
Ok(())
}
}
pub type StoreResult<T> = Result<T, StoreError>;
static TEMP_SEQ: AtomicU64 = AtomicU64::new(0);
#[derive(Debug, Clone)]
pub struct ObjectStore {
objects_root: PathBuf,
syncer: Arc<dyn Syncer>,
default_policy: SyncPolicy,
#[cfg(test)]
read_calls: Arc<AtomicU64>,
}
impl ObjectStore {
pub fn open(layout: &RepoLayout) -> StoreResult<Self> {
let objects_root = layout.objects_dir();
if !objects_root.is_dir() {
return Err(StoreError::NotAMkitRepository);
}
match fs::read_to_string(layout.format_file()) {
Ok(s) if s.trim() == FORMAT_VALUE => {}
Ok(s) => {
return Err(StoreError::IncompatibleRepoFormat {
found: Some(s.trim().to_owned()),
});
}
Err(e) if e.kind() == io::ErrorKind::NotFound => {
return Err(StoreError::IncompatibleRepoFormat { found: None });
}
Err(e) => return Err(StoreError::Io(e)),
}
Ok(Self {
objects_root,
syncer: Arc::new(RealSyncer),
default_policy: SyncPolicy::Batch,
#[cfg(test)]
read_calls: Arc::new(AtomicU64::new(0)),
})
}
pub fn init(layout: &RepoLayout) -> StoreResult<Self> {
if layout.common_dir().exists() {
return Err(StoreError::AlreadyInitialized);
}
let objects_root = layout.objects_dir();
fs::create_dir_all(&objects_root)?;
fs::write(layout.format_file(), format!("{FORMAT_VALUE}\n").as_bytes())?;
Ok(Self {
objects_root,
syncer: Arc::new(RealSyncer),
default_policy: SyncPolicy::Batch,
#[cfg(test)]
read_calls: Arc::new(AtomicU64::new(0)),
})
}
pub fn set_sync_policy(&mut self, policy: SyncPolicy) {
self.default_policy = policy;
}
#[cfg(test)]
pub(crate) fn set_syncer(&mut self, syncer: Arc<dyn Syncer>) {
self.syncer = syncer;
}
pub(crate) fn syncer(&self) -> &Arc<dyn Syncer> {
&self.syncer
}
#[cfg(test)]
pub(crate) fn read_call_count(&self) -> u64 {
self.read_calls.load(Ordering::Relaxed)
}
#[must_use]
pub fn batch(&self) -> WriteBatch<'_> {
self.batch_with_policy(self.default_policy)
}
#[must_use]
pub fn batch_with_policy(&self, policy: SyncPolicy) -> WriteBatch<'_> {
WriteBatch::new(self, policy)
}
#[must_use]
pub fn is_repo_root(root: &Path) -> bool {
root.join(MKIT_DIR).join(OBJECTS_DIR).is_dir()
}
#[must_use]
pub fn objects_root(&self) -> &Path {
&self.objects_root
}
pub(crate) fn path_for(&self, h: &Hash) -> PathBuf {
let p = object_path(h);
let dir = std::str::from_utf8(&p.dir).expect("ascii hex");
let file = std::str::from_utf8(&p.file).expect("ascii hex");
self.objects_root.join(dir).join(file)
}
#[must_use]
pub fn contains(&self, h: &Hash) -> bool {
self.path_for(h).is_file()
}
pub fn write(&self, bytes: &[u8]) -> StoreResult<Hash> {
if bytes.len() > MAX_RAW_OBJECT_SIZE {
return Err(StoreError::ObjectTooLarge);
}
let h = object_id_from_bytes(bytes);
let final_path = self.path_for(&h);
if final_path.exists() {
self.syncer()
.dir_sync(final_path.parent().expect("object path has parent"))?;
return Ok(h);
}
let shard_dir = final_path
.parent()
.expect("object path always has a 2-hex parent");
fs::create_dir_all(shard_dir)?;
write_atomic(&final_path, bytes, &**self.syncer())?;
Ok(h)
}
#[must_use]
pub fn bulk_writer(&self) -> BulkWriter<'_> {
BulkWriter {
store: self,
dirs: std::collections::HashSet::new(),
files: std::collections::HashSet::new(),
}
}
fn read_raw(&self, h: &Hash) -> StoreResult<Vec<u8>> {
#[cfg(test)]
self.read_calls.fetch_add(1, Ordering::Relaxed);
let path = self.path_for(h);
let mut file = File::open(&path).map_err(|e| match e.kind() {
io::ErrorKind::NotFound => StoreError::ObjectNotFound(to_hex(h)),
_ => StoreError::Io(e),
})?;
let meta = file.metadata()?;
let size = meta.len();
if u128::from(size) > MAX_RAW_OBJECT_SIZE as u128 {
return Err(StoreError::ObjectTooLarge);
}
let cap = usize::try_from(size).map_err(|_| StoreError::ObjectTooLarge)?;
let mut bytes = Vec::with_capacity(cap);
file.read_to_end(&mut bytes)?;
Ok(bytes)
}
pub(crate) fn read_raw_for_verification(&self, h: &Hash) -> StoreResult<Vec<u8>> {
self.read_raw(h)
}
pub fn read(&self, h: &Hash) -> StoreResult<Vec<u8>> {
let bytes = self.read_raw(h)?;
check_hash(h, &object_id_from_bytes(&bytes))?;
Ok(bytes)
}
pub(crate) fn read_with_allocator<E: From<StoreError>>(
&self,
h: &Hash,
allocate: impl FnOnce(usize) -> Result<Vec<u8>, E>,
) -> Result<Vec<u8>, E> {
#[cfg(test)]
self.read_calls.fetch_add(1, Ordering::Relaxed);
let mut file = File::open(self.path_for(h)).map_err(|e| {
E::from(match e.kind() {
io::ErrorKind::NotFound => StoreError::ObjectNotFound(to_hex(h)),
_ => StoreError::Io(e),
})
})?;
let size = file.metadata().map_err(StoreError::Io)?.len();
if size > MAX_RAW_OBJECT_SIZE as u64 {
return Err(StoreError::ObjectTooLarge.into());
}
let len = usize::try_from(size).map_err(|_| StoreError::ObjectTooLarge)?;
let mut bytes = allocate(len)?;
bytes.resize(len, 0);
file.read_exact(&mut bytes).map_err(StoreError::Io)?;
let mut extra = [0];
if file.read(&mut extra).map_err(StoreError::Io)? != 0 {
return Err(StoreError::ObjectTooLarge.into());
}
check_hash(h, &object_id_from_bytes(&bytes))?;
Ok(bytes)
}
pub fn read_unverified(&self, h: &Hash) -> StoreResult<Vec<u8>> {
self.read_raw(h)
}
pub fn object_type(&self, h: &Hash) -> StoreResult<crate::object::ObjectType> {
let path = self.path_for(h);
let mut file = File::open(&path).map_err(|e| match e.kind() {
io::ErrorKind::NotFound => StoreError::ObjectNotFound(to_hex(h)),
_ => StoreError::Io(e),
})?;
let mut prologue = [0u8; 6];
file.read_exact(&mut prologue)
.map_err(|_| StoreError::Decode(MkitError::EmptyData))?;
if prologue[1..5] != crate::object::MAGIC {
return Err(StoreError::Decode(MkitError::InvalidMagic));
}
if prologue[5] != crate::object::SCHEMA_VERSION {
return Err(StoreError::Decode(MkitError::UnsupportedObjectVersion));
}
crate::object::ObjectType::from_u8(prologue[0]).map_err(StoreError::Decode)
}
pub fn read_object(&self, h: &Hash) -> StoreResult<Object> {
let bytes = self.read_raw(h)?;
let (actual, decoded) = verified_id_and_object(&bytes);
check_hash(h, &actual)?;
match decoded {
Some(obj) => Ok(obj),
None => Ok(serialize::deserialize(&bytes)?),
}
}
pub fn verify_object_type(&self, h: &Hash) -> StoreResult<crate::object::ObjectType> {
let bytes = self.read(h)?; if bytes.len() < 6 {
return Err(StoreError::Decode(MkitError::EmptyData));
}
if bytes[1..5] != crate::object::MAGIC {
return Err(StoreError::Decode(MkitError::InvalidMagic));
}
if bytes[5] != crate::object::SCHEMA_VERSION {
return Err(StoreError::Decode(MkitError::UnsupportedObjectVersion));
}
crate::object::ObjectType::from_u8(bytes[0]).map_err(StoreError::Decode)
}
pub fn iter_object_hashes(&self) -> StoreResult<Vec<Hash>> {
let mut out = Vec::new();
let shards = match fs::read_dir(&self.objects_root) {
Ok(rd) => rd,
Err(e) if e.kind() == io::ErrorKind::NotFound => return Ok(out),
Err(e) => return Err(StoreError::Io(e)),
};
for shard in shards {
let shard = shard?;
if !shard.file_type()?.is_dir() {
continue;
}
let dir_name = shard.file_name();
let Some(dir) = dir_name.to_str() else {
continue;
};
if dir.len() != 2 || !dir.bytes().all(|b| b.is_ascii_hexdigit()) {
continue;
}
for entry in fs::read_dir(shard.path())? {
let entry = entry?;
if !entry.file_type()?.is_file() {
continue;
}
let file_name = entry.file_name();
let Some(file) = file_name.to_str() else {
continue;
};
if file.len() != 62 {
continue;
}
if let Ok(h) = hash::from_hex(&format!("{dir}{file}")) {
out.push(h);
}
}
}
Ok(out)
}
pub fn object_metadata(&self, h: &Hash) -> StoreResult<fs::Metadata> {
fs::metadata(self.path_for(h)).map_err(|e| match e.kind() {
io::ErrorKind::NotFound => StoreError::ObjectNotFound(to_hex(h)),
_ => StoreError::Io(e),
})
}
pub fn remove_object(&self, h: &Hash) -> StoreResult<()> {
match fs::remove_file(self.path_for(h)) {
Ok(()) => Ok(()),
Err(e) if e.kind() == io::ErrorKind::NotFound => Ok(()),
Err(e) => Err(StoreError::Io(e)),
}
}
}
fn check_hash(expected: &Hash, actual: &Hash) -> StoreResult<()> {
if actual != expected {
return Err(StoreError::HashMismatch {
expected: to_hex(expected),
actual: to_hex(actual),
});
}
Ok(())
}
pub(crate) fn temp_file_in(parent: &Path, file_name: &str) -> io::Result<NamedTempFile> {
let pid = process::id();
let seq = TEMP_SEQ.fetch_add(1, Ordering::Relaxed);
let tmp_name = format!(".{file_name}.tmp.{pid}.{seq}");
NamedTempFile::with_prefix_in(tmp_name, parent)
}
fn write_atomic(final_path: &Path, bytes: &[u8], syncer: &dyn Syncer) -> io::Result<()> {
let parent = final_path.parent().expect("write_atomic: path has parent");
let file_name = final_path
.file_name()
.expect("write_atomic: path has file name")
.to_string_lossy();
let mut tmp = temp_file_in(parent, &file_name)?;
tmp.as_file_mut().write_all(bytes)?;
syncer.full(tmp.as_file(), tmp.path())?;
syncer.rename(tmp.into_temp_path(), final_path)?;
syncer.dir_sync(parent)?;
Ok(())
}
pub trait ObjectSink {
fn put(&self, bytes: &[u8]) -> StoreResult<Hash>;
fn put_parts(&self, parts: &[&[u8]]) -> StoreResult<Hash>;
fn has(&self, h: &Hash) -> bool;
}
impl ObjectSink for ObjectStore {
fn put(&self, bytes: &[u8]) -> StoreResult<Hash> {
self.write(bytes)
}
fn put_parts(&self, parts: &[&[u8]]) -> StoreResult<Hash> {
let total: usize = parts.iter().map(|p| p.len()).sum();
let mut bytes = Vec::with_capacity(total);
for p in parts {
bytes.extend_from_slice(p);
}
self.write(&bytes)
}
fn has(&self, h: &Hash) -> bool {
self.contains(h)
}
}
pub(crate) fn refresh_mtime(path: &Path) -> io::Result<()> {
#[cfg(test)]
if FAIL_REFRESH.get() {
return Err(io::ErrorKind::NotFound.into());
}
let mut opts = fs::OpenOptions::new();
if cfg!(unix) {
opts.read(true);
} else {
opts.write(true);
}
opts.open(path)?.set_modified(std::time::SystemTime::now())
}
#[cfg(test)]
thread_local! {
static FAIL_REFRESH: std::cell::Cell<bool> = const { std::cell::Cell::new(false) };
}
#[cfg(unix)]
pub(crate) fn sync_parent_dir(parent: &Path) -> io::Result<()> {
match File::open(parent) {
Ok(dir) => dir.sync_all(),
Err(e) if e.kind() == io::ErrorKind::NotFound => Ok(()),
Err(e) => Err(e),
}
}
#[cfg(not(unix))]
#[allow(clippy::unnecessary_wraps)]
pub(crate) fn sync_parent_dir(_parent: &Path) -> io::Result<()> {
Ok(())
}
#[cfg(test)]
mod tests {
use super::*;
use crate::object::Blob;
use std::fs::OpenOptions;
use std::io::Seek;
use tempfile::TempDir;
fn fresh_store() -> (TempDir, ObjectStore) {
let dir = TempDir::new().expect("tempdir");
let store = ObjectStore::init(&RepoLayout::single(dir.path())).expect("init");
(dir, store)
}
#[test]
fn init_creates_layout() {
let dir = TempDir::new().unwrap();
let _ = ObjectStore::init(&RepoLayout::single(dir.path())).unwrap();
assert!(dir.path().join(MKIT_DIR).is_dir());
assert!(dir.path().join(MKIT_DIR).join(OBJECTS_DIR).is_dir());
}
#[test]
fn init_rejects_already_initialized() {
let dir = TempDir::new().unwrap();
let layout = RepoLayout::single(dir.path());
ObjectStore::init(&layout).unwrap();
let err = ObjectStore::init(&layout).unwrap_err();
assert!(matches!(err, StoreError::AlreadyInitialized));
}
#[test]
fn open_rejects_non_repo() {
let dir = TempDir::new().unwrap();
let err = ObjectStore::open(&RepoLayout::single(dir.path())).unwrap_err();
assert!(matches!(err, StoreError::NotAMkitRepository));
}
#[test]
fn init_writes_format_marker_and_open_accepts_it() {
let dir = TempDir::new().unwrap();
let layout = RepoLayout::single(dir.path());
ObjectStore::init(&layout).unwrap();
let marker = fs::read_to_string(dir.path().join(MKIT_DIR).join(FORMAT_FILE)).unwrap();
assert_eq!(marker.trim(), FORMAT_VALUE);
ObjectStore::open(&layout).unwrap();
}
#[test]
fn open_rejects_repo_missing_or_wrong_format_marker() {
let dir = TempDir::new().unwrap();
let layout = RepoLayout::single(dir.path());
ObjectStore::init(&layout).unwrap();
let marker_path = dir.path().join(MKIT_DIR).join(FORMAT_FILE);
fs::remove_file(&marker_path).unwrap();
assert!(matches!(
ObjectStore::open(&layout),
Err(StoreError::IncompatibleRepoFormat { found: None })
));
fs::write(&marker_path, b"flat-v0\n").unwrap();
assert!(matches!(
ObjectStore::open(&layout),
Err(StoreError::IncompatibleRepoFormat { found: Some(_) })
));
}
#[test]
fn is_repo_root_predicate() {
let dir = TempDir::new().unwrap();
assert!(!ObjectStore::is_repo_root(dir.path()));
ObjectStore::init(&RepoLayout::single(dir.path())).unwrap();
assert!(ObjectStore::is_repo_root(dir.path()));
}
#[test]
fn write_then_read_roundtrip() {
let (_dir, store) = fresh_store();
let bytes = b"hello world".to_vec();
let h = store.write(&bytes).unwrap();
assert!(store.contains(&h));
let got = store.read(&h).unwrap();
assert_eq!(got, bytes);
}
#[test]
fn read_object_deserialises() {
let (_dir, store) = fresh_store();
let obj = Object::Blob(Blob {
data: b"object bytes".to_vec(),
});
let bytes = serialize::serialize(&obj).unwrap();
let h = store.write(&bytes).unwrap();
let parsed = store.read_object(&h).unwrap();
assert_eq!(parsed, obj);
}
#[test]
fn read_object_deserialises_tree() {
use crate::object::{EntryMode, Tree, TreeEntry};
let (_dir, store) = fresh_store();
let obj = Object::Tree(Tree {
entries: vec![
TreeEntry {
name: b"a.txt".to_vec(),
mode: EntryMode::Blob,
object_hash: crate::hash::hash(b"a"),
},
TreeEntry {
name: b"b.txt".to_vec(),
mode: EntryMode::Blob,
object_hash: crate::hash::hash(b"b"),
},
],
});
let bytes = serialize::serialize(&obj).unwrap();
let h = store.write(&bytes).unwrap();
let parsed = store.read_object(&h).unwrap();
assert_eq!(parsed, obj);
let via_trait: &dyn ObjectSource = &store;
assert_eq!(via_trait.read_object(&h).unwrap(), obj);
}
#[test]
fn read_object_detects_tree_corruption_via_store_and_trait() {
use crate::object::{EntryMode, Tree, TreeEntry};
let (_dir, store) = fresh_store();
let obj = Object::Tree(Tree {
entries: vec![TreeEntry {
name: b"a.txt".to_vec(),
mode: EntryMode::Blob,
object_hash: crate::hash::hash(b"a"),
}],
});
let bytes = serialize::serialize(&obj).unwrap();
let h = store.write(&bytes).unwrap();
let path = store.path_for(&h);
let mut f = OpenOptions::new()
.read(true)
.write(true)
.open(&path)
.unwrap();
f.seek(io::SeekFrom::Start(0)).unwrap();
f.write_all(&[bytes[0] ^ 0xFF]).unwrap();
f.sync_all().unwrap();
assert!(matches!(
store.read_object(&h).unwrap_err(),
StoreError::HashMismatch { .. }
));
let via_trait: &dyn ObjectSource = &store;
assert!(matches!(
via_trait.read_object(&h).unwrap_err(),
StoreError::HashMismatch { .. }
));
}
#[test]
fn write_is_idempotent() {
let (_dir, store) = fresh_store();
let bytes = b"duplicate".to_vec();
let h1 = store.write(&bytes).unwrap();
let h2 = store.write(&bytes).unwrap();
assert_eq!(h1, h2);
let shard = store.path_for(&h1);
let parent = shard.parent().unwrap();
let entries: Vec<_> = fs::read_dir(parent).unwrap().collect();
assert_eq!(
entries.len(),
1,
"shard dir should contain exactly the final object, no temp leaks"
);
}
#[test]
fn read_missing_returns_not_found() {
let (_dir, store) = fresh_store();
let phony = hash::hash(b"never written");
let err = store.read(&phony).unwrap_err();
assert!(matches!(err, StoreError::ObjectNotFound(_)));
assert!(!store.contains(&phony));
}
#[test]
fn write_rejects_oversize() {
let (_dir, store) = fresh_store();
let oversize = vec![0u8; MAX_RAW_OBJECT_SIZE + 1];
let err = store.write(&oversize).unwrap_err();
assert!(matches!(err, StoreError::ObjectTooLarge), "got {err:?}");
drop(oversize);
let h = store.write(&[0u8; 16]).unwrap();
assert!(store.contains(&h));
}
#[test]
fn read_rejects_oversize_on_disk() {
let (_dir, store) = fresh_store();
let h = store.write(b"seed").unwrap();
let path = store.path_for(&h);
let f = OpenOptions::new().write(true).open(&path).unwrap();
f.set_len(MAX_RAW_OBJECT_SIZE as u64 + 1).unwrap();
drop(f);
let err = store.read(&h).unwrap_err();
assert!(matches!(err, StoreError::ObjectTooLarge));
}
#[test]
fn read_detects_corruption() {
let (_dir, store) = fresh_store();
let bytes = b"trustworthy".to_vec();
let h = store.write(&bytes).unwrap();
let path = store.path_for(&h);
{
let mut f = OpenOptions::new()
.read(true)
.write(true)
.open(&path)
.unwrap();
f.seek(io::SeekFrom::Start(0)).unwrap();
f.write_all(&[bytes[0] ^ 0xFF]).unwrap();
f.sync_all().unwrap();
}
let err = store.read(&h).unwrap_err();
match err {
StoreError::HashMismatch { expected, actual } => {
assert_eq!(expected, to_hex(&h));
assert_ne!(actual, expected, "actual must differ once corrupted");
}
other => panic!("expected HashMismatch, got {other:?}"),
}
}
fn corrupt_first_byte(store: &ObjectStore, h: &Hash, first_byte: u8) {
let path = store.path_for(h);
let mut f = OpenOptions::new()
.read(true)
.write(true)
.open(&path)
.unwrap();
f.seek(io::SeekFrom::Start(0)).unwrap();
f.write_all(&[first_byte ^ 0xFF]).unwrap();
f.sync_all().unwrap();
}
#[test]
fn read_unverified_returns_bytes_read_rejects_corruption() {
let (_dir, store) = fresh_store();
let bytes = b"trustworthy".to_vec();
let h = store.write(&bytes).unwrap();
corrupt_first_byte(&store, &h, bytes[0]);
assert!(matches!(
store.read(&h).unwrap_err(),
StoreError::HashMismatch { .. }
));
let mut corrupted = bytes.clone();
corrupted[0] ^= 0xFF;
assert_eq!(store.read_unverified(&h).unwrap(), corrupted);
}
#[test]
fn path_layout_is_2_then_62_hex() {
let (_dir, store) = fresh_store();
let bytes = b"layout test".to_vec();
let h = store.write(&bytes).unwrap();
let hex = to_hex(&h);
let path = store.path_for(&h);
let parent = path.parent().unwrap();
let parent_name = parent.file_name().unwrap().to_str().unwrap();
let file_name = path.file_name().unwrap().to_str().unwrap();
assert_eq!(parent_name.len(), 2);
assert_eq!(file_name.len(), 62);
assert_eq!(parent_name, &hex[..2]);
assert_eq!(file_name, &hex[2..]);
assert!(path.is_file(), "object file must exist at expected path");
}
#[test]
fn temp_file_left_behind_does_not_satisfy_contains() {
let (_dir, store) = fresh_store();
let target = hash::hash(b"never finalised");
let final_path = store.path_for(&target);
let shard = final_path.parent().unwrap();
fs::create_dir_all(shard).unwrap();
let stale = shard.join(format!(
".{}.tmp.0.0",
final_path.file_name().unwrap().to_string_lossy()
));
fs::write(&stale, b"partial").unwrap();
assert!(stale.is_file());
assert!(!store.contains(&target));
for entry in fs::read_dir(shard).unwrap() {
let p = entry.unwrap().path();
let bytes = fs::read(&p).unwrap();
assert_ne!(
hash::hash(&bytes),
target,
"stale temp file must not satisfy the target hash"
);
}
}
}
#[cfg(test)]
mod bulk_writer_tests {
use super::*;
#[test]
fn bulk_writer_round_trips_and_rewrites() {
let td = tempfile::tempdir().unwrap();
let store = ObjectStore::init(&RepoLayout::single(td.path())).unwrap();
let obj = crate::serialize::serialize(&crate::object::Object::Blob(crate::object::Blob {
data: b"bulk".to_vec(),
}))
.unwrap();
let mut bw = store.bulk_writer();
let h1 = bw.write(&obj).unwrap();
let h2 = bw.write(&obj).unwrap();
assert_eq!(h1, h2);
bw.commit().unwrap();
assert_eq!(store.read(&h1).unwrap(), obj);
assert_eq!(store.write(&obj).unwrap(), h1);
}
}
#[cfg(test)]
mod dedup_mtime_tests {
use super::*;
use crate::layout::RepoLayout;
use std::time::{Duration, SystemTime, UNIX_EPOCH};
const DAY: u64 = 24 * 60 * 60;
const GRACE: u64 = 14 * DAY;
fn repo() -> (tempfile::TempDir, ObjectStore) {
let td = tempfile::tempdir().unwrap();
let store = ObjectStore::init(&RepoLayout::single(td.path())).unwrap();
crate::refs::init(&RepoLayout::single(td.path())).unwrap();
(td, store)
}
fn mtime(store: &ObjectStore, h: &Hash) -> SystemTime {
fs::metadata(store.path_for(h)).unwrap().modified().unwrap()
}
fn backdate(store: &ObjectStore, h: &Hash) -> SystemTime {
let old = SystemTime::now() - Duration::from_secs(30 * DAY);
File::open(store.path_for(h))
.unwrap()
.set_modified(old)
.unwrap();
assert!(mtime(store, h) <= old + Duration::from_secs(1));
old
}
fn now_secs() -> u64 {
SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap()
.as_secs()
}
fn assert_dedup_refreshes(rewrite: impl Fn(&ObjectStore, &[u8]) -> Hash) {
let (td, store) = repo();
let bytes = b"object the import is about to publish".to_vec();
let h = store.write(&bytes).unwrap();
let control = store.write(b"old orphan nobody writes again").unwrap();
backdate(&store, &h);
backdate(&store, &control);
let before = SystemTime::now() - Duration::from_secs(1);
assert_eq!(rewrite(&store, &bytes), h);
assert!(
mtime(&store, &h) >= before,
"dedup hit must refresh the object's mtime"
);
let report = crate::ops::gc::run_gc(
&store,
&RepoLayout::single(td.path()),
now_secs(),
GRACE,
false,
)
.unwrap();
assert!(store.contains(&h), "re-written object pruned: {report:?}");
assert_eq!(store.read(&h).unwrap(), bytes);
assert!(!store.contains(&control), "control orphan kept: {report:?}");
}
#[test]
fn failed_refresh_falls_back_to_rewrite() {
fn failing<T>(f: impl FnOnce() -> T) -> T {
FAIL_REFRESH.set(true);
let out = f();
FAIL_REFRESH.set(false);
out
}
assert_dedup_refreshes(|s, b| {
let mut bw = s.bulk_writer();
let h = failing(|| bw.write(b).unwrap());
bw.commit().unwrap();
h
});
}
#[test]
fn bulk_writer_dedup_refreshes_mtime() {
assert_dedup_refreshes(|s, b| {
let mut bw = s.bulk_writer();
let h = bw.write(b).unwrap();
bw.commit().unwrap();
h
});
}
}