fix(runtime): rootfs-cache in-use guard prevents concurrent same-image corruption

RootfsCache::prune (called after a cache-miss put) evicted LRU entries with no
in-use guard, so it could remove_dir_all a cache entry that a CONCURRENT box was
using as its overlayfs lowerdir — that box's mount(2) then failed with ENOENT
('No such file or directory (os error 2)'), persisting through retries since the
backing was gone. Two pipelines from the same image collapse onto one cache key,
so this hit any concurrent same-image workload.

Fix: the same in-use guard SnapshotStore::prune already applies to live CoW
lowers. Each overlay box records the cache key it holds in <box_dir>/.rootfs-
cache-key (removed with the box dir); prune skips any still-referenced key
(prune_protecting, with prune as the empty-set wrapper).

Found via a concurrent-pipeline chaos test driven through a3s-code; verified on
a real /dev/kvm host (concurrency scenario: ~50% failure -> reliably green).
41 rootfs-cache unit tests + clippy clean.

Author: Roy Lin

CHANGELOG.md

@@ -41,6 +41,21 @@ All notable changes to A3S Box will be documented in this file.
   (use `StepResult::combined()` for the old concatenated view). Breaking for
   direct `.logs` field access on the `a3s-box-sdk` pipeline API.
 
+### Fixed
+
+- **Concurrent same-image pipelines could corrupt each other's rootfs cache.**
+  `RootfsCache::prune` (run after a cache-miss `put`) evicted least-recently-used
+  entries with no in-use guard, so it could `remove_dir_all` a cache entry that
+  another box was simultaneously using as its overlayfs **lowerdir** — the peer's
+  `mount(2)` then failed with `No such file or directory (os error 2)`, and the
+  failure persisted through retries (the backing was gone). Added the same in-use
+  guard `SnapshotStore::prune` already applies to live copy-on-write lowers: each
+  overlay box records the cache key it holds in a `<box_dir>/.rootfs-cache-key`
+  marker (removed with the box dir), and `prune` skips any still-referenced key.
+  Found via a concurrent-pipeline chaos test driven through a3s-code; root-caused
+  and verified on a real `/dev/kvm` host (the concurrency scenario went from ~50%
+  failure to reliably green).
+
 ## [2.6.0] — 2026-06-26
 
 ### Added

src/runtime/src/cache/rootfs_cache.rs

@@ -190,11 +190,27 @@ impl RootfsCache {
         Ok(())
     }
 
-    /// Prune the cache to stay within the given entry count limit.
+    /// Prune the cache to stay within the given entry count / byte limit.
     ///
