diff --git a/pacquet/crates/package-manager/src/create_virtual_store.rs b/pacquet/crates/package-manager/src/create_virtual_store.rs index 10341cc126..63630f150a 100644 --- a/pacquet/crates/package-manager/src/create_virtual_store.rs +++ b/pacquet/crates/package-manager/src/create_virtual_store.rs @@ -753,12 +753,15 @@ impl<'a> CreateVirtualStore<'a> { // future returns; under hoisted no slot was written and the // CAS index is the only output. let mut fetch_failed: HashSet = HashSet::new(); - let mut cold_cas_paths: Vec<(&PackageKey, HashMap)> = Vec::new(); + let mut cold_cas_paths: Vec<(&PackageKey, &SnapshotEntry, HashMap)> = + Vec::new(); if !cold.is_empty() { let prefetched_ref = Some(&prefetched); let verified_files_cache_ref = &verified_files_cache; - type ColdOutcome<'a> = - (Option, Option<(&'a PackageKey, HashMap)>); + type ColdOutcome<'a> = ( + Option, + Option<(&'a PackageKey, &'a SnapshotEntry, HashMap)>, + ); let outcomes: Vec> = cold .iter() .map(|(snapshot_key, snapshot)| async move { @@ -788,14 +791,15 @@ impl<'a> CreateVirtualStore<'a> { skipped, workspace_root, node_linker, + // The slot link is deferred to the parallel pass + // below so it doesn't serialize inside this + // cooperative `try_join_all` task. + defer_link: true, } .run::() .await; match result { - Ok(cas_paths) => { - let captured = is_hoisted.then_some((*snapshot_key, cas_paths)); - Ok((None, captured)) - } + Ok(cas_paths) => Ok((None, Some((*snapshot_key, *snapshot, cas_paths)))), Err(err) if snapshot.optional && is_fetch_side_failure(&err) => { // Silent swallow, matching upstream. `tracing::warn!` // gives operator visibility without polluting @@ -839,6 +843,55 @@ impl<'a> CreateVirtualStore<'a> { } } + // Cold link pass (isolated only): now that every cold snapshot's + // tarball is in the store, link each into its virtual-store slot + // in one parallel rayon pass — the same shape as the warm batch + // above. The per-snapshot download futures deferred this work + // (`defer_link: true`) so the blocking `rayon::join` link inside + // each wouldn't serialize one-at-a-time within the cooperative + // `try_join_all` task; doing it here lets every slot link + // concurrently. Hoisted writes no slots, so it skips this and + // consumes `cold_cas_paths` for the per-pkg CAS index below. + if !is_hoisted && !cold_cas_paths.is_empty() { + use rayon::prelude::*; + let import_method = config.package_import_method; + let link_work = || { + cold_cas_paths.par_iter().try_for_each(|(snapshot_key, snapshot, cas_paths)| { + let package_id = snapshot_key.without_peer().to_string(); + crate::CreateVirtualDirBySnapshot { + layout, + cas_paths, + import_method, + logged_methods, + requester, + package_id: &package_id, + package_key: snapshot_key, + snapshot, + skipped, + } + .run::() + .map_err(|error| { + CreateVirtualStoreError::InstallPackageBySnapshot( + InstallPackageBySnapshotError::CreateVirtualDir(error), + ) + }) + }) + }; + // `block_in_place` (the same guard the warm batch uses) + // migrates other futures off this worker so async progress + // continues; it panics on the `current_thread` runtime that + // `#[tokio::test]` defaults to, so fall back to a plain call + // there. + let on_multi_thread = tokio::runtime::Handle::try_current().is_ok_and(|handle| { + handle.runtime_flavor() == tokio::runtime::RuntimeFlavor::MultiThread + }); + if on_multi_thread { + tokio::task::block_in_place(link_work)?; + } else { + link_work()?; + } + } + // Build the per-pkg CAS index when the install is targeting // the hoisted linker. Upstream's // [`lockfileToHoistedDepGraph`](https://github.com/pnpm/pnpm/blob/94240bc046/installing/deps-restorer/src/lockfileToHoistedDepGraph.ts) @@ -870,7 +923,7 @@ impl<'a> CreateVirtualStore<'a> { // real install. if let Some(map) = cas_paths_by_pkg_id.as_mut() { map.reserve(cold_cas_paths.len()); - for (snapshot_key, paths) in cold_cas_paths { + for (snapshot_key, _snapshot, paths) in cold_cas_paths { // Mirrors upstream's `getPkgIdWithPatchHash` — strip // the peer-graph suffix but keep `(patch_hash=...)` so // patched packages share one CAS-paths entry across diff --git a/pacquet/crates/package-manager/src/install_package_by_snapshot.rs b/pacquet/crates/package-manager/src/install_package_by_snapshot.rs index dce1b2eac2..0e17d0fc72 100644 --- a/pacquet/crates/package-manager/src/install_package_by_snapshot.rs +++ b/pacquet/crates/package-manager/src/install_package_by_snapshot.rs @@ -114,6 +114,16 @@ pub struct InstallPackageBySnapshot<'a> { /// [`nodeLinker === 'hoisted'`](https://github.com/pnpm/pnpm/blob/94240bc046/installing/deps-restorer/src/index.ts#L411-L425) /// branch in `headlessInstall`. pub node_linker: NodeLinker, + /// When `true`, return the fetched CAS paths without populating the + /// virtual-store slot ([`CreateVirtualDirBySnapshot`]) — the caller + /// links them itself in a separate parallel pass. The cold batch in + /// [`crate::CreateVirtualStore`] sets this so the per-snapshot + /// download futures don't each run a *blocking* `rayon::join` link + /// inside the cooperative `try_join_all` task, which would serialize + /// the links one-at-a-time; instead every slot links concurrently + /// once its tarball is in the store. No effect under + /// [`NodeLinker::Hoisted`], which never writes virtual-store slots. + pub defer_link: bool, } /// Error type of [`InstallPackageBySnapshot`]. @@ -248,6 +258,7 @@ impl<'a> InstallPackageBySnapshot<'a> { skipped, workspace_root, node_linker, + defer_link, } = self; // TODO: skip when already exists in store? @@ -537,7 +548,7 @@ impl<'a> InstallPackageBySnapshot<'a> { // hoisted skips both `linkAllModules` (slot symlinks) and // `linkAllPkgs` (slot file imports), and runs // `linkHoistedModules` over the CAS paths instead. - if matches!(node_linker, NodeLinker::Isolated | NodeLinker::Pnp) { + if !defer_link && matches!(node_linker, NodeLinker::Isolated | NodeLinker::Pnp) { CreateVirtualDirBySnapshot { layout, cas_paths: &cas_paths, diff --git a/pacquet/crates/package-manager/src/install_package_by_snapshot/tests.rs b/pacquet/crates/package-manager/src/install_package_by_snapshot/tests.rs index 7231e8e49b..e95da23c4f 100644 --- a/pacquet/crates/package-manager/src/install_package_by_snapshot/tests.rs +++ b/pacquet/crates/package-manager/src/install_package_by_snapshot/tests.rs @@ -401,6 +401,7 @@ async fn cold_batch_reuses_in_flight_prefetch_from_mem_cache() { // only the download-coordination branch and gets the CAS map // back directly. node_linker: pacquet_config::NodeLinker::Hoisted, + defer_link: false, } .run::() .await @@ -469,6 +470,7 @@ async fn without_mem_cache_skips_coordination_and_downloads() { skipped: &skipped, workspace_root: store_tmp.path(), node_linker: pacquet_config::NodeLinker::Hoisted, + defer_link: false, } .run::() .await @@ -537,6 +539,7 @@ async fn cold_batch_falls_back_when_prefetch_failed() { skipped: &skipped, workspace_root: store_tmp.path(), node_linker: pacquet_config::NodeLinker::Hoisted, + defer_link: false, } .run::() .await