use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicU64, Ordering};
use anyhow::Context as _;
use serde::{Deserialize, Serialize};
/// bump when the marker or payload changes, or old entries read as fresh.
const CACHE_SCHEMA_VERSION: u32 = 1;
/// a source tree carrying this would look like a finished entry.
pub const MARKER_FILE: &str = ".complete.json";
#[derive(Debug, Serialize, Deserialize, PartialEq, Eq)]
struct DirectoryMarker {
schema_version: u32,
kind: String,
key: String,
files: Vec<EntryFile>,
}
#[derive(Serialize, Deserialize)]
struct FileEntry<T> {
schema_version: u32,
kind: String,
key: String,
payload: T,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub(crate) struct EntryFile {
pub path: String,
pub bytes: u64,
}
pub(crate) fn encode_key(key: &str) -> String {
let mut out = String::with_capacity(key.len());
for byte in key.bytes() {
match byte {
b'A'..=b'Z' | b'a'..=b'z' | b'0'..=b'9' | b'-' | b'_' | b'.' => out.push(byte as char),
_ => out.push_str(&format!("%{byte:02X}")),
}
}
out
}
static TMP_COUNTER: AtomicU64 = AtomicU64::new(0);
fn tmp_name(prefix: &str) -> String {
format!(
".tmp-{prefix}.{}.{}",
std::process::id(),
TMP_COUNTER.fetch_add(1, Ordering::Relaxed)
)
}
#[derive(Debug, Clone)]
pub struct RepoStore {
root: PathBuf,
}
impl RepoStore {
pub fn open(repo_dir: &Path) -> Self {
Self {
root: repo_dir.join("cache"),
}
}
pub fn served(repo_dir: &Path) -> Self {
Self {
root: repo_dir.join("served"),
}
}
pub(crate) fn kind_dir(&self, kind: &str) -> PathBuf {
self.root.join(kind)
}
pub fn entry_path(&self, kind: &str, key: &str) -> PathBuf {
self.kind_dir(kind).join(format!("{}.d", encode_key(key)))
}
pub fn file_path(&self, kind: &str, key: &str) -> PathBuf {
self.kind_dir(kind)
.join(format!("{}.json", encode_key(key)))
}
pub fn read_file<T: serde::de::DeserializeOwned>(&self, kind: &str, key: &str) -> Option<T> {
let path = self.file_path(kind, key);
let entry: FileEntry<T> = serde_json::from_slice(&std::fs::read(path).ok()?).ok()?;
(entry.schema_version == CACHE_SCHEMA_VERSION && entry.kind == kind && entry.key == key)
.then_some(entry.payload)
}
pub fn publish_file<T: Serialize>(
&self,
kind: &str,
key: &str,
payload: &T,
) -> anyhow::Result<()> {
let entry = FileEntry {
schema_version: CACHE_SCHEMA_VERSION,
kind: kind.to_string(),
key: key.to_string(),
payload,
};
let path = self.file_path(kind, key);
crate::cas::atomic_write(&path, &serde_json::to_vec(&entry)?)
.with_context(|| format!("publish file entry {}", path.display()))
}
pub fn pool(&self, kind: &str) -> PathBuf {
self.kind_dir(kind)
}
pub fn read_directory(&self, kind: &str, key: &str) -> Option<PathBuf> {
let dir = self.entry_path(kind, key);
let marker_bytes = std::fs::read(dir.join(MARKER_FILE)).ok()?;
let marker: DirectoryMarker = serde_json::from_slice(&marker_bytes).ok()?;
if marker.schema_version != CACHE_SCHEMA_VERSION || marker.kind != kind || marker.key != key
{
return None;
}
for file in &marker.files {
let path = dir.join(&file.path);
let metadata = std::fs::symlink_metadata(&path).ok()?;
if !metadata.is_file() || metadata.len() != file.bytes {
return None;
}
}
Some(dir)
}
pub fn publish_directory(
&self,
kind: &str,
key: &str,
source_dir: &Path,
relative_exclusions: &[&str],
) -> anyhow::Result<PathBuf> {
let mut files = inventory_tree(source_dir, relative_exclusions)?;
files.retain(|file| file.path != MARKER_FILE);
let marker = DirectoryMarker {
schema_version: CACHE_SCHEMA_VERSION,
kind: kind.to_string(),
key: key.to_string(),
files: files.clone(),
};
let kind_dir = self.kind_dir(kind);
std::fs::create_dir_all(&kind_dir)?;
let final_dir = self.entry_path(kind, key);
let staging = kind_dir.join(tmp_name("dir"));
std::fs::create_dir_all(&staging)?;
for file in &files {
let source = source_dir.join(&file.path);
let target = staging.join(&file.path);
if let Some(parent) = target.parent() {
std::fs::create_dir_all(parent)?;
}
std::fs::hard_link(&source, &target)
.or_else(|_| std::fs::copy(&source, &target).map(|_| ()))?;
}
// a source marker is another entry's inode. replace it, don't write through.
crate::cas::atomic_write(&staging.join(MARKER_FILE), &serde_json::to_vec(&marker)?)?;
crate::cas::seal_read_only(&staging)?;
// the rename fails when a concurrent publisher won, or the entry is damaged.
if std::fs::rename(&staging, &final_dir).is_err() {
if self.read_directory(kind, key).is_some() {
let _ = std::fs::remove_dir_all(&staging);
return Ok(final_dir);
}
let _ = std::fs::remove_dir_all(&final_dir);
std::fs::rename(&staging, &final_dir)
.with_context(|| format!("publish directory entry {}", final_dir.display()))?;
}
Ok(final_dir)
}
}
pub fn inventory_tree(root: &Path, relative_exclusions: &[&str]) -> anyhow::Result<Vec<EntryFile>> {
fn visit(
root: &Path,
directory: &Path,
exclusions: &[&str],
files: &mut Vec<EntryFile>,
) -> anyhow::Result<()> {
for entry in std::fs::read_dir(directory)
.with_context(|| format!("read cache staging {}", directory.display()))?
{
let entry = entry?;
let path = entry.path();
let metadata = std::fs::symlink_metadata(&path)?;
if metadata.is_symlink() {
anyhow::bail!("unexpected symlink in cache staging: {}", path.display());
}
if metadata.is_dir() {
visit(root, &path, exclusions, files)?;
} else if metadata.is_file() {
let rel = path.strip_prefix(root).expect("staged path is below root");
if let Some(rel_str) = rel.to_str() {
if exclusions.contains(&rel_str) {
continue;
}
}
files.push(EntryFile {
path: rel.to_string_lossy().into_owned(),
bytes: metadata.len(),
});
} else {
anyhow::bail!("unsupported cache staging entry: {}", path.display());
}
}
Ok(())
}
let mut files = Vec::new();
if root.exists() {
visit(root, root, relative_exclusions, &mut files)?;
}
files.sort_by(|a, b| a.path.cmp(&b.path));
Ok(files)
}
#[cfg(test)]
mod tests {
use super::*;
use std::sync::{Arc, Barrier};
fn store() -> (tempfile::TempDir, RepoStore) {
let dir = tempfile::tempdir().unwrap();
let store = RepoStore::open(dir.path());
(dir, store)
}
#[test]
fn concurrent_equal_key_directory_install_publishes_one_complete_entry() {
let dir = tempfile::tempdir().unwrap();
let store = RepoStore::open(dir.path());
let sources: Vec<PathBuf> = (0..4)
.map(|index| {
let source = dir.path().join(format!("src-{index}"));
std::fs::create_dir_all(source.join("objects")).unwrap();
std::fs::write(source.join("objects/pack.bin"), format!("payload-{index}"))
.unwrap();
source
})
.collect();
let barrier = Arc::new(Barrier::new(sources.len()));
let handles: Vec<_> = sources
.iter()
.cloned()
.map(|source| {
let store = store.clone();
let barrier = Arc::clone(&barrier);
std::thread::spawn(move || {
barrier.wait();
let published = store.publish_directory("export", "same-key", &source, &[])?;
let entry = store
.read_directory("export", "same-key")
.expect("winner's entry must be a valid complete entry");
assert_eq!(entry, published);
let files: Vec<String> = std::fs::read_dir(&entry)
.unwrap()
.map(|e| e.unwrap().file_name().to_string_lossy().into_owned())
.collect();
Ok::<Vec<String>, anyhow::Error>(files)
})
})
.collect();
let mut observed = Vec::new();
for handle in handles {
observed.push(handle.join().unwrap());
}
for files in observed {
let files = files.unwrap_or_else(|error| {
panic!("every installer must observe a complete entry: {error:#}")
});
let mut names = files;
names.sort();
assert_eq!(
names,
vec![".complete.json".to_string(), "objects".to_string()],
"a loser must observe the winner's complete entry, not a mix"
);
}
let entry = store.read_directory("export", "same-key").expect("entry");
let payload = std::fs::read(entry.join("objects/pack.bin")).unwrap();
assert!(
sources.iter().any(|source| {
std::fs::read(source.join("objects/pack.bin")).unwrap() == payload
}),
"published payload must be exactly one installer's bytes"
);
let residue: Vec<_> = std::fs::read_dir(store.kind_dir("export"))
.unwrap()
.map(|e| e.unwrap().file_name().to_string_lossy().into_owned())
.filter(|name| name.starts_with(".tmp-"))
.collect();
assert!(
residue.is_empty(),
"no staging path may survive: {residue:?}"
);
}
#[test]
fn shared_staging_name_is_the_collision_the_barrier_test_detects() {
let dir = tempfile::tempdir().unwrap();
let store = RepoStore::open(dir.path());
let kind_dir = store.kind_dir("export");
std::fs::create_dir_all(&kind_dir).unwrap();
let staging = kind_dir.join(".tmp-export.shared");
std::fs::create_dir_all(&staging).unwrap();
let final_dir = kind_dir.join(format!("{}.d", encode_key("same-key")));
let barrier = Arc::new(Barrier::new(2));
let handles: Vec<_> = ["a", "b"]
.iter()
.map(|name| {
let staging = staging.clone();
let final_dir = final_dir.clone();
let barrier = Arc::clone(&barrier);
std::thread::spawn(move || {
std::fs::write(staging.join(format!("{name}.bin")), name).unwrap();
barrier.wait();
std::fs::rename(&staging, &final_dir)
})
})
.collect();
let results: Vec<_> = handles
.into_iter()
.map(|handle| handle.join().unwrap())
.collect();
assert!(
results.iter().any(|result| result.is_err()),
"a shared staging name must make at least one installer fail: {results:?}"
);
let published: Vec<String> = std::fs::read_dir(&final_dir)
.unwrap()
.map(|e| e.unwrap().file_name().to_string_lossy().into_owned())
.collect();
assert!(
published.len() > 1,
"the shared staging path published a foreign file: {published:?}"
);
}
#[test]
fn staging_names_are_unique_across_threads_and_embed_the_pid() {
let barrier = Arc::new(Barrier::new(8));
let handles: Vec<_> = (0..8)
.map(|_| {
let barrier = Arc::clone(&barrier);
std::thread::spawn(move || {
barrier.wait();
(0..32).map(|_| tmp_name("fact")).collect::<Vec<_>>()
})
})
.collect();
let mut names = std::collections::BTreeSet::new();
let mut total = 0usize;
for handle in handles {
for name in handle.join().unwrap() {
total += 1;
assert!(
name.starts_with(&format!(".tmp-fact.{}", std::process::id())),
"{name} must carry the pid"
);
assert!(names.insert(name.clone()), "duplicate staging name {name}");
}
}
assert_eq!(names.len(), total);
}
#[test]
fn directory_entries_validate_marker_and_lengths_without_rehash() {
let dir = tempfile::tempdir().unwrap();
let staging = dir.path().join("objects");
std::fs::create_dir_all(staging.join("pack")).unwrap();
std::fs::write(staging.join("pack/pack-abc.pack"), b"pack-bytes").unwrap();
std::fs::write(staging.join("excluded.txt"), b"control-file").unwrap();
let (_dir2, store) = store();
let entry = store
.publish_directory("export", "fp-1", &staging, &["excluded.txt"])
.unwrap();
let hit = store.read_directory("export", "fp-1").unwrap();
assert!(hit.join("pack/pack-abc.pack").is_file());
assert!(
!hit.join("excluded.txt").exists(),
"exclusions stay out of the entry"
);
let pack = hit.join("pack/pack-abc.pack");
let replace = |bytes: &[u8]| {
std::fs::remove_file(&pack).unwrap();
std::fs::write(&pack, bytes).unwrap();
};
replace(b"pack");
assert!(store.read_directory("export", "fp-1").is_none());
replace(b"pack-bytes");
std::fs::remove_file(hit.join(".complete.json")).unwrap();
assert!(store.read_directory("export", "fp-1").is_none());
let _ = entry;
}
#[test]
fn publishing_never_writes_through_a_carried_marker() {
let (dir, store) = store();
let prior = store
.publish_directory("export", "prior", &seed(dir.path(), "prior-src"), &[])
.unwrap();
let prior_marker = std::fs::read(prior.join(MARKER_FILE)).unwrap();
let source = seed(dir.path(), "next-src");
std::fs::hard_link(prior.join(MARKER_FILE), source.join(MARKER_FILE)).unwrap();
let next = store
.publish_directory("export", "next", &source, &[])
.unwrap();
assert_eq!(
std::fs::read(prior.join(MARKER_FILE)).unwrap(),
prior_marker
);
assert!(store.read_directory("export", "prior").is_some());
assert!(store.read_directory("export", "next").is_some());
assert_ne!(std::fs::read(next.join(MARKER_FILE)).unwrap(), prior_marker);
}
#[test]
fn publishing_never_reads_file_contents() {
use std::os::unix::fs::PermissionsExt as _;
let (dir, store) = store();
let source = seed(dir.path(), "src");
let pack = source.join("objects/pack.bin");
std::fs::set_permissions(&pack, std::fs::Permissions::from_mode(0o000)).unwrap();
if std::fs::File::open(&pack).is_ok() {
return;
}
store
.publish_directory("export", "k", &source, &[])
.unwrap();
assert!(store.read_directory("export", "k").is_some());
}
#[test]
fn markers_with_digests_still_read() {
let (dir, store) = store();
let entry = store
.publish_directory("export", "k", &seed(dir.path(), "src"), &[])
.unwrap();
let path = entry.join(MARKER_FILE);
let mut marker: serde_json::Value =
serde_json::from_slice(&std::fs::read(&path).unwrap()).unwrap();
marker["files"][0]["sha256"] = "00".repeat(32).into();
std::fs::remove_file(&path).unwrap();
std::fs::write(&path, serde_json::to_vec(&marker).unwrap()).unwrap();
assert!(store.read_directory("export", "k").is_some());
}
fn seed(root: &Path, name: &str) -> PathBuf {
let source = root.join(name);
std::fs::create_dir_all(source.join("objects")).unwrap();
std::fs::write(source.join("objects/pack.bin"), name).unwrap();
source
}
#[test]
fn file_entries_round_trip_under_their_own_kind_and_key_only() {
let (_dir, store) = store();
store.publish_file("state", "k-1", &vec![1, 2, 3]).unwrap();
assert_eq!(
store.read_file::<Vec<u8>>("state", "k-1"),
Some(vec![1, 2, 3])
);
assert_eq!(store.read_file::<Vec<u8>>("state", "k-2"), None);
assert_eq!(store.read_file::<Vec<u8>>("other", "k-1"), None);
let path = store.kind_dir("state").join("k-1.json");
let mut entry: serde_json::Value =
serde_json::from_slice(&std::fs::read(&path).unwrap()).unwrap();
entry["key"] = "k-2".into();
std::fs::write(&path, serde_json::to_vec(&entry).unwrap()).unwrap();
assert_eq!(
store.read_file::<Vec<u8>>("state", "k-1"),
None,
"an entry under another key's name is a miss"
);
entry["key"] = "k-1".into();
entry["schema_version"] = (CACHE_SCHEMA_VERSION + 1).into();
std::fs::write(&path, serde_json::to_vec(&entry).unwrap()).unwrap();
assert_eq!(store.read_file::<Vec<u8>>("state", "k-1"), None);
}
#[test]
fn key_encoding_is_injective_and_filesystem_safe() {
assert_eq!(encode_key("x:y"), "x%3Ay");
assert_eq!(encode_key("plain-text.v1"), "plain-text.v1");
assert_ne!(encode_key("b/c"), encode_key("b:c"));
assert_ne!(encode_key("a%b"), encode_key("a%25b"));
assert_eq!(encode_key("../../etc"), "..%2F..%2Fetc");
assert!(!encode_key("../../etc").contains('/'));
}
}