From e6d0de07ecd195c6c35cc9cd05721d2f5b6d49fb Mon Sep 17 00:00:00 2001 From: elrrrrrrr Date: Wed, 22 Jul 2026 12:12:15 +0800 Subject: [PATCH] fix(pm): serialize cross-process cache writes --- crates/pm/src/service/clean_cache.rs | 16 ++++ crates/pm/src/util/cloner.rs | 46 ++++++++- crates/pm/src/util/extractor.rs | 35 +++++-- crates/pm/src/util/mod.rs | 1 + crates/pm/src/util/process_lock.rs | 136 +++++++++++++++++++++++++++ 5 files changed, 226 insertions(+), 8 deletions(-) create mode 100644 crates/pm/src/util/process_lock.rs diff --git a/crates/pm/src/service/clean_cache.rs b/crates/pm/src/service/clean_cache.rs index 47aa2d3447..0cae89ab97 100644 --- a/crates/pm/src/service/clean_cache.rs +++ b/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(()) + } } diff --git a/crates/pm/src/util/cloner.rs b/crates/pm/src/util/cloner.rs index b29863b71e..0bd9ba3090 100644 --- a/crates/pm/src/util/cloner.rs +++ b/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 { + 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()?; diff --git a/crates/pm/src/util/extractor.rs b/crates/pm/src/util/extractor.rs index 70485114c6..351285a641 100644 --- a/crates/pm/src/util/extractor.rs +++ b/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"); + } } diff --git a/crates/pm/src/util/mod.rs b/crates/pm/src/util/mod.rs index be70b3e032..1f4e1c4cf7 100644 --- a/crates/pm/src/util/mod.rs +++ b/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; diff --git a/crates/pm/src/util/process_lock.rs b/crates/pm/src/util/process_lock.rs new file mode 100644 index 0000000000..c1594191d0 --- /dev/null +++ b/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 { + 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 { + 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 { + 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 { + 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); + } +}