From 74577d73fa5bd7030f5da91c83cdc299b9f39684 Mon Sep 17 00:00:00 2001 From: Zoltan Kochan Date: Thu, 6 Aug 2026 09:55:17 +0200 Subject: [PATCH] fix(deps-restorer): materialize each global-virtual-store slot once (#13669) The link pass spawned one CreateVirtualDirBySnapshot per snapshot key. Under the global virtual store, peer variants of a package whose recursive dependency hashes are equal map to one slot directory - that sharing is the point of the hashing - so hash-equal variants raced concurrent imports against a single path. Immutable packages tolerate this (their import is marker-gated and safe to skip), but a mutable `file:` source forces a stage-and-swap, whose remove-then-rename assumes an exclusive owner: one worker's `remove_dir_all` interleaves with another's rename and the install fails with `Directory not empty` or `NotFound` depending on timing. Injected workspace packages in a peer-heavy monorepo hit this on every install. pnpm v11 never faces the race because `lockfileToDepGraph` keys its graph by directory, so one node per location survives graph construction. Restore that invariant here by grouping slot links by `slot_dir` before the rayon pass: the first variant materializes the directory, the rest only report progress, and obsolete-child aliases recorded against any variant are unioned so stale links still get removed no matter which variant runs. Without the global virtual store every group is a singleton and the pass is unchanged. --- .changeset/gvs-dedup-slot-materialization.md | 5 + .../create_virtual_dir_by_snapshot/tests.rs | 10 + .../deps-restorer/src/create_virtual_store.rs | 91 +++++- .../src/create_virtual_store/tests.rs | 295 ++++++++++++++++++ 4 files changed, 393 insertions(+), 8 deletions(-) create mode 100644 .changeset/gvs-dedup-slot-materialization.md diff --git a/.changeset/gvs-dedup-slot-materialization.md b/.changeset/gvs-dedup-slot-materialization.md new file mode 100644 index 0000000000..c024572e85 --- /dev/null +++ b/.changeset/gvs-dedup-slot-materialization.md @@ -0,0 +1,5 @@ +--- +"pacquet": patch +--- + +Fixed installs under `enableGlobalVirtualStore` failing with `failed to remove existing directory ... prior to swap: Directory not empty` (or `No such file or directory`) when peer variants of an injected `file:` dependency hash to the same slot. The link pass now materializes each unique slot directory once instead of racing one force-mode import per peer variant against the same path. diff --git a/pnpm/crates/deps-restorer/src/create_virtual_dir_by_snapshot/tests.rs b/pnpm/crates/deps-restorer/src/create_virtual_dir_by_snapshot/tests.rs index 6cf0a97b8e..c3cfd46f31 100644 --- a/pnpm/crates/deps-restorer/src/create_virtual_dir_by_snapshot/tests.rs +++ b/pnpm/crates/deps-restorer/src/create_virtual_dir_by_snapshot/tests.rs @@ -19,6 +19,10 @@ use tempfile::tempdir; pub struct LinkConcurrencyProbe { current: AtomicUsize, max: AtomicUsize, + /// Total `enter()` calls over the probe's lifetime — one per + /// [`CreateVirtualDirBySnapshot::run`] — for asserting how many + /// link tasks executed, which `max_concurrent` cannot. + total: AtomicUsize, wait_for_overlap: bool, wait_started: AtomicBool, mutex: Mutex<()>, @@ -40,7 +44,12 @@ impl LinkConcurrencyProbe { self.max.load(Ordering::SeqCst) } + pub(crate) fn total_entered(&self) -> usize { + self.total.load(Ordering::SeqCst) + } + pub(super) fn enter(&self) -> LinkConcurrencyGuard<'_> { + self.total.fetch_add(1, Ordering::SeqCst); let current = self.current.fetch_add(1, Ordering::SeqCst) + 1; let mut max = self.max.load(Ordering::SeqCst); while current > max { @@ -73,6 +82,7 @@ impl Default for LinkConcurrencyProbe { Self { current: AtomicUsize::new(0), max: AtomicUsize::new(0), + total: AtomicUsize::new(0), wait_for_overlap: false, wait_started: AtomicBool::new(false), mutex: Mutex::new(()), diff --git a/pnpm/crates/deps-restorer/src/create_virtual_store.rs b/pnpm/crates/deps-restorer/src/create_virtual_store.rs index 91b7ff9c6f..fb63528cc1 100644 --- a/pnpm/crates/deps-restorer/src/create_virtual_store.rs +++ b/pnpm/crates/deps-restorer/src/create_virtual_store.rs @@ -873,6 +873,76 @@ struct SlotLink<'a> { removed_aliases: &'a [PkgName], } +/// One unique slot directory and every [`SlotLink`] that resolved to +/// it. Under the global virtual store, hash-equal peer variants share +/// a slot path, and `stage_and_swap` in +/// [`fn@crate::import_indexed_dir`] assumes an exclusive owner per +/// directory — so the link pass runs one task per group, with the +/// `removed_aliases` of every member unioned for cleanup. +struct SlotDirGroup<'a> { + representative: &'a SlotLink<'a>, + /// Kept so each warm variant still emits its own progress line. + duplicates: Vec<&'a SlotLink<'a>>, + /// `None` until a duplicate contributes an alias the + /// representative lacks. + merged_removed_aliases: Option>, +} + +impl SlotDirGroup<'_> { + fn removed_aliases(&self) -> &[PkgName] { + self.merged_removed_aliases.as_deref().unwrap_or(self.representative.removed_aliases) + } +} + +/// Group `slots` by [`crate::VirtualStoreLayout::slot_dir`], preserving +/// first-occurrence order. +fn group_slots_by_dir<'a>( + slots: &'a [SlotLink<'a>], + layout: &crate::VirtualStoreLayout, +) -> Vec> { + if !layout.enable_global_virtual_store() { + // Project-local slot names embed the peer-suffixed key: every + // group is a singleton, so skip the path construction. + return slots + .iter() + .map(|slot| SlotDirGroup { + representative: slot, + duplicates: Vec::new(), + merged_removed_aliases: None, + }) + .collect(); + } + let mut index_by_dir: HashMap = HashMap::with_capacity(slots.len()); + let mut groups: Vec> = Vec::with_capacity(slots.len()); + for slot in slots { + match index_by_dir.entry(layout.slot_dir(slot.snapshot_key)) { + std::collections::hash_map::Entry::Vacant(entry) => { + entry.insert(groups.len()); + groups.push(SlotDirGroup { + representative: slot, + duplicates: Vec::new(), + merged_removed_aliases: None, + }); + } + std::collections::hash_map::Entry::Occupied(entry) => { + let group = &mut groups[*entry.get()]; + group.duplicates.push(slot); + if !slot.removed_aliases.is_empty() { + let merged = group + .merged_removed_aliases + .get_or_insert_with(|| group.representative.removed_aliases.to_vec()); + for alias in slot.removed_aliases { + if !merged.contains(alias) { + merged.push(alias.clone()); + } + } + } + } + } + } + groups +} + #[derive(Clone, Copy)] struct LinkSlotsParallel<'a> { batch: &'static str, @@ -911,15 +981,19 @@ fn link_slots_parallel( } = opts; let phase_start = std::time::Instant::now(); + let groups = group_slots_by_dir(slots, layout); let link_work = || { - slots.par_iter().try_for_each(|slot| { + groups.par_iter().try_for_each(|group| { + let slot = group.representative; let package_id = slot.snapshot_key.pkg_id(); - if let Some(cache_key) = slot.warm_cache_key { - emit_warm_snapshot_progress::( - &package_id, - requester, - progress_reported.contains(cache_key), - ); + for reported in std::iter::once(slot).chain(group.duplicates.iter().copied()) { + if let Some(cache_key) = reported.warm_cache_key { + emit_warm_snapshot_progress::( + &reported.snapshot_key.pkg_id(), + requester, + progress_reported.contains(cache_key), + ); + } } crate::CreateVirtualDirBySnapshot { @@ -935,7 +1009,7 @@ fn link_slots_parallel( include_optional_dependencies, symlink, skipped, - removed_aliases: slot.removed_aliases, + removed_aliases: group.removed_aliases(), needs_build_marker_source: slot.needs_build_marker_source, #[cfg(test)] link_concurrency_probe, @@ -966,6 +1040,7 @@ fn link_slots_parallel( phase = "link_slots", batch, slots = slots.len(), + unique_dirs = groups.len(), elapsed_ms = phase_start.elapsed().as_millis() as u64, "phase complete", ); diff --git a/pnpm/crates/deps-restorer/src/create_virtual_store/tests.rs b/pnpm/crates/deps-restorer/src/create_virtual_store/tests.rs index aa523fdc8e..8571220ee4 100644 --- a/pnpm/crates/deps-restorer/src/create_virtual_store/tests.rs +++ b/pnpm/crates/deps-restorer/src/create_virtual_store/tests.rs @@ -212,6 +212,129 @@ async fn cold_batch_links_slots_in_parallel() { const DUMMY_SHA512: &str = "sha512-AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA=="; +/// Under the global virtual store, peer variants hashing to one slot +/// directory must produce one link task through the whole +/// [`CreateVirtualStore::run`] pass — the probe's lifetime counter +/// distinguishes a real dedup from duplicates that happened to +/// serialize. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn gvs_link_pass_materializes_shared_slot_once() { + use crate::{AllowBuildPolicy, SkippedSnapshots, VirtualStoreLayout}; + use pacquet_config::{Config, NodeLinker, PackageImportMethod}; + use pacquet_store_dir::StoreIndexWriter; + use pacquet_tarball::{CacheValue, MemCache, SharedReportedProgressKeys}; + + let root = tempfile::tempdir().expect("create temp dir"); + let workspace_root = root.path().join("workspace"); + fs::create_dir_all(&workspace_root).expect("create workspace root"); + let modules_dir = workspace_root.join("node_modules"); + let store_dir = root.path().join("store"); + + let mut config = Config::new(); + config.registry = "https://registry.test".to_string(); + config.store_dir = store_dir.into(); + config.modules_dir = modules_dir.clone(); + config.virtual_store_dir = modules_dir.join(".pacquet"); + config.enable_global_virtual_store = true; + config.global_virtual_store_dir = root.path().join("links"); + config.package_import_method = PackageImportMethod::Copy; + config.offline = true; + let config = config.leak(); + + let mut snapshots = HashMap::new(); + let mut packages = HashMap::new(); + let mem_cache = Arc::new(MemCache::default()); + for package_name in ["shared", "solo"] { + let source_dir = workspace_root.join("prefetched").join(package_name); + fs::create_dir_all(&source_dir).expect("create prefetched package dir"); + let manifest_path = source_dir.join("package.json"); + fs::write(&manifest_path, format!(r#"{{"name":"{package_name}","version":"1.0.0"}}"#)) + .expect("write package manifest"); + let cas_paths = HashMap::from([("package.json".to_string(), manifest_path)]); + mem_cache.insert( + format!("https://registry.test/{package_name}/-/{package_name}-1.0.0.tgz"), + Arc::new(tokio::sync::RwLock::new(CacheValue::Available(Arc::new(cas_paths)))), + ); + packages.insert( + key(package_name, "1.0.0").without_peer(), + metadata_with_integrity(DUMMY_SHA512), + ); + } + // Two variants of `shared`, one bare and one peer-suffixed, with + // identical dependency sets — the collision the fix is about. + snapshots.insert(key("shared", "1.0.0"), SnapshotEntry::default()); + snapshots.insert(key("shared", "1.0.0(peer@1.0.0)"), SnapshotEntry::default()); + snapshots.insert(key("solo", "1.0.0"), SnapshotEntry::default()); + + let allow_build_policy = AllowBuildPolicy::default(); + let layout = VirtualStoreLayout::new( + config, + Some("linux-x64-node22"), + Some(&snapshots), + Some(&packages), + Some(&allow_build_policy), + None, + ); + assert_eq!( + layout.slot_dir(&key("shared", "1.0.0")), + layout.slot_dir(&key("shared", "1.0.0(peer@1.0.0)")), + "precondition: the variants must share one slot", + ); + assert_ne!( + layout.slot_dir(&key("shared", "1.0.0")), + layout.slot_dir(&key("solo", "1.0.0")), + "precondition: distinct packages must not share a slot", + ); + + let skipped = SkippedSnapshots::new(); + let logged_methods = AtomicU8::new(0); + let progress_reported = SharedReportedProgressKeys::default(); + let (store_index_writer, writer_task) = StoreIndexWriter::spawn(&config.store_dir); + let requester = workspace_root.to_string_lossy().into_owned(); + let probe = crate::create_virtual_dir_by_snapshot::tests::LinkConcurrencyProbe::default(); + + CreateVirtualStore { + http_client: &pacquet_network::ThrottledClient::default(), + config, + packages: Some(&packages), + snapshots: Some(&snapshots), + current_snapshots: None, + current_packages: None, + layout: &layout, + logged_methods: &logged_methods, + requester: &requester, + store_index_writer: &store_index_writer, + allow_build_policy: &allow_build_policy, + skipped: &skipped, + include_optional_dependencies: true, + supported_architectures: None, + workspace_root: &workspace_root, + node_linker: NodeLinker::Isolated, + progress_reported: &progress_reported, + tarball_mem_cache: Some(&mem_cache), + custom_fetcher_picker: None, + link_concurrency_probe: Some(&probe), + } + .run::() + .await + .expect("global-virtual-store creation should succeed from the mem cache"); + + drop(store_index_writer); + writer_task.await.expect("join store-index writer").expect("flush store-index writer"); + + assert_eq!( + probe.total_entered(), + 2, + "three snapshots over two unique slot directories must run exactly two link tasks", + ); + let shared_manifest = layout + .slot_dir(&key("shared", "1.0.0")) + .join("node_modules") + .join("shared") + .join("package.json"); + assert!(shared_manifest.is_file(), "the shared slot must be materialized: {shared_manifest:?}"); +} + /// `emit_warm_snapshot_progress` fires `resolved` then /// `found_in_store` when no earlier fetch path already emitted the /// package status. Both events carry the same identifiers — pnpm's @@ -491,3 +614,175 @@ fn tarball_metadata_without_integrity() -> PackageMetadata { peer_dependencies_meta: None, } } + +/// Helpers for the `group_slots_by_dir` tests: a GVS layout over the +/// given snapshots/metadata, scoped to a lockfile dir so directory +/// resolutions hash the way a real project's do. +fn gvs_layout( + snapshots: &HashMap, + packages: &HashMap, + lockfile_dir: &std::path::Path, +) -> crate::VirtualStoreLayout { + let mut config = pacquet_config::Config::new(); + config.enable_global_virtual_store = true; + config.virtual_store_dir = std::path::PathBuf::from("/tmp/proj/node_modules/.pnpm"); + config.global_virtual_store_dir = std::path::PathBuf::from("/tmp/store/links"); + let config = config.leak(); + crate::VirtualStoreLayout::new( + config, + Some("linux-x64-node22"), + Some(snapshots), + Some(packages), + None, + Some(lockfile_dir), + ) +} + +fn directory_metadata(directory: &str) -> PackageMetadata { + PackageMetadata { + resolution: LockfileResolution::Directory(pacquet_lockfile::DirectoryResolution { + directory: directory.to_string(), + }), + version: Some("1.0.0".to_string()), + engines: None, + cpu: None, + os: None, + libc: None, + deprecated: None, + has_bin: None, + prepare: None, + bundled_dependencies: None, + peer_dependencies: None, + peer_dependencies_meta: None, + } +} + +fn slot_link<'a>( + snapshot_key: &'a PackageKey, + snapshot: &'a SnapshotEntry, + cas_paths: &'a HashMap, + removed_aliases: &'a [PkgName], +) -> super::SlotLink<'a> { + super::SlotLink { + snapshot_key, + snapshot, + cas_paths, + warm_cache_key: None, + source_is_mutable: true, + needs_build_marker_source: None, + removed_aliases, + } +} + +/// Hash-equal peer variants of a directory dependency collapse into +/// one group, with the obsolete-alias cleanup covering both variants' +/// recorded removals. +#[test] +fn group_slots_by_dir_collapses_hash_equal_peer_variants() { + let plain: PackageKey = "comp@file:packages/comp".parse().expect("parse plain key"); + let peered: PackageKey = + "comp@file:packages/comp(peer@1.0.0)".parse().expect("parse peered key"); + + let mut snapshots = HashMap::new(); + snapshots.insert(plain.clone(), SnapshotEntry::default()); + snapshots.insert(peered.clone(), SnapshotEntry::default()); + let mut packages = HashMap::new(); + packages.insert(plain.clone(), directory_metadata("packages/comp")); + + let layout = gvs_layout(&snapshots, &packages, std::path::Path::new("/home/user/project")); + assert_eq!( + layout.slot_dir(&plain), + layout.slot_dir(&peered), + "precondition: hash-equal variants must resolve to one slot", + ); + + let snapshot = SnapshotEntry::default(); + let cas_paths = HashMap::new(); + let removed_plain = [name("dropped-a")]; + let removed_peered = [name("dropped-a"), name("dropped-b")]; + let slots = [ + slot_link(&plain, &snapshot, &cas_paths, &removed_plain), + slot_link(&peered, &snapshot, &cas_paths, &removed_peered), + ]; + + let groups = super::group_slots_by_dir(&slots, &layout); + + assert_eq!(groups.len(), 1, "hash-equal variants must share one link task"); + assert_eq!(groups[0].duplicates.len(), 1); + let mut merged: Vec = + groups[0].removed_aliases().iter().map(PkgName::to_string).collect(); + merged.sort(); + assert_eq!( + merged, + vec!["dropped-a".to_string(), "dropped-b".to_string()], + "cleanup must cover aliases recorded against either variant", + ); +} + +/// Peer variants whose subtrees differ hash to different slots and must +/// keep their own link tasks. +#[test] +fn group_slots_by_dir_keeps_diverging_peer_variants_apart() { + let plain: PackageKey = "comp@file:packages/comp".parse().expect("parse plain key"); + let peered: PackageKey = + "comp@file:packages/comp(peer@1.0.0)".parse().expect("parse peered key"); + let child: PackageKey = key("leaf", "1.0.0"); + + let mut snapshots = HashMap::new(); + snapshots.insert(plain.clone(), SnapshotEntry::default()); + // The peered variant resolves an extra child, so its recursive + // hash — and therefore its slot — must diverge from the plain one. + snapshots.insert(peered.clone(), snapshot_with_dep("leaf", "1.0.0")); + snapshots.insert(child.clone(), SnapshotEntry::default()); + let mut packages = HashMap::new(); + packages.insert(plain.clone(), directory_metadata("packages/comp")); + packages.insert(child, metadata_with_integrity(DUMMY_SHA512)); + + let layout = gvs_layout(&snapshots, &packages, std::path::Path::new("/home/user/project")); + assert_ne!( + layout.slot_dir(&plain), + layout.slot_dir(&peered), + "precondition: diverging subtrees must resolve to distinct slots", + ); + + let snapshot = SnapshotEntry::default(); + let cas_paths = HashMap::new(); + let slots = [ + slot_link(&plain, &snapshot, &cas_paths, &[]), + slot_link(&peered, &snapshot, &cas_paths, &[]), + ]; + + let groups = super::group_slots_by_dir(&slots, &layout); + + assert_eq!(groups.len(), 2, "distinct slots must keep distinct link tasks"); + assert!(groups.iter().all(|group| group.duplicates.is_empty())); + assert!(groups.iter().all(|group| group.merged_removed_aliases.is_none())); +} + +/// Without the global virtual store, `slot_dir` embeds the full +/// peer-suffixed key, so grouping never merges anything and the link +/// pass matches the ungrouped behavior exactly. +#[test] +fn group_slots_by_dir_is_identity_without_gvs() { + let plain: PackageKey = "comp@file:packages/comp".parse().expect("parse plain key"); + let peered: PackageKey = + "comp@file:packages/comp(peer@1.0.0)".parse().expect("parse peered key"); + + let mut config = pacquet_config::Config::new(); + config.enable_global_virtual_store = false; + config.virtual_store_dir = std::path::PathBuf::from("/tmp/proj/node_modules/.pnpm"); + let config = config.leak(); + let layout = crate::VirtualStoreLayout::new(config, None, None, None, None, None); + + let snapshot = SnapshotEntry::default(); + let cas_paths = HashMap::new(); + let slots = [ + slot_link(&plain, &snapshot, &cas_paths, &[]), + slot_link(&peered, &snapshot, &cas_paths, &[]), + ]; + + let groups = super::group_slots_by_dir(&slots, &layout); + + assert_eq!(groups.len(), 2, "non-GVS slots are unique per key; nothing may merge"); + assert!(groups.iter().all(|group| group.duplicates.is_empty())); +}