perf(package-manager): parallelize cold-batch slot linking (#12249)

The cold batch in `CreateVirtualStore` ran each per-snapshot
`InstallPackageBySnapshot` as a future inside one cooperative
`try_join_all` task, and each future called the *blocking*
`CreateVirtualDirBySnapshot::run` (a `rayon::join` over CAS import +
symlink layout) directly. Because all the futures share a single task,
a blocking call in one prevents the others from being polled — so the
1308 slot-link operations on a fresh install serialized one-at-a-time,
each paying its own `rayon::join` dispatch. The warm batch already
avoids this by linking every slot in a single `rayon` `par_iter`.

This was invisible on warm/restore installs (everything goes through
the warm batch) but dominated the cold path: a pnpr client install,
whose fast offloaded resolution routes every package through the cold
batch, spent ~14.5s in `CreateVirtualStore` where an equivalent local
fresh install — which pre-fills the store during its slower resolution
and so hits the warm batch — spent ~5-7s linking.

Defer the slot link out of the per-snapshot download future (new
`InstallPackageBySnapshot::defer_link`) and run all the cold links in
one parallel `rayon` pass after the downloads, mirroring the warm
batch. On a fresh ~1300-package install through a pnpr server this cut
`CreateVirtualStore` from ~14.5s to ~9.5s and the end-to-end install
from ~18.5s to ~13s — now faster than the equivalent non-pnpr install
(~15.5s), which it should be since the pnpr client offloads resolution
and downloads less metadata. The progress events are unchanged in kind
and order (`fetched`/`found_in_store` still fire during download,
`imported` still fires once the slot is linked).

---
Written by an agent (Claude Code, claude-opus-4-8).

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<PackageKey> = HashSet::new();
-        let mut cold_cas_paths: Vec<(&PackageKey, HashMap<String, PathBuf>)> = Vec::new();
+        let mut cold_cas_paths: Vec<(&PackageKey, &SnapshotEntry, HashMap<String, PathBuf>)> =
+            Vec::new();
         if !cold.is_empty() {
             let prefetched_ref = Some(&prefetched);
             let verified_files_cache_ref = &verified_files_cache;
-            type ColdOutcome<'a> =
-                (Option<PackageKey>, Option<(&'a PackageKey, HashMap<String, PathBuf>)>);
+            type ColdOutcome<'a> = (
+                Option<PackageKey>,
+                Option<(&'a PackageKey, &'a SnapshotEntry, HashMap<String, PathBuf>)>,
+            );
             let outcomes: Vec<ColdOutcome<'_>> = 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::<Reporter>()
                     .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::<Reporter>()
+                    .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

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,

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::<pacquet_reporter::SilentReporter>()
     .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::<pacquet_reporter::SilentReporter>()
     .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::<pacquet_reporter::SilentReporter>()
     .await