perf(pm): stack PR #2818 unmerged modules on top of worker-pool

crates/pm/src/main.rs

@@ -352,6 +352,14 @@ 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; ~worker_threads count is enough to
+        // saturate a 4-10 core machine on short CPU-bound tasks while
+        // keeping the pool warm. Default pool gave bimodal walls
+        // (2.7s fast / 6.9s thrash on M2); capping eliminates thrash.
+        .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/logger.rs

@@ -53,8 +53,23 @@ pub fn init_tracing(verbose: bool) -> Result<(PathBuf, WorkerGuard)> {
     let console_filter = EnvFilter::try_from_default_env()
         .unwrap_or_else(|_| EnvFilter::new(format!("utoo={console_level}")));
 
-    // File filter: always capture debug+ for troubleshooting
-    let file_filter = EnvFilter::new("utoo=debug");
+    // File filter: capture info+ by default. Hot-path debug!() calls
+    // (cache hits, per-package preload events, BFS dispatch) emit
+    // ~5-10 events per resolved manifest. With 2730+ manifests during
+    // a cold preload that's 15-30k events that — even routed through
+    // the non_blocking appender's channel — pay format/serialise CPU
+    // on the resolving thread. CI standalone manifest-bench reaches
+    // avg_conc=92 at cap=128 with the same reqwest stack; ruborist
+    // sat at avg_conc=56 after every other Mutex/clone hot-path was
+    // eliminated. Tracing was the last remaining shared cost.
+    //
+    // Override via `UTOO_FILE_LOG=debug` (or any RUST_LOG-style spec)
+    // to bring the troubleshooting captures back when actually
+    // diagnosing.
+    let file_filter = match env::var("UTOO_FILE_LOG") {
+        Ok(spec) if !spec.is_empty() => EnvFilter::new(spec),
+        _ => EnvFilter::new("utoo=info"),
+    };
 
     // 2. Create log file
     let timestamp = SystemTime::now()

crates/pm/src/util/user_config.rs

@@ -132,9 +132,23 @@ pub fn get_install_scope() -> InstallScope {
     INSTALL_SCOPE.get().copied().unwrap_or_default()
 }
 
-// Manifest fetch concurrency configuration
+// Manifest fetch concurrency configuration.
+//
+// Sweep history under worker-pool + HTTP/1.1 + DNS per-family rotation:
+//   cap=64  wall=3.10s avg_conc=56 (FuturesUnordered ceiling)
+//   cap=96  wall=2.87-3.08s ratio 0.83-1.03 (s9-s10 with H1 stack)
+//   cap=128 wall=2.15s avg_conc=84 (PR #2818 historical sweet spot, inline parse)
+//   cap=160 wall=2.14s avg_conc=119 (Cloudflare per-IP throttle bites)
+//
+// 96 was chosen empirically when burst aggression on H2 was triggering
+// Cloudflare throttle (pre-H1 era). Once `.http1_only()` landed and
+// requests stopped multiplexing through a single TCP connection, cap=128
+// becomes the right setting again — independent TCP streams distribute
+// across the rotated DNS pool without tripping per-IP throttle, and the
+// extra 33% concurrency saturates more of the Cloudflare connection
+// budget.
 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/lib.rs

@@ -22,6 +22,7 @@
 //! }).await?;
 //! ```
 
+pub mod maybe_send;
 pub mod model;
 pub mod resolver;
 pub mod service;

crates/ruborist/src/maybe_send.rs

@@ -0,0 +1,27 @@
+//! `Send`/`Sync` shim that is `Send`/`Sync` on native targets and
+//! no-op on `wasm32`. Lets the resolver share a single trait surface
+//! across multi-thread tokio (where futures must be `Send` to be
+//! `tokio::spawn`-able) and wasm-bindgen (where `JsFuture` is `!Send`
+//! and tasks run via `wasm_bindgen_futures::spawn_local`).
+
+/// `Send` on native, no-op on wasm32.
+#[cfg(not(target_arch = "wasm32"))]
+pub trait MaybeSend: Send {}
+#[cfg(not(target_arch = "wasm32"))]
+impl<T: Send> MaybeSend for T {}
+
+#[cfg(target_arch = "wasm32")]
+pub trait MaybeSend {}
+#[cfg(target_arch = "wasm32")]
+impl<T> MaybeSend for T {}
+
+/// `Sync` on native, no-op on wasm32.
+#[cfg(not(target_arch = "wasm32"))]
+pub trait MaybeSync: Sync {}
+#[cfg(not(target_arch = "wasm32"))]
+impl<T: Sync> MaybeSync for T {}
+
+#[cfg(target_arch = "wasm32")]
+pub trait MaybeSync {}
+#[cfg(target_arch = "wasm32")]
+impl<T> MaybeSync for T {}