new: Warm local cache when a hit from remote cache. (#2592)

* Add impl.

* Update changelog.

* Handle zero blob.

Author: milesj

CHANGELOG.md

@@ -13,6 +13,8 @@
     performance and reduce disk space usage.
   - When the local/remote cache is missing a blob, we'll attempt to retrieve it from the other
     cache, which can improve cache hit rates in some scenarios.
+  - When a remote cache hit, we'll now warm the local cache with the hydrated manifest and its
+    blobs, so the next run resolves locally instead of round-tripping to the remote.
 - **Processes**
   - Improved our "stream and capture output" child process handling to operate on bytes instead of
     lines, which should resolve some edge cases with output not being written to the console, or

crates/cache-storage/src/manifest.rs

@@ -254,17 +254,18 @@ impl Manifest {
 
         for file in &self.files {
             if let Some(digest) = &file.digest {
-                if let Some(bytes) = &file.bytes {
-                    sources.push(BlobInput {
-                        content: BlobContent::Inline(bytes.clone()),
-                        digest: digest.to_owned(),
-                    });
+                let content = if digest.size == 0 {
+                    BlobContent::Inline(Bytes::new())
+                } else if let Some(bytes) = &file.bytes {
+                    BlobContent::Inline(bytes.clone())
                 } else {
-                    sources.push(BlobInput {
-                        content: BlobContent::File(file.path.to_logical_path(workspace_root)),
-                        digest: digest.to_owned(),
-                    });
-                }
+                    BlobContent::File(file.path.to_logical_path(workspace_root))
+                };
+
+                sources.push(BlobInput {
+                    content,
+                    digest: digest.to_owned(),
+                });
             }
         }
 

crates/cache-storage/src/storage.rs

@@ -29,7 +29,7 @@ pub struct CacheContext {
     pub workspace_root: PathBuf,
 }
 
-#[derive(Debug)]
+#[derive(Clone, Debug)]
 pub struct StorageOptions {
     pub only_backends: Vec<Id>,
     pub include_local: bool,
@@ -115,25 +115,43 @@ impl Storage {
     }
 
     pub fn get_backends(&self) -> Vec<&BoxedStorageBackend> {
-        let mut backends = vec![];
+        self.get_backends_with_options(&self.options)
+    }
 
-        if self.options.only_backends.is_empty() {
-            if self.options.include_local {
-                backends.extend(self.local_backends.iter());
-            }
+    pub fn get_backends_with_options(&self, options: &StorageOptions) -> Vec<&BoxedStorageBackend> {
+        let mut backends = vec![];
 
-            if self.options.include_remote {
-                backends.extend(self.remote_backends.iter());
-            }
-        } else {
+        if options.include_local {
             backends.extend(self.local_backends.iter());
+        }
+
+        if options.include_remote {
             backends.extend(self.remote_backends.iter());
-            backends.retain(|backend| self.options.only_backends.contains(backend.get_id()));
+        }
+
+        if !options.only_backends.is_empty() {
+            backends.retain(|backend| options.only_backends.contains(backend.get_id()));
         }
 
         backends
     }
 
+    pub fn get_local_backends(&self) -> Vec<&BoxedStorageBackend> {
+        self.get_backends_with_options(&StorageOptions {
+            // Respect previously configured options
+            include_remote: false,
+            ..self.options.clone()
+        })
+    }
+
+    pub fn get_remote_backends(&self) -> Vec<&BoxedStorageBackend> {
+        self.get_backends_with_options(&StorageOptions {
+            // Respect previously configured options
+            include_local: false,
+            ..self.options.clone()
+        })
+    }
+
     pub fn is_local_enabled(&self) -> bool {
         !self.local_backends.is_empty()
     }
@@ -227,7 +245,7 @@ impl Storage {
                 continue;
             }
 
-            background_tasks.push(tokio::spawn(Box::pin(archive_manifest_in_backend(
+            background_tasks.push(tokio::spawn(Box::pin(persist_manifest_in_backend(
                 Arc::clone(backend),
                 digest.to_owned(),
                 manifest.clone(),
@@ -254,7 +272,7 @@ impl Storage {
         let ManifestSource {
             mut manifest,
             backend: original_backend,
-            ..
+            remote,
         } = manifest_source;
         let mut backends = VecDeque::from_iter(self.get_backends());
         let mut count = 1;
@@ -294,6 +312,13 @@ impl Storage {
                 "Hydrated cache manifest from {count} storage backends"
             );
 
+            // A remote hit leaves the local tier cold. Warm it from the
+            // now-in-memory blobs so the next run resolves locally instead of
+            // round-tripping to the remote again
+            if remote {
+                self.warm_local_backends(digest, &manifest).await;
+            }
+
             return Ok(Some(manifest));
         }
 
@@ -305,6 +330,33 @@ impl Storage {
         Ok(None)
     }
 
+    /// Warm the local tier after a remote cache hit by persisting the fully
+    /// hydrated manifest and its blobs into every active, writable local
+    /// backend, so the next run resolves locally instead of round-tripping to
+    /// the remote.
+    async fn warm_local_backends(&self, digest: &Digest, manifest: &Manifest) {
+        let mut background_tasks = self.background_tasks.lock().unwrap();
+
+        for backend in self.get_local_backends() {
+            if !backend.is_writable() {
+                continue;
+            }
+
+            trace!(
+                storage = backend.get_id().as_str(),
+                hash = digest.hash.as_str(),
+                "Warming local storage backend from remote cache hit"
+            );
+
+            background_tasks.push(tokio::spawn(Box::pin(persist_manifest_in_backend(
+                Arc::clone(backend),
+                digest.to_owned(),
+                manifest.clone(),
+                self.context.workspace_root.clone(),
+            ))));
+        }
+    }
+
     pub async fn wait_for_background_tasks(&self) -> miette::Result<()> {
         let background_tasks = {
             self.background_tasks
@@ -360,7 +412,7 @@ impl Storage {
     }
 }
 
-async fn archive_manifest_in_backend(
+async fn persist_manifest_in_backend(
     backend: BoxedStorageBackend,
     digest: Digest,
     mut manifest: Manifest,

crates/cache-storage/tests/manifest_test.rs

@@ -160,6 +160,34 @@ mod collect_blob_sources {
 
         assert_eq!(sources.len(), 2);
     }
+
+    #[test]
+    fn empty_file_uses_inline_empty_blob_not_a_path() {
+        // A size-0 output must become a shared inline empty blob, never a
+        // File(path): the path isn't materialized when warming runs, and empty
+        // files dedupe to the one empty digest in CAS.
+        let manifest = Manifest {
+            files: vec![
+                file(None, Some(digest('a', 0))),
+                file(Some(Bytes::from_static(b"hi")), Some(digest('c', 2))),
+            ],
+            ..Default::default()
+        };
+
+        let sources = manifest.collect_blob_inputs(Path::new("/workspace"));
+
+        assert_eq!(sources.len(), 2);
+
+        let empty = sources
+            .iter()
+            .find(|source| source.digest == digest('a', 0))
+            .expect("empty file should still produce a blob input");
+
+        match &empty.content {
+            BlobContent::Inline(bytes) => assert!(bytes.is_empty()),
+            _ => panic!("size-0 output must be an inline empty blob, not a file path"),
+        }
+    }
 }
 
 mod hydration {