mise_cache_core/
local.rs

1use crate::{CacheDigest, RemoteActionResult, canonical_json};
2use eyre::{Result, bail};
3use std::fs;
4use std::io::Write;
5use std::path::{Path, PathBuf};
6
7/// A validated, content-addressed store on the local filesystem.
8#[derive(Debug, Clone)]
9pub struct LocalCas {
10    root: PathBuf,
11}
12
13/// A local index from action digests to their referenced cache objects.
14#[derive(Debug, Clone)]
15pub struct LocalActionCache {
16    root: PathBuf,
17    cas: LocalCas,
18}
19
20impl LocalCas {
21    /// Create a local content-addressed store beneath `root`.
22    pub fn new(root: impl Into<PathBuf>) -> Self {
23        Self { root: root.into() }
24    }
25
26    /// Return the root shared by this store and its action-result index.
27    pub fn root(&self) -> &Path {
28        &self.root
29    }
30
31    /// Resolve the storage path for a validated digest.
32    pub fn path_for(&self, digest: &CacheDigest) -> Result<PathBuf> {
33        digest.validate()?;
34        Ok(self
35            .root
36            .join("cas/v1")
37            .join(&digest.algorithm)
38            .join(&digest.hash[..2])
39            .join(format!("{}-{}", digest.hash, digest.size)))
40    }
41
42    /// Find and verify a stored object.
43    pub fn find(&self, digest: &CacheDigest) -> Result<Option<PathBuf>> {
44        let path = self.path_for(digest)?;
45        if !path.exists() {
46            return Ok(None);
47        }
48        if !digest.matches_file(&path)? {
49            bail!(
50                "local CAS blob failed digest verification: {}",
51                path.display()
52            );
53        }
54        Ok(Some(path))
55    }
56
57    /// Atomically store bytes after verifying their declared digest.
58    pub fn store_bytes(&self, digest: &CacheDigest, bytes: &[u8]) -> Result<PathBuf> {
59        if !digest.matches_bytes(bytes)? {
60            bail!("bytes do not match the declared CAS digest");
61        }
62        self.store_with(digest, |temporary| {
63            temporary.write_all(bytes)?;
64            Ok(())
65        })
66    }
67
68    /// Atomically store a file after verifying its declared digest.
69    pub fn store_file(&self, digest: &CacheDigest, source: &Path) -> Result<PathBuf> {
70        self.store_file_inner(digest, source, true)
71    }
72
73    /// Store a file whose digest was already verified by this crate.
74    pub(crate) fn store_verified_file(
75        &self,
76        digest: &CacheDigest,
77        source: &Path,
78    ) -> Result<PathBuf> {
79        self.store_file_inner(digest, source, false)
80    }
81
82    fn store_file_inner(
83        &self,
84        digest: &CacheDigest,
85        source: &Path,
86        verify: bool,
87    ) -> Result<PathBuf> {
88        let destination = self.path_for(digest)?;
89        if let Some(existing) = self.find(digest)? {
90            return Ok(existing);
91        }
92        let parent = destination.parent().expect("CAS path has a parent");
93        fs::create_dir_all(parent)?;
94        let staging = tempfile::tempdir_in(parent)?;
95        let temporary = staging.path().join("blob");
96        reflink_copy::reflink_or_copy(source, &temporary)?;
97        let temporary = tempfile::TempPath::try_from_path(temporary)?;
98        make_owner_writable(&temporary)?;
99        fs::OpenOptions::new()
100            .write(true)
101            .open(&temporary)?
102            .sync_all()?;
103        if verify && !digest.matches_file(&temporary)? {
104            bail!("staged blob does not match the declared CAS digest");
105        }
106        if fs::metadata(&temporary)?.len() != digest.size {
107            bail!("staged blob size does not match the declared CAS digest");
108        }
109        match temporary.persist_noclobber(&destination) {
110            Ok(()) => Ok(destination),
111            Err(error) if error.error.kind() == std::io::ErrorKind::AlreadyExists => self
112                .find(digest)?
113                .ok_or_else(|| eyre::eyre!("concurrent CAS write did not publish a valid blob")),
114            Err(error) => Err(error.error.into()),
115        }
116    }
117
118    fn store_with(
119        &self,
120        digest: &CacheDigest,
121        write: impl FnOnce(&mut tempfile::NamedTempFile) -> Result<()>,
122    ) -> Result<PathBuf> {
123        let destination = self.path_for(digest)?;
124        if let Some(existing) = self.find(digest)? {
125            return Ok(existing);
126        }
127        let parent = destination.parent().expect("CAS path has a parent");
128        fs::create_dir_all(parent)?;
129        let mut temporary = tempfile::NamedTempFile::new_in(parent)?;
130        write(&mut temporary)?;
131        temporary.flush()?;
132        temporary.as_file().sync_all()?;
133        if !digest.matches_file(temporary.path())? {
134            bail!("staged blob does not match the declared CAS digest");
135        }
136        match temporary.persist_noclobber(&destination) {
137            Ok(_) => Ok(destination),
138            Err(error) if error.error.kind() == std::io::ErrorKind::AlreadyExists => self
139                .find(digest)?
140                .ok_or_else(|| eyre::eyre!("concurrent CAS write did not publish a valid blob")),
141            Err(error) => Err(error.error.into()),
142        }
143    }
144}
145
146#[cfg(unix)]
147fn make_owner_writable(path: &Path) -> Result<()> {
148    use std::os::unix::fs::PermissionsExt as _;
149    let mut permissions = fs::metadata(path)?.permissions();
150    permissions.set_mode(permissions.mode() | 0o200);
151    fs::set_permissions(path, permissions)?;
152    Ok(())
153}
154
155#[cfg(windows)]
156fn make_owner_writable(path: &Path) -> Result<()> {
157    let mut permissions = fs::metadata(path)?.permissions();
158    permissions.set_readonly(false);
159    fs::set_permissions(path, permissions)?;
160    Ok(())
161}
162
163impl LocalActionCache {
164    /// Create an action-result index beneath `root`.
165    pub fn new(root: impl Into<PathBuf>) -> Self {
166        let root = root.into();
167        Self {
168            cas: LocalCas::new(root.clone()),
169            root,
170        }
171    }
172
173    /// Resolve the storage path for an action digest.
174    pub fn path_for(&self, action: &CacheDigest) -> Result<PathBuf> {
175        action.validate()?;
176        if action.algorithm != "blake3" {
177            bail!("local action keys must use blake3");
178        }
179        Ok(self
180            .root
181            .join("action-results/v1")
182            .join(&action.algorithm)
183            .join(&action.hash[..2])
184            .join(format!("{}-{}.json", action.hash, action.size)))
185    }
186
187    /// Find and strictly validate a canonical action result.
188    pub fn find(&self, action: &CacheDigest) -> Result<Option<RemoteActionResult>> {
189        let path = self.path_for(action)?;
190        if !path.exists() {
191            return Ok(None);
192        }
193        let bytes = fs::read(&path)?;
194        let result: RemoteActionResult = serde_json::from_slice(&bytes)?;
195        if result.version != 1 || result.action != *action || canonical_json(&result)? != bytes {
196            bail!("local action result is invalid: {}", path.display());
197        }
198        Ok(Some(result))
199    }
200
201    /// Atomically publish an action result after validating all referenced objects.
202    pub fn store(&self, result: &RemoteActionResult) -> Result<PathBuf> {
203        if result.version != 1 {
204            bail!("unsupported local action result version");
205        }
206        for digest in [
207            Some(&result.action),
208            result.metadata.as_ref(),
209            result.output_root.as_ref(),
210        ]
211        .into_iter()
212        .flatten()
213        {
214            if self.cas.find(digest)?.is_none() {
215                bail!("cannot publish an action result with a missing blob");
216            }
217        }
218        let destination = self.path_for(&result.action)?;
219        let replace_invalid = match self.find(&result.action) {
220            Ok(Some(existing)) => {
221                if existing == *result {
222                    return Ok(destination);
223                }
224                bail!("local action key already has a different result");
225            }
226            Ok(None) => false,
227            Err(_) => true,
228        };
229        let parent = destination
230            .parent()
231            .expect("action-result path has a parent");
232        fs::create_dir_all(parent)?;
233        let mut temporary = tempfile::NamedTempFile::new_in(parent)?;
234        temporary.write_all(&canonical_json(result)?)?;
235        temporary.flush()?;
236        temporary.as_file().sync_all()?;
237        if replace_invalid {
238            temporary
239                .persist(&destination)
240                .map_err(|error| error.error)?;
241            return Ok(destination);
242        }
243        match temporary.persist_noclobber(&destination) {
244            Ok(_) => Ok(destination),
245            Err(error) if error.error.kind() == std::io::ErrorKind::AlreadyExists => self
246                .find(&result.action)?
247                .filter(|existing| existing == result)
248                .map(|_| destination)
249                .ok_or_else(|| eyre::eyre!("concurrent action write was invalid or conflicting")),
250            Err(error) => Err(error.error.into()),
251        }
252    }
253}
254
255#[cfg(test)]
256mod tests {
257    use super::*;
258
259    #[test]
260    fn stores_and_validates_blobs_atomically() {
261        let directory = tempfile::tempdir().unwrap();
262        let cas = LocalCas::new(directory.path());
263        let digest = CacheDigest::blake3(b"cached object");
264
265        let path = cas.store_bytes(&digest, b"cached object").unwrap();
266        assert_eq!(cas.find(&digest).unwrap(), Some(path.clone()));
267        assert_eq!(fs::read(&path).unwrap(), b"cached object");
268        assert_eq!(cas.store_bytes(&digest, b"cached object").unwrap(), path);
269        assert!(cas.store_bytes(&digest, b"other object").is_err());
270    }
271
272    #[test]
273    fn stored_files_are_independent_from_the_source() {
274        let directory = tempfile::tempdir().unwrap();
275        let cas = LocalCas::new(directory.path().join("cache"));
276        let source = directory.path().join("source");
277        fs::write(&source, b"cached object").unwrap();
278        let digest = CacheDigest::blake3(b"cached object");
279
280        let stored = cas.store_file(&digest, &source).unwrap();
281        fs::write(source, b"other object!").unwrap();
282
283        assert_eq!(fs::read(stored).unwrap(), b"cached object");
284        assert!(cas.find(&digest).unwrap().is_some());
285    }
286
287    #[test]
288    fn rejects_files_with_the_wrong_digest() {
289        let directory = tempfile::tempdir().unwrap();
290        let cas = LocalCas::new(directory.path().join("cache"));
291        let source = directory.path().join("source");
292        fs::write(&source, b"other object").unwrap();
293        let digest = CacheDigest::blake3(b"cached object");
294
295        assert!(cas.store_file(&digest, &source).is_err());
296        assert!(!cas.path_for(&digest).unwrap().exists());
297    }
298
299    #[test]
300    fn stores_read_only_source_files() {
301        let directory = tempfile::tempdir().unwrap();
302        let cas = LocalCas::new(directory.path().join("cache"));
303        let source = directory.path().join("source");
304        fs::write(&source, b"cached object").unwrap();
305        let mut permissions = fs::metadata(&source).unwrap().permissions();
306        permissions.set_readonly(true);
307        fs::set_permissions(&source, permissions).unwrap();
308        let digest = CacheDigest::blake3(b"cached object");
309
310        let stored = cas.store_file(&digest, &source).unwrap();
311
312        assert_eq!(fs::read(stored).unwrap(), b"cached object");
313        assert!(fs::metadata(&source).unwrap().permissions().readonly());
314        make_owner_writable(&source).unwrap();
315    }
316
317    #[test]
318    fn rejects_corrupt_existing_blobs() {
319        let directory = tempfile::tempdir().unwrap();
320        let cas = LocalCas::new(directory.path());
321        let digest = CacheDigest::blake3(b"cached object");
322        let path = cas.store_bytes(&digest, b"cached object").unwrap();
323        fs::write(path, b"corrupt").unwrap();
324
325        assert!(cas.find(&digest).is_err());
326    }
327
328    #[test]
329    fn publishes_action_results_after_referenced_blobs() {
330        let directory = tempfile::tempdir().unwrap();
331        let cas = LocalCas::new(directory.path());
332        let actions = LocalActionCache::new(directory.path());
333        let action = CacheDigest::blake3(b"action");
334        let metadata = CacheDigest::blake3(b"metadata");
335        let output_root = CacheDigest::blake3(b"directory");
336        let result = RemoteActionResult {
337            action: action.clone(),
338            metadata: Some(metadata.clone()),
339            output_root: Some(output_root.clone()),
340            version: 1,
341        };
342
343        assert!(actions.store(&result).is_err());
344        cas.store_bytes(&action, b"action").unwrap();
345        cas.store_bytes(&metadata, b"metadata").unwrap();
346        cas.store_bytes(&output_root, b"directory").unwrap();
347        actions.store(&result).unwrap();
348        assert_eq!(actions.find(&action).unwrap(), Some(result));
349    }
350
351    #[test]
352    fn atomically_replaces_a_corrupt_action_result() {
353        let directory = tempfile::tempdir().unwrap();
354        let cas = LocalCas::new(directory.path());
355        let actions = LocalActionCache::new(directory.path());
356        let action = CacheDigest::blake3(b"action");
357        let result = RemoteActionResult {
358            action: action.clone(),
359            metadata: None,
360            output_root: None,
361            version: 1,
362        };
363        cas.store_bytes(&action, b"action").unwrap();
364        let path = actions.path_for(&action).unwrap();
365        fs::create_dir_all(path.parent().unwrap()).unwrap();
366        fs::write(&path, b"truncated").unwrap();
367
368        assert!(actions.find(&action).is_err());
369        assert_eq!(actions.store(&result).unwrap(), path);
370        assert_eq!(actions.find(&action).unwrap(), Some(result));
371    }
372}
mise_cache_core/local.rs

mise_cache_core/
local.rs
