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.
This commit is contained in:
Zoltan Kochan authored and GitHub committed 2026-08-06 09:55:17 +02:00
1 parent b0e819458a
commit 74577d73fa
4 files changed
+393 -8

No files matched your search

@@ -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.
@@ -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(()),
@@ -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<Vec<PkgName>>,
}
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<SlotDirGroup<'a>> {
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<PathBuf, usize> = HashMap::with_capacity(slots.len());
let mut groups: Vec<SlotDirGroup<'a>> = 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<Reporter: self::Reporter>(
} = 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::<Reporter>(
&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::<Reporter>(
&reported.snapshot_key.pkg_id(),
requester,
progress_reported.contains(cache_key),
);
}
}
crate::CreateVirtualDirBySnapshot {
@@ -935,7 +1009,7 @@ fn link_slots_parallel<Reporter: self::Reporter>(
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<Reporter: self::Reporter>(
phase = "link_slots",
batch,
slots = slots.len(),
unique_dirs = groups.len(),
elapsed_ms = phase_start.elapsed().as_millis() as u64,
"phase complete",
);
@@ -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::<SilentReporter>()
.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<PackageKey, SnapshotEntry>,
packages: &HashMap<PackageKey, PackageMetadata>,
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<String, std::path::PathBuf>,
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<String> =
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()));
}