fix(pacquet): make the global virtual store safe to share, and reachable from the Node API (#13648)

The global virtual store is shared across projects, so concurrent installs may materialize the same content-addressed slot. The existing swap removed an occupied target before moving its staged copy into place. This could race with another writer, fail with a directory-not-empty error, or briefly remove a directory another process was reading.

Add a safe-to-skip import mode for immutable global virtual-store slots. It first attempts a non-destructive rename. If another importer already completed the target, discard only the staging directory. Incomplete targets still use the repair path. Mutable directory sources do not use this mode because a complete slot may still contain stale files, and the transient needs-build marker is excluded from completeness checks so completed build output is preserved.

Expose enableGlobalVirtualStore, globalVirtualStoreDir, packageExtensions, and patchedDependencies through @pnpm/napi. Resolve relative patch paths from the requested install directory, recompute global virtual-store derivation after applying the overlay, and return both configured and effective virtual-store paths from readConfig.
This commit is contained in:
Zoltan Kochan authored and GitHub committed 2026-08-05 02:01:06 +02:00
1 parent a540b24e99
commit a550eca68a
14 files changed
+380 -9

No files matched your search

+5
View File
@@ -0,0 +1,5 @@
---
"pacquet": patch
---
Concurrent installs sharing a global virtual store no longer fail with `failed to remove existing directory ... prior to swap: Directory not empty`, and no longer briefly remove a package directory another process is reading.
+5
View File
@@ -0,0 +1,5 @@
---
"@pnpm/napi": minor
---
`install` accepts `enableGlobalVirtualStore`, `globalVirtualStoreDir`, `packageExtensions` and `patchedDependencies`, and `readConfig` reports `enableGlobalVirtualStore`, `globalVirtualStoreDir`, `virtualStoreDir` and `effectiveVirtualStoreDir`. Hosts embedding the engine can now use the global virtual store, declare dependencies a package failed to declare, patch a package, and locate the virtual store instead of assuming `node_modules/.pnpm`.
+1 -1
View File
@@ -741,7 +741,7 @@ fn copy_project<ReporterT: Reporter>(
PackageImportMethod::CloneOrCopy,
dest,
&output.files_map,
ImportIndexedDirOpts { force: true, keep_modules_dir: false },
ImportIndexedDirOpts { force: true, ..ImportIndexedDirOpts::default() },
)
.map_err(miette::Report::new)
.wrap_err("copy project files")
@@ -1559,7 +1559,11 @@ fn materialize_side_effects<Reporter: self::Reporter>(
import_method,
pkg_dir,
overlay,
ImportIndexedDirOpts { force: true, keep_modules_dir: true },
ImportIndexedDirOpts {
force: true,
keep_modules_dir: true,
..ImportIndexedDirOpts::default()
},
)
.map_err(BuildModulesError::MaterializeSideEffects)
}
@@ -173,10 +173,12 @@ impl CreateVirtualDirBySnapshot<'_> {
} else {
cas_paths
};
// Mutable sources can reuse a slot for different contents, so a complete import may be stale.
let safe_to_skip = layout.enable_global_virtual_store() && !source_is_mutable;
let import_opts = if interrupted_build || source_is_mutable {
ImportIndexedDirOpts { force: true, keep_modules_dir: true }
ImportIndexedDirOpts { force: true, keep_modules_dir: true, safe_to_skip }
} else {
ImportIndexedDirOpts::default()
ImportIndexedDirOpts { safe_to_skip, ..ImportIndexedDirOpts::default() }
};
let import_package = || {
@@ -37,6 +37,10 @@ pub struct ImportIndexedDirOpts {
/// installed by a sibling pass must not be clobbered when the
/// parent package is re-imported.
pub keep_modules_dir: bool,
/// Whether an occupied, complete target is equivalent to this import.
///
/// Callers must ensure that the target path uniquely identifies its contents.
pub safe_to_skip: bool,
}
/// Error type for [`import_indexed_dir`].
@@ -170,6 +174,7 @@ pub fn import_indexed_dir<Reporter: self::Reporter>(
dir_path,
cas_paths,
opts.keep_modules_dir,
opts.safe_to_skip,
)
.inspect(|()| unquarantine()),
}
@@ -261,6 +266,26 @@ fn marker_file(cas_paths: &HashMap<String, PathBuf>) -> Option<&str> {
cas_paths.keys().map(String::as_str).filter(|path| *path != crate::NEEDS_BUILD_MARKER).min()
}
// Supported platforms report an occupied directory with different error kinds.
fn is_occupied_target(error: &io::Error) -> bool {
matches!(
error.kind(),
io::ErrorKind::DirectoryNotEmpty
| io::ErrorKind::AlreadyExists
| io::ErrorKind::PermissionDenied,
)
}
// The completion marker is placed last. Verify every package file in case a completed import was
// later damaged. The needs-build marker is transient and does not identify package contents.
fn is_complete_import(dir_path: &Path, cas_paths: &HashMap<String, PathBuf>) -> bool {
marker_present(dir_path, cas_paths)
&& cas_paths
.keys()
.filter(|entry| entry.as_str() != crate::NEEDS_BUILD_MARKER)
.all(|entry| dir_path.join(entry).exists())
}
/// Whether `dir_path` holds the completion marker. An empty map has no
/// marker, so it counts as present — there is nothing to import.
#[must_use]
@@ -310,6 +335,7 @@ fn stage_and_swap<Reporter: self::Reporter>(
dir_path: &Path,
cas_paths: &HashMap<String, PathBuf>,
keep_modules_dir: bool,
safe_to_skip: bool,
) -> Result<(), ImportIndexedDirError> {
let stage = pick_stage_path(dir_path);
let target_modules = dir_path.join("node_modules");
@@ -370,6 +396,20 @@ fn stage_and_swap<Reporter: self::Reporter>(
Some(_) | None => false,
};
// 3b. Keep a complete content-addressed target instead of removing a directory that another
// importer may be populating or reading. When step 3 moved `node_modules/`, continue with
// the swap so that it can be restored.
if safe_to_skip && !nm_moved {
match fs::rename(&stage, dir_path) {
Ok(()) => return Ok(()),
Err(error) if is_occupied_target(&error) && is_complete_import(dir_path, cas_paths) => {
let _ = fs::remove_dir_all(&stage);
return Ok(());
}
Err(_) => {}
}
}
// 4. Remove the old contents. If this fails after step 3, the
// staged copy of `node_modules/` is the user's only copy —
// try to move it back into place before bailing, and leak
@@ -24,9 +24,11 @@ fn cas_map(entries: &[(&str, PathBuf)]) -> HashMap<String, PathBuf> {
}
const FORCE_KEEP: ImportIndexedDirOpts =
ImportIndexedDirOpts { force: true, keep_modules_dir: true };
ImportIndexedDirOpts { force: true, keep_modules_dir: true, safe_to_skip: false };
const FORCE_ONLY: ImportIndexedDirOpts =
ImportIndexedDirOpts { force: true, keep_modules_dir: false };
ImportIndexedDirOpts { force: true, keep_modules_dir: false, safe_to_skip: false };
const FORCE_SHARED: ImportIndexedDirOpts =
ImportIndexedDirOpts { force: true, keep_modules_dir: false, safe_to_skip: true };
#[test]
fn fresh_target_links_files() {
@@ -679,3 +681,179 @@ fn marker_only_map_creates_target_and_places_marker() {
);
}
}
#[test]
fn safe_to_skip_keeps_a_target_a_concurrent_importer_already_completed() {
let tmp = tempdir().unwrap();
let src_root = tmp.path().join("cas");
fs::create_dir_all(&src_root).unwrap();
let pkg_json = write_source(&src_root, "package.json", b"{\"version\":\"1.0.0\"}");
let index = write_source(&src_root, "index.js", b"module.exports = 1");
let cas = cas_map(&[("package.json", pkg_json), ("index.js", index)]);
let target = tmp.path().join("slot");
fs::create_dir_all(&target).unwrap();
fs::write(target.join("package.json"), b"{\"version\":\"1.0.0\"}").unwrap();
fs::write(target.join("index.js"), b"module.exports = 1").unwrap();
import_indexed_dir::<SilentReporter>(
&AtomicU8::new(0),
PackageImportMethod::Copy,
&target,
&cas,
FORCE_SHARED,
)
.expect("a slot another importer already completed is not a conflict");
assert_eq!(fs::read(target.join("package.json")).unwrap(), b"{\"version\":\"1.0.0\"}");
assert_eq!(fs::read(target.join("index.js")).unwrap(), b"module.exports = 1");
let strays: Vec<_> = fs::read_dir(tmp.path())
.unwrap()
.map(|entry| entry.unwrap().file_name().to_string_lossy().into_owned())
.filter(|name| name != "cas" && name != "slot")
.collect();
assert_eq!(strays, Vec::<String>::new(), "staging dir must be cleaned up");
}
#[test]
fn safe_to_skip_does_not_accept_a_slot_that_is_still_being_written() {
let tmp = tempdir().unwrap();
let src_root = tmp.path().join("cas");
fs::create_dir_all(&src_root).unwrap();
let pkg_json = write_source(&src_root, "package.json", b"{\"version\":\"1.0.0\"}");
let index = write_source(&src_root, "index.js", b"module.exports = 1");
let cas = cas_map(&[("package.json", pkg_json), ("index.js", index)]);
let target = tmp.path().join("slot");
fs::create_dir_all(&target).unwrap();
fs::write(target.join("index.js"), b"half-written").unwrap();
import_indexed_dir::<SilentReporter>(
&AtomicU8::new(0),
PackageImportMethod::Copy,
&target,
&cas,
FORCE_SHARED,
)
.expect("an unfinished slot must be completed, not accepted");
assert_eq!(fs::read(target.join("package.json")).unwrap(), b"{\"version\":\"1.0.0\"}");
assert_eq!(
fs::read(target.join("index.js")).unwrap(),
b"module.exports = 1",
"the half-written file must be replaced, not kept",
);
}
#[test]
fn safe_to_skip_still_repairs_an_incomplete_target() {
let tmp = tempdir().unwrap();
let src_root = tmp.path().join("cas");
fs::create_dir_all(&src_root).unwrap();
let pkg_json = write_source(&src_root, "package.json", b"{\"version\":\"1.0.0\"}");
let index = write_source(&src_root, "index.js", b"module.exports = 1");
let cas = cas_map(&[("package.json", pkg_json), ("index.js", index)]);
let target = tmp.path().join("slot");
fs::create_dir_all(&target).unwrap();
fs::write(target.join("package.json"), b"truncated").unwrap();
import_indexed_dir::<SilentReporter>(
&AtomicU8::new(0),
PackageImportMethod::Copy,
&target,
&cas,
FORCE_SHARED,
)
.expect("an incomplete slot must be repaired");
assert_eq!(fs::read(target.join("package.json")).unwrap(), b"{\"version\":\"1.0.0\"}");
assert_eq!(fs::read(target.join("index.js")).unwrap(), b"module.exports = 1");
}
// Windows needs retry handling while a competing importer holds file handles.
#[cfg(unix)]
#[test]
fn concurrent_importers_of_one_shared_slot_both_succeed() {
use std::sync::{Arc, Barrier};
let tmp = tempdir().unwrap();
let src_root = tmp.path().join("cas");
fs::create_dir_all(&src_root).unwrap();
let pkg_json = write_source(&src_root, "package.json", b"{\"version\":\"1.0.0\"}");
let index = write_source(&src_root, "index.js", b"module.exports = 1");
let cas = Arc::new(cas_map(&[("package.json", pkg_json), ("index.js", index)]));
let target = Arc::new(tmp.path().join("slot"));
let barrier = Arc::new(Barrier::new(2));
let handles: Vec<_> = (0..2)
.map(|_| {
let cas = Arc::clone(&cas);
let target = Arc::clone(&target);
let barrier = Arc::clone(&barrier);
std::thread::spawn(move || {
barrier.wait();
import_indexed_dir::<SilentReporter>(
&AtomicU8::new(0),
PackageImportMethod::Copy,
&target,
&cas,
FORCE_SHARED,
)
})
})
.collect();
for handle in handles {
handle.join().expect("importer thread panicked").expect("both importers must succeed");
}
assert_eq!(fs::read(target.join("package.json")).unwrap(), b"{\"version\":\"1.0.0\"}");
assert_eq!(fs::read(target.join("index.js")).unwrap(), b"module.exports = 1");
let strays: Vec<String> = fs::read_dir(tmp.path())
.unwrap()
.map(|entry| entry.unwrap().file_name().to_string_lossy().into_owned())
.filter(|name| name != "cas" && name != "slot")
.collect();
assert_eq!(strays, Vec::<String>::new(), "neither importer may leak a staging dir");
}
#[test]
fn safe_to_skip_keeps_a_slot_whose_build_removed_the_needs_build_marker() {
let tmp = tempdir().unwrap();
let src_root = tmp.path().join("cas");
fs::create_dir_all(&src_root).unwrap();
let pkg_json = write_source(&src_root, "package.json", b"{\"version\":\"1.0.0\"}");
let index = write_source(&src_root, "index.js", b"module.exports = 1");
let needs_build = write_source(&src_root, "needs-build", b"");
let cas = cas_map(&[
("package.json", pkg_json),
("index.js", index),
(crate::NEEDS_BUILD_MARKER, needs_build),
]);
let target = tmp.path().join("slot");
fs::create_dir_all(&target).unwrap();
fs::write(target.join("package.json"), b"{\"version\":\"1.0.0\"}").unwrap();
fs::write(target.join("index.js"), b"module.exports = 1").unwrap();
fs::write(target.join("built.node"), b"native addon").unwrap();
import_indexed_dir::<SilentReporter>(
&AtomicU8::new(0),
PackageImportMethod::Copy,
&target,
&cas,
FORCE_SHARED,
)
.expect("a built slot must be left alone");
assert!(
target.join("built.node").exists(),
"the build output must survive; the slot was rebuilt from the staging copy",
);
assert!(
!target.join(crate::NEEDS_BUILD_MARKER).exists(),
"the consumed needs-build marker must not be put back",
);
}
@@ -289,7 +289,11 @@ fn import_node<Reporter: self::Reporter>(
opts.import_method,
&node.dir,
cas_paths,
ImportIndexedDirOpts { force: true, keep_modules_dir: true },
ImportIndexedDirOpts {
force: true,
keep_modules_dir: true,
..ImportIndexedDirOpts::default()
},
)
.map_err(LinkHoistedModulesError::ImportIndexedDir)
}
+35 -1
View File
@@ -35,7 +35,7 @@ use dashmap::DashMap;
use indexmap::IndexMap;
use pacquet_config::{
Config, GetHomeDir, Host, LinkWorkspacePackages, LoadWorkspaceYamlError, NodeLinker,
PackageImportMethod, default_registry,
PackageExtension, PackageImportMethod, default_registry,
};
use pacquet_network::{AuthHeaders, ProxyConfig, TlsConfig, nerf_dart, normalize_auth_key};
use pacquet_store_dir::StoreDir;
@@ -59,6 +59,10 @@ pub struct ConfigOverlay {
pub link_workspace_packages: Option<LinkWorkspacePackages>,
pub package_import_method: Option<PackageImportMethod>,
pub virtual_store_dir_max_length: Option<u64>,
pub enable_global_virtual_store: Option<bool>,
pub global_virtual_store_dir: Option<PathBuf>,
pub package_extensions: Option<IndexMap<String, PackageExtension>>,
pub patched_dependencies: Option<IndexMap<String, String>>,
pub hoist_pattern: Option<Vec<String>>,
pub public_hoist_pattern: Option<Vec<String>>,
pub external_dependencies: Option<BTreeSet<String>>,
@@ -280,6 +284,24 @@ fn build_config(dir: &Path, overlay: &ConfigOverlay) -> Result<Config, LoadWorks
if let Some(max_length) = overlay.virtual_store_dir_max_length {
config.virtual_store_dir_max_length = max_length;
}
if let Some(value) = overlay.enable_global_virtual_store {
config.enable_global_virtual_store = value;
}
if let Some(package_extensions) = &overlay.package_extensions {
config.package_extensions = Some(package_extensions.clone());
}
if let Some(patched_dependencies) = &overlay.patched_dependencies {
// Embedded installs resolve relative patch paths from `dir`, even without a workspace file.
config.patched_dependencies = Some(
patched_dependencies
.iter()
.map(|(key, path)| (key.clone(), dir.join(path).display().to_string()))
.collect(),
);
if config.workspace_dir.is_none() {
config.workspace_dir = Some(dir.to_path_buf());
}
}
if let Some(hoist_pattern) = &overlay.hoist_pattern {
config.hoist_pattern = Some(hoist_pattern.clone());
}
@@ -403,6 +425,18 @@ fn build_config(dir: &Path, overlay: &ConfigOverlay) -> Result<Config, LoadWorks
&overlay_default_registry(overlay),
)));
}
// Overlay fields may invalidate the path derived by `Config::current`.
if let Some(global_virtual_store_dir) = &overlay.global_virtual_store_dir {
config.global_virtual_store_dir.clone_from(global_virtual_store_dir);
} else if overlay.enable_global_virtual_store.is_some() || overlay.store_dir.is_some() {
let virtual_store_dir_explicit = config.explicit_settings.contains_key("virtualStoreDir");
let global_virtual_store_dir_explicit =
config.explicit_settings.contains_key("globalVirtualStoreDir");
config.apply_global_virtual_store_derivation(
virtual_store_dir_explicit,
global_virtual_store_dir_explicit,
);
}
Ok(config)
}
+59
View File
@@ -90,6 +90,14 @@ pub struct InstallOptions {
pub prefer_offline: Option<bool>,
pub offline: Option<bool>,
pub virtual_store_dir_max_length: Option<u32>,
/// Whether to use the shared global virtual store for dependency slots.
pub enable_global_virtual_store: Option<bool>,
/// Overrides the global virtual store directory.
pub global_virtual_store_dir: Option<String>,
/// Manifest fields to add to packages selected by name or version range.
pub package_extensions: Option<IndexMap<String, PackageExtensionInput>>,
/// Patch paths keyed by package selector. Relative paths resolve from `dir`.
pub patched_dependencies: Option<IndexMap<String, String>>,
pub peers_suffix_max_length: Option<u32>,
pub dedupe_peer_dependents: Option<bool>,
pub dedupe_peers: Option<bool>,
@@ -171,6 +179,21 @@ pub struct NetworkConfigInput {
pub user_agent: Option<String>,
}
/// Manifest fields to add to a matching package.
#[napi(object)]
pub struct PackageExtensionInput {
pub dependencies: Option<HashMap<String, String>>,
pub optional_dependencies: Option<HashMap<String, String>>,
pub peer_dependencies: Option<HashMap<String, String>>,
pub peer_dependencies_meta: Option<HashMap<String, PeerDependencyMetaInput>>,
}
/// Metadata for a peer dependency.
#[napi(object)]
pub struct PeerDependencyMetaInput {
pub optional: Option<bool>,
}
/// `peerDependencyRules` input. Mirrors `PeerDependencyRules` in `index.d.ts`.
#[napi(object)]
pub struct PeerDependencyRulesInput {
@@ -548,6 +571,42 @@ fn build_overlay(options: &InstallOptions) -> napi::Result<ConfigOverlay> {
.as_deref()
.and_then(parse_import_method),
virtual_store_dir_max_length: options.virtual_store_dir_max_length.map(u64::from),
enable_global_virtual_store: options.enable_global_virtual_store,
global_virtual_store_dir: options.global_virtual_store_dir.as_ref().map(PathBuf::from),
package_extensions: options.package_extensions.as_ref().map(|extensions| {
extensions
.iter()
.map(|(selector, extension)| {
let to_sorted = |map: &Option<HashMap<String, String>>| {
map.as_ref()
.map(|map| map.iter().map(|(k, v)| (k.clone(), v.clone())).collect())
};
(
selector.clone(),
pacquet_config::PackageExtension {
dependencies: to_sorted(&extension.dependencies),
optional_dependencies: to_sorted(&extension.optional_dependencies),
peer_dependencies: to_sorted(&extension.peer_dependencies),
peer_dependencies_meta: extension.peer_dependencies_meta.as_ref().map(
|meta| {
meta.iter()
.map(|(name, entry)| {
(
name.clone(),
pacquet_config::PeerDependencyMeta {
optional: entry.optional,
},
)
})
.collect()
},
),
},
)
})
.collect()
}),
patched_dependencies: options.patched_dependencies.clone(),
hoist_pattern: options.hoist_pattern.clone(),
public_hoist_pattern: options.public_hoist_pattern.clone(),
external_dependencies: options
+4
View File
@@ -531,6 +531,10 @@ fn install_options() -> InstallOptions {
prefer_offline: None,
offline: None,
virtual_store_dir_max_length: None,
enable_global_virtual_store: None,
global_virtual_store_dir: None,
package_extensions: None,
patched_dependencies: None,
peers_suffix_max_length: None,
dedupe_peer_dependents: None,
dedupe_peers: None,
+12
View File
@@ -57,6 +57,14 @@ pub struct ResolvedConfig {
pub store_dir: String,
pub cache_dir: String,
pub virtual_store_dir_max_length: u32,
/// Whether the resolved configuration uses the global virtual store.
pub enable_global_virtual_store: bool,
/// Shared virtual-store root.
pub global_virtual_store_dir: String,
/// Project-local virtual-store directory.
pub virtual_store_dir: String,
/// Virtual-store directory used by this configuration and recorded in `.modules.yaml`.
pub effective_virtual_store_dir: String,
pub network_concurrency: u32,
pub max_sockets: Option<u32>,
pub fetch_retries: u32,
@@ -148,6 +156,10 @@ fn project_config(config: &pacquet_config::Config) -> ResolvedConfig {
cache_dir: config.cache_dir.display().to_string(),
virtual_store_dir_max_length: u32::try_from(config.virtual_store_dir_max_length)
.unwrap_or(u32::MAX),
enable_global_virtual_store: config.enable_global_virtual_store,
global_virtual_store_dir: config.global_virtual_store_dir.display().to_string(),
virtual_store_dir: config.virtual_store_dir.display().to_string(),
effective_virtual_store_dir: config.effective_virtual_store_dir().display().to_string(),
network_concurrency: u32::try_from(config.network_concurrency).unwrap_or(u32::MAX),
max_sockets: config.max_sockets.map(|value| u32::try_from(value).unwrap_or(u32::MAX)),
fetch_retries: config.fetch_retries,
+1 -1
View File
@@ -282,7 +282,7 @@ impl WritePackageForPatch<'_> {
PackageImportMethod::CloneOrCopy,
dest,
&cas_paths,
ImportIndexedDirOpts { force: true, keep_modules_dir: false },
ImportIndexedDirOpts { force: true, ..ImportIndexedDirOpts::default() },
)
.map_err(WritePackageForPatchError::ImportIndexedDir)
}
+24
View File
@@ -104,6 +104,14 @@ export interface SharedEngineOptions {
cacheDir?: string
}
/** Manifest fields to add to a matching package. */
export interface PackageExtension {
dependencies?: Record<string, string>
optionalDependencies?: Record<string, string>
peerDependencies?: Record<string, string>
peerDependenciesMeta?: Record<string, { optional?: boolean }>
}
export interface InstallOptions extends SharedEngineOptions {
/** Lockfile / workspace root directory. */
dir: string
@@ -132,6 +140,14 @@ export interface InstallOptions extends SharedEngineOptions {
preferOffline?: boolean
offline?: boolean
virtualStoreDirMaxLength?: number
/** Whether to use the shared global virtual store for dependency slots. */
enableGlobalVirtualStore?: boolean
/** Overrides the global virtual store directory. */
globalVirtualStoreDir?: string
/** Manifest fields to add to packages selected by name or version range. */
packageExtensions?: Record<string, PackageExtension>
/** Patch paths keyed by package selector. Relative paths resolve from `dir`. */
patchedDependencies?: Record<string, string>
peersSuffixMaxLength?: number
dedupePeerDependents?: boolean
/**
@@ -386,6 +402,14 @@ export interface ResolvedConfig {
storeDir: string
cacheDir: string
virtualStoreDirMaxLength: number
/** Whether the resolved configuration uses the global virtual store. */
enableGlobalVirtualStore: boolean
/** Shared virtual-store root. */
globalVirtualStoreDir: string
/** Project-local virtual-store directory. */
virtualStoreDir: string
/** Virtual-store directory used by this configuration and recorded in `.modules.yaml`. */
effectiveVirtualStoreDir: string
networkConcurrency: number
maxSockets?: number
fetchRetries: number