fix(pm): serialize cross-process cache writes

crates/pm/src/service/clean_cache.rs

@@ -20,6 +20,9 @@ async fn collect_matching_versions(
 ) -> Result<()> {
     let mut version_entries = fs::read_dir(path).await?;
     while let Some(version_entry) = version_entries.next_entry().await? {
+        if !version_entry.file_type().await?.is_dir() {
+            continue;
+        }
         let version = version_entry.file_name();
         let version_str = version.to_string_lossy();
         if matches_pattern(&version_str, version_pattern) {
@@ -170,4 +173,17 @@ mod tests {
         assert_eq!(to_delete.len(), 0);
         Ok(())
     }
+
+    #[tokio::test]
+    async fn test_collect_matching_versions_ignores_lock_files() -> Result<()> {
+        let temp_dir = setup_test_dir().await?;
+        fs::write(temp_dir.path().join(".1.0.0.lock"), b"").await?;
+        let mut to_delete = Vec::new();
+
+        collect_matching_versions(temp_dir.path(), "test-pkg".to_string(), "*", &mut to_delete)
+            .await?;
+
+        assert_eq!(to_delete.len(), 4);
+        Ok(())
+    }
 }

crates/pm/src/util/cloner.rs

@@ -11,6 +11,7 @@ use serde::de::DeserializeOwned;
 use utoo_ruborist::manifest::IdentityView;
 
 use super::downloader::is_git_url;
+use super::process_lock::{lock_exclusive_sync, sibling_lock_path};
 
 /// How a cached package's real contents are laid out under its cache dir.
 /// Derived from the resolved tarball URL so callers never pass a bare bool.
@@ -80,10 +81,24 @@ mod hardlink_clone {
     }
 
     fn copy_file_sync(src: &Path, dst: &Path) -> io::Result<()> {
-        fs::copy(src, dst)?;
+        let mut src_file = fs::File::open(src)?;
+        #[cfg(unix)]
+        let src_perms = fs::metadata(src)?.permissions();
+
+        // `dst` may already be a hardlink to `src`. Unlink it before copy,
+        // then create exclusively so a racing writer cannot reintroduce it.
+        match fs::remove_file(dst) {
+            Ok(()) => {}
+            Err(error) if error.kind() == io::ErrorKind::NotFound => {}
+            Err(error) => return Err(error),
+        }
+        let mut dst_file = fs::OpenOptions::new()
+            .write(true)
+            .create_new(true)
+            .open(dst)?;
+        io::copy(&mut src_file, &mut dst_file)?;
         #[cfg(unix)]
         {
-            let src_perms = fs::metadata(src)?.permissions();
             fs::set_permissions(dst, src_perms)?;
         }
         Ok(())
@@ -281,6 +296,9 @@ fn clone_sync(src: &Path, dst: &Path, layout: CacheLayout, policy: ClonePolicy)
 /// valid directory was reused. Stateless — callers (the install scheduler) own
 /// dedup and counting.
 pub fn clone_package_sync(req: &PackageClone<'_>) -> Result<bool> {
+    let lock_path = sibling_lock_path(req.target, ".clone.lock")?;
+    let _lock = lock_exclusive_sync(&lock_path)?;
+
     if req.target.try_exists()? {
         if validate_name_version_sync(req.target, req.name, req.version) {
             return Ok(false);
@@ -521,6 +539,30 @@ mod tests {
             Ok(())
         }
 
+        #[cfg(target_os = "linux")]
+        #[tokio::test]
+        async fn copy_fallback_does_not_truncate_hardlinked_source() -> Result<()> {
+            let temp = TempDir::new()?;
+            let src_dir = temp.path().join("src");
+            let dst_dir = temp.path().join("dst");
+            fs::create_dir_all(&src_dir).await?;
+            fs::create_dir_all(&dst_dir).await?;
+            fs::write(src_dir.join("package.json"), b"cache contents").await?;
+            fs::hard_link(src_dir.join("package.json"), dst_dir.join("package.json")).await?;
+
+            hardlink_clone::clone_dir(&src_dir, &dst_dir, ClonePolicy::Shared).await?;
+
+            assert_eq!(
+                fs::read(src_dir.join("package.json")).await?,
+                b"cache contents"
+            );
+            assert_eq!(
+                fs::read(dst_dir.join("package.json")).await?,
+                b"cache contents"
+            );
+            Ok(())
+        }
+
         #[tokio::test]
         async fn test_clone_dir_nested() -> Result<()> {
             let temp = TempDir::new()?;

crates/pm/src/util/extractor.rs

@@ -32,10 +32,15 @@ use utoo_ruborist::tar::{
     commit_cache_dir_atomic, gzip_decompress, is_safe_tar_entry_path, normalize_entry_mode,
 };
 
+use super::process_lock::{lock_exclusive, sibling_lock_path};
+
 /// Extract gzip tarball bytes and atomically commit them to `dest`.
 ///
 /// Skips if the `_resolved` marker already exists (warm cache).
 pub async fn extract_and_write(gzip_bytes: Bytes, dest: &Path) -> Result<()> {
+    let lock_path = sibling_lock_path(dest, ".lock")?;
+    let _lock = lock_exclusive(&lock_path).await?;
+
     // Check if already resolved (warm cache scenario)
     let resolved_path = dest.join("_resolved");
     if crate::fs::try_exists(&resolved_path).await? {
@@ -314,12 +319,11 @@ mod tests {
 
         assert!(extract_and_write(truncated, &dest).await.is_err());
         assert!(!dest.exists());
-        let leftovers = if parent.exists() {
-            std::fs::read_dir(&parent).unwrap().count()
-        } else {
-            0
-        };
-        assert_eq!(leftovers, 0, "staging dir leaked into the cache parent");
+        let leftovers: Vec<_> = std::fs::read_dir(&parent)
+            .unwrap()
+            .map(|entry| entry.unwrap().file_name())
+            .collect();
+        assert_eq!(leftovers, vec![std::ffi::OsString::from(".1.0.0.lock")]);
     }
 
     /// A pre-staging-protocol slot (files present, no `_resolved`) is
@@ -338,4 +342,23 @@ mod tests {
         assert!(dest.join("package").join("index.js").exists());
         assert!(!dest.join("stale.js").exists());
     }
+
+    #[tokio::test]
+    async fn concurrent_extracts_share_one_complete_slot() {
+        let tmp = TempDir::new().unwrap();
+        let dest = tmp.path().join("pkg").join("1.0.0");
+        let first = gzip(&make_tar(&[("package/index.js", b"first")]));
+        let second = gzip(&make_tar(&[("package/index.js", b"second")]));
+
+        let (first_result, second_result) = tokio::join!(
+            extract_and_write(first, &dest),
+            extract_and_write(second, &dest)
+        );
+
+        first_result.unwrap();
+        second_result.unwrap();
+        assert!(dest.join("_resolved").exists());
+        let content = std::fs::read(dest.join("package/index.js")).unwrap();
+        assert!(content == b"first" || content == b"second");
+    }
 }

crates/pm/src/util/mod.rs

@@ -34,6 +34,7 @@ pub mod manifest_store;
 pub mod npmrc;
 pub mod package_cache;
 pub mod platform_const;
+pub mod process_lock;
 pub mod proxy_env;
 pub mod registry;
 pub mod retry;

crates/pm/src/util/process_lock.rs

@@ -0,0 +1,136 @@
+use std::ffi::OsString;
+use std::fs::{File, OpenOptions};
+use std::path::{Path, PathBuf};
+
+use anyhow::{Context, Result};
+
+/// An exclusive OS lock held for as long as the backing file handle is open.
+///
+/// The lock file is intentionally retained after release. Removing it could
+/// split waiters and new contenders across different filesystem objects.
+pub struct ProcessLock {
+    _file: File,
+}
+
+fn open_lock_file(lock_path: &Path) -> Result<File> {
+    if let Some(parent) = lock_path.parent() {
+        std::fs::create_dir_all(parent).with_context(|| {
+            format!(
+                "Failed to create lock parent directory {}",
+                parent.display()
+            )
+        })?;
+    }
+
+    OpenOptions::new()
+        .read(true)
+        .write(true)
+        .create(true)
+        .truncate(false)
+        .open(lock_path)
+        .with_context(|| format!("Failed to open lock file {}", lock_path.display()))
+}
+
+/// Acquire an exclusive process lock without blocking an async executor thread.
+pub async fn lock_exclusive(lock_path: &Path) -> Result<ProcessLock> {
+    let lock_path = lock_path.to_path_buf();
+    tokio::task::spawn_blocking(move || lock_exclusive_sync(&lock_path))
+        .await
+        .context("Lock task failed")?
+}
+
+/// Acquire an exclusive process lock from an existing blocking worker.
+pub fn lock_exclusive_sync(lock_path: &Path) -> Result<ProcessLock> {
+    let file = open_lock_file(lock_path)?;
+    file.lock()
+        .with_context(|| format!("Failed to acquire lock on {}", lock_path.display()))?;
+    Ok(ProcessLock { _file: file })
+}
+
+/// Return a stable sibling lock path outside the directory being protected.
+///
+/// `/cache/tapable/2.3.3` with suffix `.lock` becomes
+/// `/cache/tapable/.2.3.3.lock`.
+pub fn sibling_lock_path(path: &Path, suffix: &str) -> Result<PathBuf> {
+    let file_name = path
+        .file_name()
+        .with_context(|| format!("Lock target has no file name: {}", path.display()))?;
+    let mut lock_name = OsString::from(".");
+    lock_name.push(file_name);
+    lock_name.push(suffix);
+    Ok(path.with_file_name(lock_name))
+}
+
+#[cfg(test)]
+mod tests {
+    use std::process::Command;
+    use std::thread;
+    use std::time::{Duration, Instant};
+
+    use tempfile::TempDir;
+
+    use super::*;
+
+    const CHILD_LOCK_PATH: &str = "UTOO_PROCESS_LOCK_TEST_PATH";
+    const CHILD_STARTED_PATH: &str = "UTOO_PROCESS_LOCK_TEST_STARTED";
+    const CHILD_ACQUIRED_PATH: &str = "UTOO_PROCESS_LOCK_TEST_ACQUIRED";
+
+    #[test]
+    fn sibling_path_keeps_lock_outside_protected_directory() {
+        let target = Path::new("cache/tapable/2.3.3");
+        assert_eq!(
+            sibling_lock_path(target, ".lock").unwrap(),
+            Path::new("cache/tapable/.2.3.3.lock")
+        );
+    }
+
+    #[test]
+    fn exclusive_lock_serializes_processes_and_keeps_empty_file() {
+        if let Some(lock_path) = std::env::var_os(CHILD_LOCK_PATH) {
+            let started_path = PathBuf::from(std::env::var_os(CHILD_STARTED_PATH).unwrap());
+            let acquired_path = PathBuf::from(std::env::var_os(CHILD_ACQUIRED_PATH).unwrap());
+            std::fs::write(started_path, b"").unwrap();
+            let _lock = lock_exclusive_sync(Path::new(&lock_path)).unwrap();
+            std::fs::write(acquired_path, b"").unwrap();
+            return;
+        }
+
+        let temp = TempDir::new().unwrap();
+        let target = temp.path().join("tapable/2.3.3");
+        let lock_path = sibling_lock_path(&target, ".lock").unwrap();
+        let started_path = temp.path().join("child-started");
+        let acquired_path = temp.path().join("child-acquired");
+        let parent_lock = lock_exclusive_sync(&lock_path).unwrap();
+
+        let mut child = Command::new(std::env::current_exe().unwrap())
+            .arg("exclusive_lock_serializes_processes_and_keeps_empty_file")
+            .arg("--nocapture")
+            .env(CHILD_LOCK_PATH, &lock_path)
+            .env(CHILD_STARTED_PATH, &started_path)
+            .env(CHILD_ACQUIRED_PATH, &acquired_path)
+            .spawn()
+            .unwrap();
+
+        let deadline = Instant::now() + Duration::from_secs(5);
+        while !started_path.exists() && Instant::now() < deadline {
+            thread::sleep(Duration::from_millis(10));
+        }
+        if !started_path.exists() {
+            drop(parent_lock);
+            let _ = child.kill();
+            let _ = child.wait();
+            panic!("child process did not reach the lock attempt");
+        }
+
+        thread::sleep(Duration::from_millis(100));
+        assert!(
+            !acquired_path.exists(),
+            "child acquired an exclusive lock while the parent held it"
+        );
+
+        drop(parent_lock);
+        assert!(child.wait().unwrap().success());
+        assert!(acquired_path.exists());
+        assert_eq!(std::fs::metadata(lock_path).unwrap().len(), 0);
+    }
+}