//! Oxygen Strata (docs/stores.md): the storage a repository can be kept //! on, as a trait, so that a bucket, a directory or a hub's disk all look //! the same to the code that pushes to and fetches from them. //! //! A store has two halves. **Objects** are written once and never //! change: packs and their indexes, named by content, which any store //! can hold safely. **References** are the one mutable record (bookmarks, //! tags, the policy), updated only by compare-and-swap on a version the //! store hands out: an ETag on S3, a generation on GCS, a content hash on //! a directory. Two writers can therefore share a store with no lock //! server: the loser of a race reads again, merges, and tries again. //! //! A repository's own objects (`.oxy/objects`, see [`crate::store`]) are //! kept on a store too, a directory one, with the same keys a bucket //! would hold; a remote is a store as well. This module holds the trait, //! a directory store, an in-memory store for tests, and the conformance //! check every store must pass. Cloud stores live in a crate of their //! own, because of what they pull in. use std::collections::BTreeMap; use std::fs; use std::io::{self, Read, Seek, SeekFrom, Write}; use std::ops::Range; use std::path::{Path, PathBuf}; use std::sync::Mutex; use std::time::{Duration, Instant}; /// The version of a reference record as the store reports it: opaque, /// compared only for equality, and only ever valid for the store that /// issued it. Writing the same bytes again may keep the version (S3's /// ETag and a directory's content hash are both of the content): a /// compare-and-swap guards against losing a *different* record, which is /// all it needs to. #[derive(Clone, Debug, PartialEq, Eq)] pub struct Version(pub String); /// One key in a listing. #[derive(Clone, Debug, PartialEq, Eq)] pub struct Entry { pub key: String, pub size: u64, /// When the object was written, in wall milliseconds, where the store /// knows (garbage collection's grace period reads it). pub modified_ms: Option, } #[derive(Debug, thiserror::Error)] pub enum Error { #[error("{0}: not in the store")] NotFound(String), #[error("{0}: already in the store")] Exists(String), #[error("{0}: changed since it was read")] Conflict(String), #[error("bad store key {0:?}: {1}")] Key(String, &'static str), #[error("{0}: busy (another writer holds it)")] Busy(String), #[error(transparent)] Io(#[from] io::Error), #[error("{0}")] Backend(String), } pub type Result = std::result::Result; /// Longest key a store accepts; S3 allows 1024 bytes. pub const MAX_KEY_LEN: usize = 1024; /// Refuse a key that is not a plain relative path: `/`-separated /// segments of `[A-Za-z0-9._-]`, none empty, none `.` or `..`. The same /// rule on every store, so a key written to one can be read from another, /// and so a key can never escape a directory store's root. pub fn check_key(key: &str) -> Result<()> { let bad = |why| Error::Key(key.to_string(), why); if key.is_empty() { return Err(bad("empty")); } if key.len() > MAX_KEY_LEN { return Err(bad("longer than 1024 bytes")); } for seg in key.split('/') { if seg.is_empty() { return Err(bad( "an empty segment (a leading, trailing or doubled slash)", )); } if seg == "." || seg == ".." { return Err(bad("a . or .. segment")); } if !seg .bytes() .all(|b| b.is_ascii_alphanumeric() || b == b'.' || b == b'_' || b == b'-') { return Err(bad("a character outside A-Z a-z 0-9 . _ -")); } } Ok(()) } /// Where a repository's packs and references are kept. Every method is /// synchronous and may be called from several threads at once. pub trait Store: Send + Sync { /// A name for messages: the URL, or the directory. fn describe(&self) -> String; // ---- objects: written once, never changed fn has(&self, key: &str) -> Result; /// The whole object; `NotFound` when there is none. fn get(&self, key: &str) -> Result>; /// Bytes `range` of an object; an error when the range runs past its end. fn get_range(&self, key: &str, range: Range) -> Result>; /// Write an object that does not exist yet; `Exists` when it does, and /// then nothing is changed. Readers never see a partial object. fn put_new(&self, key: &str, bytes: &[u8]) -> Result<()>; /// Every key under `prefix` (a key prefix, not necessarily a whole /// segment), sorted, with sizes. fn list(&self, prefix: &str) -> Result>; /// Remove an object; fine when it is already gone. fn delete(&self, key: &str) -> Result<()>; /// One object's listing entry without listing its neighbours; `None` /// when there is none. fn stat(&self, key: &str) -> Result> { Ok(self.list(key)?.into_iter().find(|e| e.key == key)) } /// Write several objects that do not exist yet; one that does is left /// as it is, which is not an error (objects are named by content). A /// store that can batch its syncs does; this does them one at a time. fn put_many_new(&self, objects: &[(String, Vec)]) -> Result<()> { for (key, bytes) in objects { match self.put_new(key, bytes) { Ok(()) | Err(Error::Exists(_)) => {} Err(e) => return Err(e), } } Ok(()) } /// `put_new` of a file's contents: a store on the same disk can move /// the file into place instead of copying it. fn put_new_file(&self, key: &str, path: &Path) -> Result<()> { self.put_new(key, &fs::read(path)?) } // ---- references: one small record, updated by compare-and-swap /// The record and its current version, or `None` when it was never /// written. fn read_ref(&self, key: &str) -> Result, Version)>>; /// Replace the record only if its version is still `expected` /// (`None`: only if it does not exist yet). `Conflict` otherwise, and /// then nothing is changed. Returns the new version. fn write_ref(&self, key: &str, bytes: &[u8], expected: Option<&Version>) -> Result; } /// A reference record's version on stores that have no native one: the /// hash of its bytes, which two different records never share. fn content_version(bytes: &[u8]) -> Version { Version(blake3::hash(bytes).to_hex().to_string()) } fn check_expected(key: &str, current: Option<&Version>, expected: Option<&Version>) -> Result<()> { if current == expected { Ok(()) } else { Err(Error::Conflict(key.to_string())) } } // ---------------------------------------------------------------- directory /// A store in a directory: each key a file under the root. Create-only /// writes go through a hard link, which the file system refuses when the /// name is taken, and a reference is updated under a lock file beside it. /// Several processes on one machine can share it; it is also what the /// tests and `oxy remote check` use. pub struct LocalStore { root: PathBuf, } /// How long a reference's lock may be held before it counts as abandoned /// by a process that died, and how long a writer waits for it. const LOCK_STALE: Duration = Duration::from_secs(60); const LOCK_WAIT: Duration = Duration::from_secs(10); impl LocalStore { /// Open the store at `root`, creating the directory if needed. pub fn open(root: impl Into) -> Result { let root = root.into(); fs::create_dir_all(&root)?; Ok(LocalStore { root }) } pub fn root(&self) -> &Path { &self.root } fn path(&self, key: &str) -> Result { check_key(key)?; Ok(self.root.join(key)) } fn temp_beside(&self, path: &Path) -> PathBuf { let mut nonce = [0u8; 6]; rand::RngCore::fill_bytes(&mut rand::rngs::OsRng, &mut nonce); let name = path .file_name() .map(|n| n.to_string_lossy().into_owned()) .unwrap_or_default(); path.with_file_name(format!(".{name}.tmp-{}", hex::encode(nonce))) } fn list_dir(&self, dir: &Path, rel: &str, prefix: &str, out: &mut Vec) -> Result<()> { let entries = match fs::read_dir(dir) { Ok(e) => e, Err(e) if e.kind() == io::ErrorKind::NotFound => return Ok(()), Err(e) => return Err(e.into()), }; for entry in entries { let entry = entry?; let name = entry.file_name().to_string_lossy().into_owned(); // Temporary and lock files are not objects. if name.starts_with('.') { continue; } let key = if rel.is_empty() { name.clone() } else { format!("{rel}/{name}") }; let ty = entry.file_type()?; if ty.is_dir() { // Descend only where the prefix could still match. let dir_prefix = format!("{key}/"); if prefix.starts_with(&dir_prefix) || dir_prefix.starts_with(prefix) { self.list_dir(&entry.path(), &key, prefix, out)?; } } else if ty.is_file() && key.starts_with(prefix) { let meta = entry.metadata()?; out.push(Entry { key, size: meta.len(), modified_ms: modified_ms(&meta), }); } } Ok(()) } /// Take `path.lock`, waiting for another writer, and removing a lock /// a dead process left behind. fn lock(&self, key: &str, path: &Path) -> Result { let lock = path.with_file_name(format!( ".{}.lock", path.file_name() .map(|n| n.to_string_lossy().into_owned()) .unwrap_or_default() )); let started = Instant::now(); loop { match fs::OpenOptions::new() .write(true) .create_new(true) .open(&lock) { Ok(mut f) => { let _ = writeln!(f, "{}", std::process::id()); return Ok(lock); } Err(e) if e.kind() == io::ErrorKind::AlreadyExists => { if let Ok(m) = fs::metadata(&lock) { let age = m.modified().ok().and_then(|t| t.elapsed().ok()); if age.is_some_and(|a| a > LOCK_STALE) { let _ = fs::remove_file(&lock); continue; } } if started.elapsed() > LOCK_WAIT { return Err(Error::Busy(key.to_string())); } std::thread::sleep(Duration::from_millis(10)); } Err(e) => return Err(e.into()), } } } } /// A file's modification time as wall milliseconds. fn modified_ms(meta: &fs::Metadata) -> Option { meta.modified() .ok() .and_then(|t| t.duration_since(std::time::UNIX_EPOCH).ok()) .map(|d| d.as_millis() as u64) } impl LocalStore { /// Link `tmp` into `path`, create-only; the directory is synced and /// `tmp` removed either way. `Exists` when the name is taken. fn link_into_place(&self, key: &str, tmp: &Path, path: &Path) -> Result<()> { let linked = fs::hard_link(tmp, path); let _ = fs::remove_file(tmp); match linked { Ok(()) => { if let Some(dir) = path.parent() { crate::durable::sync_dir(dir)?; } Ok(()) } Err(e) if e.kind() == io::ErrorKind::AlreadyExists => { Err(Error::Exists(key.to_string())) } Err(e) => Err(e.into()), } } /// Create `path`'s directory, syncing the first directory that was /// made so its entry is durable too. fn ensure_dir(&self, path: &Path) -> Result<()> { let Some(dir) = path.parent() else { return Ok(()); }; if dir.is_dir() { return Ok(()); } let mut made = dir.to_path_buf(); while let Some(parent) = made.parent() { if parent.is_dir() || parent == self.root { break; } made = parent.to_path_buf(); } fs::create_dir_all(dir)?; if let Some(parent) = made.parent() { crate::durable::sync_dir(parent)?; } Ok(()) } } impl Store for LocalStore { fn describe(&self) -> String { self.root.display().to_string() } fn stat(&self, key: &str) -> Result> { match fs::metadata(self.path(key)?) { Ok(m) if m.is_file() => Ok(Some(Entry { key: key.to_string(), size: m.len(), modified_ms: modified_ms(&m), })), Ok(_) => Ok(None), Err(e) if e.kind() == io::ErrorKind::NotFound => Ok(None), Err(e) => Err(e.into()), } } /// Every file written and synced (on every core), then each linked /// into place and each directory synced once: durable when this /// returns, as one at a time would be, with far fewer waits. fn put_many_new(&self, objects: &[(String, Vec)]) -> Result<()> { use rayon::prelude::*; let mut todo: Vec<(PathBuf, PathBuf, &[u8])> = Vec::new(); for (key, bytes) in objects { let path = self.path(key)?; if path.exists() || todo.iter().any(|(p, ..)| *p == path) { continue; } self.ensure_dir(&path)?; todo.push((path.clone(), self.temp_beside(&path), bytes)); } if todo.is_empty() { return Ok(()); } crate::durable::step("write loose objects")?; todo.par_iter() .try_for_each(|(_, tmp, bytes)| -> io::Result<()> { let mut f = fs::File::create(tmp)?; f.write_all(bytes)?; f.sync_all() })?; crate::durable::step("rename loose objects")?; let mut dirs: std::collections::BTreeSet = std::collections::BTreeSet::new(); for (path, tmp, _) in &todo { let linked = fs::hard_link(tmp, path); let _ = fs::remove_file(tmp); match linked { Ok(()) => { if let Some(d) = path.parent() { dirs.insert(d.to_path_buf()); } } Err(e) if e.kind() == io::ErrorKind::AlreadyExists => {} Err(e) => return Err(e.into()), } } dirs.par_iter() .try_for_each(|d| crate::durable::sync_dir(d))?; Ok(()) } /// The file is linked into place when it is on this file system, /// else copied; it is removed either way. fn put_new_file(&self, key: &str, file: &Path) -> Result<()> { let path = self.path(key)?; if path.exists() { let _ = fs::remove_file(file); return Err(Error::Exists(key.to_string())); } self.ensure_dir(&path)?; match fs::hard_link(file, &path) { Ok(()) => { let _ = fs::remove_file(file); if let Some(dir) = path.parent() { crate::durable::sync_dir(dir)?; } Ok(()) } Err(e) if e.kind() == io::ErrorKind::AlreadyExists => { let _ = fs::remove_file(file); Err(Error::Exists(key.to_string())) } Err(_) => { // Another file system: copy beside the name, then link. let tmp = self.temp_beside(&path); let copied = (|| { let mut src = fs::File::open(file)?; let mut dst = fs::File::create(&tmp)?; io::copy(&mut src, &mut dst)?; dst.sync_all() })(); let _ = fs::remove_file(file); copied?; self.link_into_place(key, &tmp, &path) } } } fn has(&self, key: &str) -> Result { Ok(self.path(key)?.is_file()) } fn get(&self, key: &str) -> Result> { let path = self.path(key)?; match fs::read(&path) { Ok(b) => Ok(b), Err(e) if e.kind() == io::ErrorKind::NotFound => Err(Error::NotFound(key.to_string())), Err(e) => Err(e.into()), } } fn get_range(&self, key: &str, range: Range) -> Result> { let path = self.path(key)?; let mut f = match fs::File::open(&path) { Ok(f) => f, Err(e) if e.kind() == io::ErrorKind::NotFound => { return Err(Error::NotFound(key.to_string())) } Err(e) => return Err(e.into()), }; let len = f.metadata()?.len(); if range.end > len || range.start > range.end { return Err(Error::Backend(format!( "{key}: bytes {}..{} asked of an object of {len} bytes", range.start, range.end ))); } f.seek(SeekFrom::Start(range.start))?; let mut out = vec![0u8; (range.end - range.start) as usize]; f.read_exact(&mut out)?; Ok(out) } fn put_new(&self, key: &str, bytes: &[u8]) -> Result<()> { let path = self.path(key)?; if path.exists() { return Err(Error::Exists(key.to_string())); } self.ensure_dir(&path)?; // Written beside its name, then linked into it: the link fails if // the name was taken meanwhile, and a reader never sees a part. let tmp = self.temp_beside(&path); let written = (|| { let mut f = fs::File::create(&tmp)?; f.write_all(bytes)?; f.sync_all() })(); if let Err(e) = written { let _ = fs::remove_file(&tmp); return Err(e.into()); } self.link_into_place(key, &tmp, &path) } fn list(&self, prefix: &str) -> Result> { let mut out = Vec::new(); self.list_dir(&self.root, "", prefix, &mut out)?; out.sort_by(|a, b| a.key.cmp(&b.key)); Ok(out) } fn delete(&self, key: &str) -> Result<()> { let path = self.path(key)?; match fs::remove_file(&path) { Ok(()) => {} Err(e) if e.kind() == io::ErrorKind::NotFound => return Ok(()), Err(e) => return Err(e.into()), } // A directory left empty goes too, up to the root, so the store // holds nothing but its objects. let mut dir = path.parent(); while let Some(d) = dir { if d == self.root || fs::remove_dir(d).is_err() { break; } dir = d.parent(); } Ok(()) } fn read_ref(&self, key: &str) -> Result, Version)>> { let path = self.path(key)?; match fs::read(&path) { Ok(b) => { let v = content_version(&b); Ok(Some((b, v))) } Err(e) if e.kind() == io::ErrorKind::NotFound => Ok(None), Err(e) => Err(e.into()), } } fn write_ref(&self, key: &str, bytes: &[u8], expected: Option<&Version>) -> Result { let path = self.path(key)?; if let Some(dir) = path.parent() { fs::create_dir_all(dir)?; } let lock = self.lock(key, &path)?; let result = (|| { let current = self.read_ref(key)?.map(|(_, v)| v); check_expected(key, current.as_ref(), expected)?; crate::durable::write_atomic(&path, bytes)?; Ok(content_version(bytes)) })(); let _ = fs::remove_file(&lock); result } } // ---------------------------------------------------------------- memory /// A store in memory, for tests: the conditional writes behave as the /// cloud stores' do. #[derive(Default)] pub struct MemoryStore { objects: Mutex, u64)>>, } fn now_ms() -> u64 { std::time::SystemTime::now() .duration_since(std::time::UNIX_EPOCH) .map(|d| d.as_millis() as u64) .unwrap_or(0) } impl MemoryStore { pub fn new() -> MemoryStore { MemoryStore::default() } fn map(&self) -> std::sync::MutexGuard<'_, BTreeMap, u64)>> { self.objects.lock().unwrap_or_else(|p| p.into_inner()) } /// Pretend `key` was written `age` ago (for grace-period tests). pub fn age(&self, key: &str, age: Duration) { if let Some(entry) = self.map().get_mut(key) { entry.1 = now_ms().saturating_sub(age.as_millis() as u64); } } } impl Store for MemoryStore { fn describe(&self) -> String { "memory".into() } fn has(&self, key: &str) -> Result { check_key(key)?; Ok(self.map().contains_key(key)) } fn get(&self, key: &str) -> Result> { check_key(key)?; self.map() .get(key) .map(|(b, _)| b.clone()) .ok_or_else(|| Error::NotFound(key.to_string())) } fn get_range(&self, key: &str, range: Range) -> Result> { let all = self.get(key)?; let len = all.len() as u64; if range.end > len || range.start > range.end { return Err(Error::Backend(format!( "{key}: bytes {}..{} asked of an object of {len} bytes", range.start, range.end ))); } Ok(all[range.start as usize..range.end as usize].to_vec()) } fn put_new(&self, key: &str, bytes: &[u8]) -> Result<()> { check_key(key)?; let mut map = self.map(); if map.contains_key(key) { return Err(Error::Exists(key.to_string())); } map.insert(key.to_string(), (bytes.to_vec(), now_ms())); Ok(()) } fn list(&self, prefix: &str) -> Result> { Ok(self .map() .range(prefix.to_string()..) .take_while(|(k, _)| k.starts_with(prefix)) .map(|(k, (v, t))| Entry { key: k.clone(), size: v.len() as u64, modified_ms: Some(*t), }) .collect()) } fn delete(&self, key: &str) -> Result<()> { check_key(key)?; self.map().remove(key); Ok(()) } fn read_ref(&self, key: &str) -> Result, Version)>> { check_key(key)?; Ok(self .map() .get(key) .map(|(b, _)| (b.clone(), content_version(b)))) } fn write_ref(&self, key: &str, bytes: &[u8], expected: Option<&Version>) -> Result { check_key(key)?; let mut map = self.map(); let current = map.get(key).map(|(b, _)| content_version(b)); check_expected(key, current.as_ref(), expected)?; map.insert(key.to_string(), (bytes.to_vec(), now_ms())); Ok(content_version(bytes)) } } // ---------------------------------------------------------------- conformance /// Exercise a store under `prefix` (which must be free, and is left /// empty) and report what passed: object round trips, ranges, create-only /// refusal, listing, deletion, and compare-and-swap on a reference. Every /// store runs this in its tests, and `oxy remote check` runs it against a /// remote before trusting it with a repository. pub fn conformance(store: &dyn Store, prefix: &str) -> Result> { let mut passed = Vec::new(); let key = |name: &str| format!("{prefix}/{name}"); let fail = |what: &str| Error::Backend(format!("{}: {what}", store.describe())); let expect = |ok: bool, what: &str| if ok { Ok(()) } else { Err(fail(what)) }; expect( store.list(prefix)?.is_empty(), "the probe prefix is not empty", )?; expect( !store.has(&key("a"))?, "has() of a missing object says true", )?; expect( matches!(store.get(&key("a")), Err(Error::NotFound(_))), "get() of a missing object is not NotFound", )?; passed.push("missing objects are reported missing"); let a: Vec = (0..256u32 * 1024).map(|i| (i % 251) as u8).collect(); store.put_new(&key("a"), &a)?; expect(store.has(&key("a"))?, "has() after put_new says false")?; expect( store.get(&key("a"))? == a, "get() returns other bytes than put_new wrote", )?; passed.push("an object round-trips"); expect( store.get_range(&key("a"), 1000..1500)? == a[1000..1500], "get_range() returns the wrong bytes", )?; expect( store.get_range(&key("a"), 0..0)?.is_empty(), "an empty range is not empty", )?; expect( store.get_range(&key("a"), 0..a.len() as u64 + 1).is_err(), "a range past the end is not refused", )?; passed.push("ranges read correctly and are bounded"); expect( matches!(store.put_new(&key("a"), b"other"), Err(Error::Exists(_))), "put_new() over an existing object is not refused", )?; expect( store.get(&key("a"))? == a, "a refused put_new() changed the object", )?; passed.push("objects are written once"); store.put_new(&key("dir/b"), b"bb")?; let listed: Vec<(String, u64)> = store .list(prefix)? .into_iter() .map(|e| (e.key, e.size)) .collect(); expect( listed == vec![(key("a"), a.len() as u64), (key("dir/b"), 2)], "list() does not return the two objects, sorted, with their sizes", )?; expect( store .stat(&key("dir/b"))? .is_some_and(|e| e.key == key("dir/b") && e.size == 2) && store.stat(&key("dir/c"))?.is_none(), "stat() does not describe one object, or describes a missing one", )?; expect( store .list(&key("dir/"))? .into_iter() .map(|e| (e.key, e.size)) .collect::>() == vec![(key("dir/b"), 2)], "list() of a sub-prefix is wrong", )?; expect( store.list(&key("nothing/"))?.is_empty(), "list() of an unused prefix is not empty", )?; passed.push("listing by prefix"); store.delete(&key("dir/b"))?; expect( !store.has(&key("dir/b"))?, "an object is still there after delete()", )?; store.delete(&key("dir/b"))?; passed.push("deletion, and deleting again, succeed"); let r = key("refs"); expect( store.read_ref(&r)?.is_none(), "an unwritten reference reads as something", )?; let v1 = store.write_ref(&r, b"one", None)?; expect( matches!(store.write_ref(&r, b"again", None), Err(Error::Conflict(_))), "a second create-only write of the reference is not refused", )?; let (bytes, seen) = store .read_ref(&r)? .ok_or_else(|| fail("the reference vanished"))?; expect( bytes == b"one" && seen == v1, "the reference reads back wrong", )?; let v2 = store.write_ref(&r, b"two", Some(&v1))?; expect(v2 != v1, "the version did not change on write")?; expect( matches!( store.write_ref(&r, b"three", Some(&v1)), Err(Error::Conflict(_)) ), "a write with a stale version is not refused", )?; let (bytes, seen) = store .read_ref(&r)? .ok_or_else(|| fail("the reference vanished"))?; expect( bytes == b"two" && seen == v2, "a refused write changed the reference", )?; passed.push("references update by compare-and-swap"); store.delete(&key("a"))?; store.delete(&r)?; expect( store.list(prefix)?.is_empty(), "the probe prefix is not empty after cleanup", )?; passed.push("cleanup"); Ok(passed) } #[cfg(test)] mod tests { use super::*; use std::sync::Arc; #[test] fn keys_are_plain_relative_paths() { for ok in ["a", "packs/ab.pack", "a-b_c.d/e"] { check_key(ok).unwrap(); } for bad in [ "", "/a", "a/", "a//b", "..", "a/../b", "./a", "a b", "a\\b", "รค", ] { assert!(check_key(bad).is_err(), "{bad:?}"); } assert!(check_key(&"x".repeat(MAX_KEY_LEN + 1)).is_err()); } #[test] fn the_memory_store_conforms() { let store = MemoryStore::new(); let passed = conformance(&store, "probe").unwrap(); assert_eq!(passed.len(), 8, "{passed:?}"); } #[test] fn the_directory_store_conforms() { let dir = tempfile::tempdir().unwrap(); let store = LocalStore::open(dir.path().join("bucket")).unwrap(); let passed = conformance(&store, "repos/one").unwrap(); assert_eq!(passed.len(), 8, "{passed:?}"); // Nothing but the (empty) directories and no stray temporaries. let stray: Vec<_> = walkdir(dir.path()) .into_iter() .filter(|p| p.is_file()) .collect(); assert!(stray.is_empty(), "{stray:?}"); assert_eq!( fs::read_dir(dir.path().join("bucket")).unwrap().count(), 0, "no empty directories left" ); } fn walkdir(dir: &Path) -> Vec { let mut out = vec![]; if let Ok(rd) = fs::read_dir(dir) { for e in rd.flatten() { let p = e.path(); if p.is_dir() { out.extend(walkdir(&p)); } else { out.push(p); } } } out } #[test] fn a_directory_store_refuses_keys_that_leave_its_root() { let dir = tempfile::tempdir().unwrap(); let store = LocalStore::open(dir.path().join("bucket")).unwrap(); assert!(matches!( store.put_new("../escape", b"x"), Err(Error::Key(..)) )); assert!(matches!(store.get("/etc/passwd"), Err(Error::Key(..)))); assert!(!dir.path().join("escape").exists()); } /// Many threads race to move a reference from the same version: exactly /// one wins each round, on both stores. #[test] fn compare_and_swap_admits_one_writer_per_round() { let dir = tempfile::tempdir().unwrap(); let local: Arc = Arc::new(LocalStore::open(dir.path()).unwrap()); let memory: Arc = Arc::new(MemoryStore::new()); for store in [local, memory] { let v0 = store.write_ref("refs", b"0", None).unwrap(); let wins: Vec = (0..8) .map(|i| { let store = store.clone(); let v0 = v0.clone(); std::thread::spawn(move || { store .write_ref("refs", format!("{}", i + 1).as_bytes(), Some(&v0)) .is_ok() }) }) .collect::>() .into_iter() .map(|h| h.join().unwrap()) .collect(); assert_eq!( wins.iter().filter(|w| **w).count(), 1, "{}: {wins:?}", store.describe() ); let (bytes, _) = store.read_ref("refs").unwrap().unwrap(); assert_ne!(bytes, b"0"); } } #[test] fn a_stale_lock_is_broken_and_a_live_one_waited_for() { let dir = tempfile::tempdir().unwrap(); let store = LocalStore::open(dir.path()).unwrap(); let lock = dir.path().join(".refs.lock"); fs::write(&lock, "dead").unwrap(); let old = std::time::SystemTime::now() - LOCK_STALE - Duration::from_secs(5); fs::File::open(&lock).unwrap().set_modified(old).unwrap(); store.write_ref("refs", b"after a crash", None).unwrap(); assert!(!lock.exists()); } #[test] fn temporaries_and_locks_are_not_listed() { let dir = tempfile::tempdir().unwrap(); let store = LocalStore::open(dir.path()).unwrap(); store.put_new("packs/a", b"a").unwrap(); fs::write(dir.path().join("packs/.a.tmp-ffffff"), "junk").unwrap(); fs::write(dir.path().join(".refs.lock"), "1").unwrap(); let keys: Vec = store.list("").unwrap().into_iter().map(|e| e.key).collect(); assert_eq!(keys, ["packs/a"]); } }