perf(pm): sequential writes within tarball extraction

crates/pm/src/main.rs

@@ -352,6 +352,15 @@ fn main() {
     let result = tokio::runtime::Builder::new_multi_thread()
         .enable_all()
         .worker_threads(worker_threads)
+        // Cap the blocking pool to reduce thread-creation thrash on
+        // runs that burst-dispatch ~128 spawn_blocking calls (JSON
+        // parse for manifest fetches). Default 512 leaves room for
+        // unbounded growth; ~32 threads is enough to saturate a 4-10
+        // core machine on short CPU-bound tasks while keeping the
+        // pool warm across runs. Observed locally: default pool gave
+        // bimodal walls (2.7s fast / 6.9s thrash on M2); capping at
+        // 32 eliminates the thrash peak.
+        .max_blocking_threads(worker_threads)
         .build()
         .expect("failed to build tokio runtime")
         .block_on(async_main());

crates/pm/src/util/extractor.rs

@@ -65,13 +65,24 @@ async fn extract_tarball(gzip_bytes: Bytes, dest: &Path) -> Result<()> {
     rx.await.with_context(|| "Extract task panicked")?
 }
 
-/// Synchronous extraction: decompress + parse + parallel write, all on rayon.
+/// Synchronous extraction: decompress + parse + chunked parallel write, all on rayon.
+///
+/// Writes are batched into fixed-size chunks: each rayon task writes a contiguous
+/// run of `WRITE_CHUNK_SIZE` files sequentially. This keeps multi-core parallelism
+/// for IO-overlap while cutting the rayon task count (and its work-stealing
+/// futex park/unpark pressure) by the chunk factor. Cross-package parallelism is
+/// preserved by the outer `rayon::spawn` in `extract_tarball`.
 fn extract_tarball_sync(gzip_bytes: Bytes, estimated_size: usize, dest: &Path) -> Result<()> {
     use rayon::prelude::*;
     use std::collections::HashSet;
     use std::fs;
     use std::io::{Cursor, Read, Write};
 
+    /// Files per rayon task when writing extracted tar entries. Sized to amortize
+    /// work-stealing overhead while leaving enough tasks to keep cores busy on
+    /// large packages (e.g. 500-file packages → ~16 tasks with WRITE_CHUNK_SIZE=32).
+    const WRITE_CHUNK_SIZE: usize = 32;
+
     // Decompress gzip using libdeflate
     let mut output = vec![0u8; estimated_size];
 
@@ -152,21 +163,27 @@ fn extract_tarball_sync(gzip_bytes: Bytes, estimated_size: usize, dest: &Path) -
         fs::create_dir(dir).ok();
     }
 
-    // Write files in parallel using rayon
-    entries.par_iter().try_for_each(|entry| -> Result<()> {
-        let mut file = fs::File::create(&entry.path)
-            .with_context(|| format!("Failed to create: {}", entry.path.display()))?;
-        file.write_all(&entry.content)
-            .with_context(|| format!("Failed to write: {}", entry.path.display()))?;
-
-        // Skip chmod for 0o644 (most files) — File::create() already produces
-        // this via umask (0o666 & ~0o022 = 0o644).
-        #[cfg(unix)]
-        if entry.mode != 0o644 {
-            fs::set_permissions(&entry.path, fs::Permissions::from_mode(entry.mode)).ok();
-        }
-        Ok(())
-    })?;
+    // Chunked parallel writes: N files per rayon task. Retains IO-overlap
+    // parallelism across cores while cutting the rayon task count (and the
+    // associated work-stealing futex traffic) relative to per-file par_iter.
+    entries
+        .par_chunks(WRITE_CHUNK_SIZE)
+        .try_for_each(|chunk| -> Result<()> {
+            for entry in chunk {
+                let mut file = fs::File::create(&entry.path)
+                    .with_context(|| format!("Failed to create: {}", entry.path.display()))?;
+                file.write_all(&entry.content)
+                    .with_context(|| format!("Failed to write: {}", entry.path.display()))?;
+
+                // Skip chmod for 0o644 (most files) — File::create() already
+                // produces this via umask (0o666 & ~0o022 = 0o644).
+                #[cfg(unix)]
+                if entry.mode != 0o644 {
+                    fs::set_permissions(&entry.path, fs::Permissions::from_mode(entry.mode)).ok();
+                }
+            }
+            Ok(())
+        })?;
 
     // Set directory permissions
     #[cfg(unix)]

crates/pm/src/util/user_config.rs

@@ -132,9 +132,22 @@ pub fn get_install_scope() -> InstallScope {
     INSTALL_SCOPE.get().copied().unwrap_or_default()
 }
 
-// Manifest fetch concurrency configuration
+// Manifest fetch concurrency configuration.
+//
+// Sweep history under the worker-pool architecture:
+//   cap=96  spawn_blocking parse: wall=2.23s avg_conc=66
+//   cap=128 spawn_blocking parse: wall=2.15s avg_conc=84
+//   cap=160 spawn_blocking parse: wall=1.90s avg_conc=81 (parse queue p95=200ms)
+//   cap=160 inline parse        : wall=2.14s avg_conc=119 (parse queue=0ms)
+//
+// Inline parse eliminated blocking-pool queue but cap=160 lifted
+// concurrency past the per-source Cloudflare throttle, inflating
+// per-req from 55 ms → 93 ms. Net wall flat. cap=128 + inline parse
+// should land near the sweet spot: enough concurrency to saturate
+// independent connections, low enough to stay under Cloudflare's
+// per-source throttle threshold.
 static MANIFESTS_CONCURRENCY_LIMIT: LazyLock<ConfigValue<usize>> =
-    LazyLock::new(|| ConfigValue::new("manifests-concurrency-limit", 64));
+    LazyLock::new(|| ConfigValue::new("manifests-concurrency-limit", 128));
 
 pub fn set_manifests_concurrency_limit(value: Option<usize>) {
     MANIFESTS_CONCURRENCY_LIMIT.set(value);

crates/ruborist/Cargo.toml

@@ -7,46 +7,47 @@ license     = "MIT"
 
 # tombi: format.rules.table-keys-order.disabled = true
 [dependencies]
-anyhow       = { workspace = true }
-async-trait  = { workspace = true }
-bytes        = "1"
-dashmap      = "6.1.0"
-pathdiff     = { workspace = true }
-deno_semver  = "0.7"
-dirs         = { workspace = true }
-futures      = "0.3"
-parking_lot  = { workspace = true }
-petgraph     = "0.6"
-serde        = { version = "1.0", features = ["derive"] }
-serde_json   = { workspace = true }
-simd-json    = "0.17"
-thiserror    = { workspace = true }
-tokio-fs-ext = { workspace = true }
-tokio-retry  = "0.3"
-tracing      = { workspace = true }
+anyhow          = { workspace = true }
+async-trait     = { workspace = true }
+bytes           = "1"
+crossbeam-queue = "0.3"
+dashmap         = "6.1.0"
+pathdiff        = { workspace = true }
+deno_semver     = "0.7"
+dirs            = { workspace = true }
+futures         = "0.3"
+parking_lot     = { workspace = true }
+petgraph        = "0.6"
+serde           = { version = "1.0", features = ["derive"] }
+serde_json      = { workspace = true }
+simd-json       = "0.17"
+thiserror       = { workspace = true }
+tokio-fs-ext    = { workspace = true }
+tokio-retry     = "0.3"
+tracing         = { workspace = true }
 # TLS provider is selected per-target below: macOS overrides to
 # aws-lc-rs (see service/http.rs); every other target keeps ring.
-reqwest      = { version = "0.12", default-features = false, features = ["http2", "json", "socks"] }
+reqwest         = { version = "0.12", default-features = false, features = ["http2", "json", "socks"] }
 # Shared URL-hash + staging helpers for native-only resolvers (git / http)
-sha2         = { workspace = true, optional = true }
-tempfile     = { workspace = true, optional = true }
+sha2            = { workspace = true, optional = true }
+tempfile        = { workspace = true, optional = true }
 # Git clone (gated on `native-git`)
-gix          = { workspace = true, features = [
+gix             = { workspace = true, features = [
   "blocking-http-transport-reqwest-rust-tls",
   "blocking-network-client",
 ], optional = true }
 # HTTP tarball extraction (gated on `http-tarball`)
 # Uses libdeflater to match pm's extractor and avoid pulling a second gzip
 # implementation (flate2) into the workspace.
-libdeflater  = { version = "1.25", optional = true }
-tar          = { version = "0.4", optional = true }
+libdeflater     = { version = "1.25", optional = true }
+tar             = { version = "0.4", optional = true }
 
 # Async runtime (works on both native and WASM with forked tokio)
 [dependencies.tokio]
 workspace = true
 
 [dev-dependencies]
-tokio = { workspace = true, features = ["macros", "rt"] }
+tokio = { workspace = true, features = ["macros", "rt", "rt-multi-thread"] }
 
 # Native (non-macOS) targets: reqwest's default rustls + ring.
 [target.'cfg(not(any(target_arch = "wasm32", target_os = "macos")))'.dependencies]

crates/ruborist/src/resolver/builder.rs

@@ -30,11 +30,12 @@ use crate::model::graph::{DependencyGraph, FindResult, PackageNode};
 use crate::model::manifest::NodeManifest;
 use crate::model::node::EdgeType;
 use crate::model::package_json::PackageJson;
+#[cfg(not(target_arch = "wasm32"))]
 use crate::resolver::preload::{PreloadConfig, preload_manifests};
 use crate::resolver::registry::{ResolveError, resolve_registry_dep};
 use crate::spec::{Catalogs, PackageSpec, Protocol};
 use crate::traits::progress::{BuildEvent, EventReceiver, NoopReceiver};
-use crate::traits::registry::{RegistryClient, ResolvedPackage};
+use crate::traits::registry::{MaybeSend, MaybeSync, RegistryClient, ResolvedPackage};
 
 /// Dispatch a git/github spec to the real `gix`-backed resolver when the
 /// `native-git` feature is enabled, otherwise error with a hint.
@@ -119,11 +120,19 @@ pub struct BuildDepsConfig {
     pub catalogs: Catalogs,
 }
 
+/// Fallback concurrency for wasm where the worker-pool preload module is
+/// gated out. wasm runs serial BFS resolution so this only sizes the inner
+/// `FuturesUnordered` for transitive fanout — keep it modest.
+#[cfg(target_arch = "wasm32")]
+const DEFAULT_CONCURRENCY: usize = 16;
+#[cfg(not(target_arch = "wasm32"))]
+const DEFAULT_CONCURRENCY: usize = crate::resolver::preload::DEFAULT_CONCURRENCY;
+
 impl Default for BuildDepsConfig {
     fn default() -> Self {
         Self {
             peer_deps: PeerDeps::Skip,
-            concurrency: crate::resolver::preload::DEFAULT_CONCURRENCY,
+            concurrency: DEFAULT_CONCURRENCY,
             skip_preload: false,
             cache_dir: dirs::home_dir().map(|d| d.join(".cache/nm")),
             git_clone_cache: Arc::new(GitCloneCache::new()),
@@ -450,7 +459,7 @@ async fn process_file_dep<E>(
 ///
 /// # Returns
 /// The result of processing (reused, created, or skipped)
-pub async fn process_dependency<R: RegistryClient>(
+pub async fn process_dependency<R: RegistryClient + MaybeSync>(
     graph: &mut DependencyGraph,
     registry: &R,
     node_index: NodeIndex,
@@ -667,11 +676,14 @@ pub async fn process_dependency<R: RegistryClient>(
 /// // Add initial dependency edges to root...
 /// build_deps(&mut graph, &registry, PeerDeps::Include).await?;
 /// ```
-pub async fn build_deps<R: RegistryClient>(
+pub async fn build_deps<R: RegistryClient + Clone + MaybeSend + MaybeSync + 'static>(
     graph: &mut DependencyGraph,
     registry: &R,
     peer_deps: PeerDeps,
-) -> Result<(), ResolveError<R::Error>> {
+) -> Result<(), ResolveError<R::Error>>
+where
+    R::Error: MaybeSend,
+{
     let config = BuildDepsConfig::default().with_peer_deps(peer_deps);
     build_deps_with_config(graph, registry, config, &NoopReceiver).await
 }
@@ -689,12 +701,18 @@ pub async fn build_deps<R: RegistryClient>(
 /// * `registry` - Registry client for fetching packages
 /// * `peer_deps` - How to handle peer dependencies
 /// * `receiver` - Event receiver for handling build events
-pub async fn build_deps_with_receiver<R: RegistryClient, E: EventReceiver>(
+pub async fn build_deps_with_receiver<
+    R: RegistryClient + Clone + MaybeSend + MaybeSync + 'static,
+    E: EventReceiver,
+>(
     graph: &mut DependencyGraph,
     registry: &R,
     peer_deps: PeerDeps,
     receiver: &E,
-) -> Result<(), ResolveError<R::Error>> {
+) -> Result<(), ResolveError<R::Error>>
+where
+    R::Error: MaybeSend,
+{
     let config = BuildDepsConfig::default().with_peer_deps(peer_deps);
     build_deps_with_config(graph, registry, config, receiver).await
 }
@@ -719,20 +737,28 @@ pub async fn build_deps_with_receiver<R: RegistryClient, E: EventReceiver>(
 ///
 /// build_deps_with_config(&mut graph, &registry, config, &receiver).await?;
 /// ```
-pub async fn build_deps_with_config<R: RegistryClient, E: EventReceiver>(
+pub async fn build_deps_with_config<
+    R: RegistryClient + Clone + MaybeSend + MaybeSync + 'static,
+    E: EventReceiver,
+>(
     graph: &mut DependencyGraph,
     registry: &R,
     config: BuildDepsConfig,
     receiver: &E,
-) -> Result<(), ResolveError<R::Error>> {
+) -> Result<(), ResolveError<R::Error>>
+where
+    R::Error: MaybeSend,
+{
     tracing::debug!(
         "Starting dependency tree build, peer_deps: {:?}, concurrency: {}, skip_preload: {}",
         config.peer_deps,
         config.concurrency,
         config.skip_preload
     );
 
-    // Phase 1: Preload manifests in parallel (unless skipped)
+    // Phase 1: Preload manifests in parallel (unless skipped).
+    // wasm has no worker-pool — preload runs serially in BFS instead.
+    #[cfg(not(target_arch = "wasm32"))]
     run_preload_phase(graph, registry, &config, receiver).await;
 
     // Phase 2: BFS traversal to build the dependency tree
@@ -746,12 +772,18 @@ pub async fn build_deps_with_config<R: RegistryClient, E: EventReceiver>(
 }
 
 /// Run the preload phase to warm up the cache with manifests.
-async fn run_preload_phase<R: RegistryClient, E: EventReceiver>(
+#[cfg(not(target_arch = "wasm32"))]
+async fn run_preload_phase<
+    R: RegistryClient + Clone + MaybeSend + MaybeSync + 'static,
+    E: EventReceiver,
+>(
     graph: &DependencyGraph,
     registry: &R,
     config: &BuildDepsConfig,
     receiver: &E,
-) {
+) where
+    R::Error: MaybeSend,
+{
     if config.skip_preload {
         return;
     }
@@ -764,18 +796,20 @@ async fn run_preload_phase<R: RegistryClient, E: EventReceiver>(
     }
 
     tracing::debug!("Preload phase: {} initial dependencies", initial_deps.len());
-    receiver.on_event(BuildEvent::PreloadStart {
-        count: initial_deps.len(),
-    });
 
     let preload_config = PreloadConfig {
         peer_deps: config.peer_deps,
         concurrency: config.concurrency,
     };
 
+    // Wrap registry in Arc for the worker pool. Each worker holds an
+    // Arc<R> so resolution runs on tokio's global executor independently
+    // of the main task driving result aggregation.
+    let registry_arc = std::sync::Arc::new(registry.clone());
+
     let stats = preload_manifests(
         initial_deps,
-        registry,
+        registry_arc,
         preload_config,
         receiver,
         |_name, _manifest| {
@@ -784,11 +818,6 @@ async fn run_preload_phase<R: RegistryClient, E: EventReceiver>(
     )
     .await;
 
-    tracing::debug!(
-        "Preload phase completed: {} success, {} failed",
-        stats.success_count,
-        stats.failed_count
-    );
     receiver.on_event(BuildEvent::PreloadComplete {
         success: stats.success_count,
         failed: stats.failed_count,
@@ -798,7 +827,7 @@ async fn run_preload_phase<R: RegistryClient, E: EventReceiver>(
 }
 
 /// Run the BFS traversal phase to build the dependency tree.
-async fn run_bfs_phase<R: RegistryClient, E: EventReceiver>(
+async fn run_bfs_phase<R: RegistryClient + MaybeSync, E: EventReceiver>(
     graph: &mut DependencyGraph,
     registry: &R,
     config: &BuildDepsConfig,
@@ -921,10 +950,13 @@ use std::path::Path;
 /// let pkg: PackageJson = serde_json::from_str(&pkg_content)?;
 /// let lock = resolve(&pkg, &registry).await?;
 /// ```
-pub async fn resolve<R: RegistryClient>(
+pub async fn resolve<R: RegistryClient + Clone + MaybeSend + MaybeSync + 'static>(
     pkg: &PackageJson,
     registry: &R,
-) -> Result<PackageLock, ResolveError<R::Error>> {
+) -> Result<PackageLock, ResolveError<R::Error>>
+where
+    R::Error: MaybeSend,
+{
     resolve_with_options(pkg, registry, PeerDeps::Include, &NoopReceiver).await
 }
 
@@ -935,12 +967,18 @@ pub async fn resolve<R: RegistryClient>(
 /// * `registry` - Registry client for fetching packages
 /// * `peer_deps` - How to handle peer dependencies
 /// * `receiver` - Event receiver for progress tracking
-pub async fn resolve_with_options<R: RegistryClient, E: EventReceiver>(
+pub async fn resolve_with_options<
+    R: RegistryClient + Clone + MaybeSend + MaybeSync + 'static,
+    E: EventReceiver,
+>(
     pkg: &PackageJson,
     registry: &R,
     peer_deps: PeerDeps,
     receiver: &E,
-) -> Result<PackageLock, ResolveError<R::Error>> {
+) -> Result<PackageLock, ResolveError<R::Error>>
+where
+    R::Error: MaybeSend,
+{
     // Create graph with root node
     let mut graph = DependencyGraph::from_package_json(PathBuf::from("."), pkg.clone());