-    /// Evicts least-recently-accessed entries first.
-    /// Returns the number of entries evicted.
+    /// Evicts least-recently-accessed entries first. Returns the number evicted.
     pub fn prune(&self, max_entries: usize, max_bytes: u64) -> Result<usize> {
+        self.prune_protecting(max_entries, max_bytes, &std::collections::HashSet::new())
+    }
+
+    /// Like [`RootfsCache::prune`], but never evicts an entry whose key is in
+    /// `protected`. Such an entry is currently serving as a box's overlayfs
+    /// **lowerdir**, and `remove_dir_all`-ing it out from under a concurrent box's
+    /// `mount(2)` makes the mount fail with ENOENT ("No such file or directory").
+    /// This is the same in-use guard [`crate::SnapshotStore::prune`] applies to
+    /// live copy-on-write lowers — without it, two pipelines built from the same
+    /// image (one cache-hit overlay box, one cache-miss box that prunes after its
+    /// put) can race and corrupt each other.
+    pub fn prune_protecting(
+        &self,
+        max_entries: usize,
+        max_bytes: u64,
+        protected: &std::collections::HashSet<String>,
+    ) -> Result<usize> {
         let mut entries = self.list_entries()?;
 
         if entries.len() <= max_entries {
@@ -215,6 +231,11 @@ impl RootfsCache {
             if current_count <= max_entries && current_size <= max_bytes {
                 break;
             }
+            // Never evict an entry in use as a live overlay lower — deleting the
+            // lowerdir under a concurrent box's mount(2) is the bug this guards.
+            if protected.contains(&entry.key) {
+                continue;
+            }
             self.invalidate(&entry.key)?;
             current_count -= 1;
             current_size = current_size.saturating_sub(entry.size_bytes);
@@ -495,6 +516,48 @@ mod tests {
         assert_eq!(cache.entry_count().unwrap(), 1);
     }
 
+    #[test]
+    fn prune_protecting_never_evicts_in_use_key() {
+        let tmp = TempDir::new().unwrap();
+        let cache = RootfsCache::new(tmp.path()).unwrap();
+        for i in 0..4 {
+            let src = tmp.path().join(format!("s{i}"));
+            create_test_rootfs(&src, &[("f", "x")]);
+            cache.put(&format!("k{i}"), &src, &format!("e{i}")).unwrap();
+            std::thread::sleep(std::time::Duration::from_millis(10));
+        }
+        // k0 is the OLDEST (normally evicted first) but is in use as an overlay lower.
+        let mut protected = std::collections::HashSet::new();
+        protected.insert("k0".to_string());
+        // keep=2 over 4 entries evicts two; the protected k0 is never one of them.
+        // (last_accessed is second-resolution, so WHICH two unprotected entries go
+        // is not asserted — only that the in-use lower survives.)
+        let evicted = cache.prune_protecting(2, u64::MAX, &protected).unwrap();
+        assert_eq!(evicted, 2, "two unprotected entries evicted to meet keep=2");
+        assert!(
+            cache.get("k0").unwrap().is_some(),
+            "the in-use (protected) lower must survive prune"
+        );
+        assert_eq!(cache.entry_count().unwrap(), 2, "k0 + one unprotected remain");
+    }
+
+    #[test]
+    fn prune_protecting_keeps_all_when_all_in_use() {
+        let tmp = TempDir::new().unwrap();
+        let cache = RootfsCache::new(tmp.path()).unwrap();
+        for i in 0..2 {
+            let src = tmp.path().join(format!("s{i}"));
+            create_test_rootfs(&src, &[("f", "x")]);
+            cache.put(&format!("k{i}"), &src, "e").unwrap();
+        }
+        let protected: std::collections::HashSet<String> =
+            ["k0", "k1"].iter().map(|s| s.to_string()).collect();
+        // Even asked to keep 0, nothing is evicted — every entry is a live lower.
+        let evicted = cache.prune_protecting(0, 0, &protected).unwrap();
+        assert_eq!(evicted, 0, "all in-use -> nothing evicted");
+        assert_eq!(cache.entry_count().unwrap(), 2);
+    }
+
     #[test]
     fn test_rootfs_cache_metadata_persists() {
         let tmp = TempDir::new().unwrap();

src/runtime/src/vm/layout.rs

@@ -156,6 +156,9 @@ impl VmManager {
             let cache_key = RootfsCache::compute_key(reference, &[], &[], &[]);
             if let Some(cached_path) = self.try_rootfs_cache_path(&cache_key)? {
                 let rootfs_path = self.rootfs_provider.prepare(&box_dir, &cached_path)?;
+                // Record that this box holds `cache_key` as its overlay lower, so a
+                // concurrent box's cache prune won't evict it mid-mount (ENOENT).
+                self.mark_rootfs_cache_key(&box_dir, &cache_key);
                 let tee_instance_config = self.generate_tee_config(&box_dir)?;
                 return Ok(BoxLayout {
                     rootfs_path,
@@ -204,6 +207,9 @@ impl VmManager {
                     prom.rootfs_cache_hits.inc();
                 }
                 let rootfs_path = self.rootfs_provider.prepare(&box_dir, &cached_path)?;
+                // Record that this box holds `cache_key` as its overlay lower, so a
+                // concurrent box's cache prune won't evict it mid-mount (ENOENT).
+                self.mark_rootfs_cache_key(&box_dir, &cache_key);
 
                 if let Ok(guest_init_path) = Self::find_guest_init() {
                     tracing::info!(
@@ -377,10 +383,14 @@ impl VmManager {
                     description = %description,
                     "Stored rootfs in cache"
                 );
-                // Prune if needed
-                if let Err(e) = cache.prune(
+                // Prune if needed — but never evict a cache entry that is in use as
+                // a live overlay lower for a concurrent box (deleting the lowerdir
+                // under its mount(2) is the same-image concurrency bug this guards).
+                let protected = self.referenced_rootfs_cache_keys();
+                if let Err(e) = cache.prune_protecting(
                     self.config.cache.max_rootfs_entries,
                     self.config.cache.max_cache_bytes,
+                    &protected,
                 ) {
                     tracing::warn!(error = %e, "Failed to prune rootfs cache");
                 }
@@ -391,6 +401,29 @@ impl VmManager {
         }
     }
 
+    /// Record which rootfs-cache key this box holds as its overlay lower, in a
+    /// `<box_dir>/.rootfs-cache-key` marker (mirror of the snapshot store's
+    /// `.snapshot-lower`). Read back by [`Self::referenced_rootfs_cache_keys`] so
+    /// the cache prune never evicts a live lower. Best-effort; removed with box_dir.
+    fn mark_rootfs_cache_key(&self, box_dir: &Path, cache_key: &str) {
+        let _ = std::fs::write(box_dir.join(".rootfs-cache-key"), cache_key);
+    }
+
+    /// Rootfs-cache keys currently in use as an overlay lower by some live box.
+    /// Boxes live under `<home>/boxes/<id>/`; a removed box's marker is gone with
+    /// its dir, so an evictable key is simply one no live box references.
+    fn referenced_rootfs_cache_keys(&self) -> std::collections::HashSet<String> {
+        let mut set = std::collections::HashSet::new();
+        if let Ok(entries) = std::fs::read_dir(self.home_dir.join("boxes")) {
+            for entry in entries.flatten() {
+                if let Ok(k) = std::fs::read_to_string(entry.path().join(".rootfs-cache-key")) {
+                    set.insert(k.trim().to_string());
+                }
+            }
+        }
+        set
+    }
+
     /// Resolve the cache directory from config or default.
     pub(crate) fn resolve_cache_dir(&self) -> PathBuf {
         self.config