perf(pm): manifest resolve hot-path — rayon offload + network layer (h1+pool+dns)

crates/ruborist/Cargo.toml

@@ -54,7 +54,11 @@ reqwest = { version = "0.12", default-features = false, features = ["rustls-tls-
 
 # Native-only dependencies (not compiled for WASM)
 [target.'cfg(not(target_arch = "wasm32"))'.dependencies]
-libc = "0.2"
+libc  = "0.2"
+# Dedicated CPU thread pool for simd_json manifest parsing — keeps the
+# tokio runtime free of CPU-bound work so in-flight network IO is not
+# blocked by sibling parses. WASM falls back to inline parse.
+rayon = { workspace = true }
 
 # Browser handles TLS via the fetch API, so reqwest needs no rustls
 # feature on this target.

crates/ruborist/src/model/manifest.rs

@@ -149,6 +149,36 @@ impl FullManifest {
     }
 }
 
+/// Extract a single version from `FullManifest` on rayon's CPU pool
+/// (native) or inline (wasm32). The native path keeps the tokio runtime
+/// free of `simd_json::to_borrowed_value` work so sibling manifest
+/// fetches keep driving network IO while this one re-parses.
+///
+/// Returns `(version, Option<Arc<CoreVersionManifest>>)` — the input
+/// `version` is handed back to the caller alongside the result so it
+/// can be reused for the error path and the outer return without a
+/// clone. The native path requires `'static` for the rayon closure,
+/// so the closure owns the string for its duration.
+pub async fn extract_core_version_off_runtime(
+    full: Arc<FullManifest>,
+    version: String,
+) -> (String, Option<Arc<CoreVersionManifest>>) {
+    #[cfg(not(target_arch = "wasm32"))]
+    {
+        let (tx, rx) = tokio::sync::oneshot::channel();
+        rayon::spawn(move || {
+            let core = full.get_core_version(&version).map(Arc::new);
+            let _ = tx.send((version, core));
+        });
+        rx.await.expect("rayon parse worker dropped before sending")
+    }
+    #[cfg(target_arch = "wasm32")]
+    {
+        let core = full.get_core_version(&version).map(Arc::new);
+        (version, core)
+    }
+}
+
 /// Deserialize a versions map by extracting only the keys, skipping all values.
 ///
 /// Uses `IgnoredAny` to skip over version manifest JSON objects without allocating,

crates/ruborist/src/resolver/preload.rs

@@ -15,7 +15,7 @@ use crate::traits::progress::{BuildEvent, EventReceiver};
 use crate::traits::registry::RegistryClient;
 
 /// Default concurrency limit for manifest fetching
-pub const DEFAULT_CONCURRENCY: usize = 64;
+pub const DEFAULT_CONCURRENCY: usize = 128;
 
 /// A dependency spec: (name, version_spec)
 pub type Dep = (String, String);

crates/ruborist/src/service/dns.rs

@@ -14,6 +14,7 @@
 mod native {
     use std::collections::HashMap;
     use std::net::SocketAddr;
+    use std::sync::atomic::{AtomicUsize, Ordering};
     use std::sync::{Arc, LazyLock};
     use std::time::{Duration, Instant};
 
@@ -70,6 +71,12 @@ mod native {
         cache: Arc<Mutex<HashMap<String, CacheEntry>>>,
         inflight: Arc<Mutex<HashMap<String, Arc<InflightEntry>>>>,
         ttl: Duration,
+        /// Round-robin counter used to rotate the returned address order
+        /// on every `resolve` call. Spreads concurrent pool connections
+        /// across all DNS A records (e.g. antgroup registry returns 8
+        /// IPs) instead of pinning every new connection to the first IP
+        /// like reqwest does by default.
+        rr_counter: AtomicUsize,
     }
 
     impl CachingResolver {
@@ -79,10 +86,28 @@ mod native {
                 cache: Arc::new(Mutex::new(HashMap::new())),
                 inflight: Arc::new(Mutex::new(HashMap::new())),
                 ttl,
+                rr_counter: AtomicUsize::new(0),
             }
         }
     }
 
+    /// Rotate `addrs` left by `offset` positions, producing a fresh Vec so
+    /// reqwest's connect loop tries a different IP on every new connection.
+    ///
+    /// e.g. `[A, B, C, D]` with offset 2 → `[C, D, A, B]`.
+    fn rotate_addrs(addrs: &[SocketAddr], offset: usize) -> Vec<SocketAddr> {
+        if addrs.is_empty() {
+            return Vec::new();
+        }
+        let n = addrs.len();
+        let start = offset % n;
+        addrs[start..]
+            .iter()
+            .chain(&addrs[..start])
+            .copied()
+            .collect()
+    }
+
     /// Perform a system DNS lookup and cache the result.
     async fn do_lookup(
         hostname: &str,
@@ -132,10 +157,11 @@ mod native {
                 if let Some(entry) = cache.get(&hostname)
                     && entry.expires_at > Instant::now()
                 {
-                    let cached = entry.addrs.to_vec();
-                    return Box::pin(std::future::ready(
-                        Ok(Box::new(cached.into_iter()) as Addrs),
-                    ));
+                    let offset = self.rr_counter.fetch_add(1, Ordering::Relaxed);
+                    let rotated = rotate_addrs(&entry.addrs, offset);
+                    return Box::pin(std::future::ready(Ok(
+                        Box::new(rotated.into_iter()) as Addrs
+                    )));
                 }
             }
 
@@ -155,6 +181,7 @@ mod native {
             // Clone Arcs for the 'static async block (required by Resolve trait).
             let cache = Arc::clone(&self.cache);
             let inflight_map = Arc::clone(&self.inflight);
+            let offset = self.rr_counter.fetch_add(1, Ordering::Relaxed);
 
             Box::pin(async move {
                 let result = inflight
@@ -167,13 +194,8 @@ mod native {
                 inflight_map.lock().remove(&hostname);
 
                 let resolved = result?;
-
-                // resolved is a &Arc<Vec<SocketAddr>> borrowed from the OnceCell,
-                // but Addrs requires a 'static owned iterator.
-                // to_vec() + into_iter() creates an owned copy; iter().copied()
-                // would borrow from the local reference and fail lifetime checks.
-                #[allow(clippy::unnecessary_to_owned)]
-                let addrs: Addrs = Box::new(resolved.to_vec().into_iter());
+                let rotated = rotate_addrs(resolved.as_ref(), offset);
+                let addrs: Addrs = Box::new(rotated.into_iter());
                 Ok(addrs)
             })
         }

crates/ruborist/src/service/http.rs

@@ -159,7 +159,13 @@ pub fn client_builder() -> Result<reqwest::ClientBuilder> {
         let mut builder = builder
             .no_proxy()
             .dns_resolver(shared_resolver())
-            .connect_timeout(CONNECT_TIMEOUT);
+            .connect_timeout(CONNECT_TIMEOUT)
+            // Force HTTP/1.1 with a connection pool. reqwest multiplexes all
+            // requests over a single HTTP/2 connection by default, which
+            // makes head-of-line blocking on one slow response stall the
+            // whole manifest fetch phase. An H1 pool lets concurrent
+            // manifest requests open independent TCP streams instead.
+            .http1_only();
 
         #[cfg(target_os = "macos")]
         {

crates/ruborist/src/service/manifest.rs

@@ -13,6 +13,31 @@ use super::fetch::{
 use super::http::get_client;
 use crate::model::manifest::{CoreVersionManifest, FullManifest};
 
+/// Parse JSON bytes on rayon's CPU thread pool (native) or inline
+/// (wasm32). Keeps the tokio runtime free of `simd_json` work so other
+/// in-flight manifest fetches keep driving network IO while this one
+/// parses.
+async fn parse_json_off_runtime<T>(mut bytes: Vec<u8>) -> Result<T, anyhow::Error>
+where
+    T: serde::de::DeserializeOwned + Send + 'static,
+{
+    #[cfg(not(target_arch = "wasm32"))]
+    {
+        let (tx, rx) = tokio::sync::oneshot::channel();
+        rayon::spawn(move || {
+            let result = simd_json::serde::from_slice::<T>(&mut bytes)
+                .map_err(|e| anyhow!("JSON parse error: {e}"));
+            let _ = tx.send(result);
+        });
+        rx.await
+            .map_err(|e| anyhow!("rayon parse channel closed: {e}"))?
+    }
+    #[cfg(target_arch = "wasm32")]
+    {
+        simd_json::serde::from_slice::<T>(&mut bytes).map_err(|e| anyhow!("JSON parse error: {e}"))
+    }
+}
+
 /// Result of a full manifest fetch with ETag support.
 /// Transient return value, immediately destructured — Box not needed.
 #[allow(clippy::large_enum_variant)]
@@ -89,11 +114,12 @@ pub async fn fetch_full_manifest(opts: FetchManifestOptions<'_>) -> Result<Fetch
                         .await
                         .map_err(|e| FetchError::Permanent(anyhow!("Response read error: {e}")))?
                         .to_vec();
-                    // Save raw bytes before simd_json mutates the parse buffer.
-                    let mut parse_buf = raw_bytes.clone();
-                    let mut manifest: FullManifest =
-                        simd_json::serde::from_slice(&mut parse_buf)
-                            .map_err(|e| FetchError::Permanent(anyhow!("JSON parse error: {e}")))?;
+                    // simd_json mutates the parse buffer; clone so the raw
+                    // bytes survive for `manifest.raw`.
+                    let parse_buf = raw_bytes.clone();
+                    let mut manifest: FullManifest = parse_json_off_runtime(parse_buf)
+                        .await
+                        .map_err(FetchError::Permanent)?;
                     manifest.raw = std::sync::Arc::from(raw_bytes);
 
                     Ok(FetchManifestResult::Ok(manifest, new_etag))
@@ -173,13 +199,14 @@ pub async fn fetch_version_manifest(
                     .map_err(classify_reqwest_error)?;
 
                 if response.status().is_success() {
-                    let mut bytes = response
+                    let bytes = response
                         .bytes()
                         .await
                         .map_err(|e| FetchError::Permanent(anyhow!("Response read error: {e}")))?
                         .to_vec();
-                    simd_json::serde::from_slice(&mut bytes)
-                        .map_err(|e| FetchError::Permanent(anyhow!("JSON parse error: {e}")))
+                    parse_json_off_runtime::<CoreVersionManifest>(bytes)
+                        .await
+                        .map_err(FetchError::Permanent)
                 } else {
                     Err(classify_status(response.status(), &url))
                 }

crates/ruborist/src/service/registry.rs

@@ -39,10 +39,12 @@ fn current_timestamp_secs() -> u64 {
     (js_sys::Date::now() / 1000.0) as u64
 }
 
+use dashmap::DashSet;
+
 use super::cache::{PackageCache, Versions, VersionsInfo};
 use super::manifest;
 use super::store::{ManifestStore, NoopStore};
-use crate::model::manifest::{CoreVersionManifest, FullManifest};
+use crate::model::manifest::{CoreVersionManifest, FullManifest, extract_core_version_off_runtime};
 use crate::resolver::semver::normalize_spec;
 use crate::resolver::version::resolve_target_version;
 use crate::traits::registry::{RegistryClient, RegistryError, ResolvedPackage, is_npm_registry};
@@ -84,6 +86,13 @@ pub struct UnifiedRegistry {
     /// `(name, spec)`. Same gate-only pattern: the canonical `Arc<…>`
     /// lives in `PackageCache.version_manifests`; the gate stores `()`.
     inflight_version: Arc<OnceMap<(String, String), ()>>,
+    /// Dedup set for `store_version_manifest` disk writes keyed by
+    /// `(name, resolved_version)`. `inflight_version` is keyed by
+    /// `(name, spec)`, so sibling specs (e.g. `^1.0.0` and `^1.2.0`)
+    /// resolving to the same version each fire the gate independently
+    /// and would otherwise issue duplicate writes for the same path.
+    /// First insert wins; subsequent specs skip the redundant write.
+    stored_version: Arc<DashSet<(String, String)>>,
 }
 
 /// Builder for `UnifiedRegistry`.
@@ -152,6 +161,7 @@ impl UnifiedRegistryBuilder {
             supports_semver,
             inflight_full: Arc::new(OnceMap::new()),
             inflight_version: Arc::new(OnceMap::new()),
+            stored_version: Arc::new(DashSet::new()),
         }
     }
 }
@@ -171,6 +181,7 @@ impl Clone for UnifiedRegistry {
             supports_semver: self.supports_semver,
             inflight_full: Arc::clone(&self.inflight_full),
             inflight_version: Arc::clone(&self.inflight_version),
+            stored_version: Arc::clone(&self.stored_version),
         }
     }
 }
@@ -369,9 +380,8 @@ impl UnifiedRegistry {
                         // version from the full manifest. `resolve_full_manifest`
                         // is itself inflight-gated, so concurrent specs for the
                         // same name share one full-manifest fetch.
-                        let (resolved_version, manifest) =
+                        let (resolved_version, arc) =
                             self.resolve_via_full_manifest(name, spec).await?;
-                        let arc = Arc::new(manifest);
                         self.cache.set_version_manifest(
                             name.to_string(),
                             spec.to_string(),
@@ -384,11 +394,16 @@ impl UnifiedRegistry {
                                 Arc::clone(&arc),
                             );
                         }
-                        self.store.store_version_manifest(
-                            name,
-                            &resolved_version,
-                            Arc::clone(&arc),
-                        );
+                        if self
+                            .stored_version
+                            .insert((name.to_string(), resolved_version.clone()))
+                        {
+                            self.store.store_version_manifest(
+                                name,
+                                &resolved_version,
+                                Arc::clone(&arc),
+                            );
+                        }
                         return Ok(());
                     }
 
@@ -432,15 +447,26 @@ impl UnifiedRegistry {
         &self,
         name: &str,
         spec: &str,
-    ) -> Result<(String, CoreVersionManifest), RegistryError> {
+    ) -> Result<(String, Arc<CoreVersionManifest>), RegistryError> {
         match self.resolve_full_manifest(name).await? {
             FullManifestResult::Full(full) => {
                 if full.versions.is_empty() {
                     return Err(RegistryError(anyhow!("No versions available for {}", name)));
                 }
                 let resolved_version = resolve_target_version((&*full).into(), spec)
                     .map_err(|e| RegistryError(anyhow!("{}@{}: {}", name, spec, e)))?;
-                let core = full.get_core_version(&resolved_version).ok_or_else(|| {
+                // Race window: while we awaited `resolve_full_manifest` (gated
+                // by `inflight_full<name>`), a sibling spec for the same
+                // package may have resolved to this same version and populated
+                // `version_manifests` cache (writer at line 373-387 stores
+                // both `(name, spec)` and `(name, resolved_version)` keys).
+                // Reuse the Arc instead of paying the off-runtime reparse.
+                if let Some(cached) = self.cache.get_version_manifest(name, &resolved_version) {
+                    return Ok((resolved_version, cached));
+                }
+                let (resolved_version, core) =
+                    extract_core_version_off_runtime(full, resolved_version).await;
+                let core = core.ok_or_else(|| {
                     RegistryError(anyhow!(
                         "Version {} not found in manifest for {}",
                         resolved_version,
@@ -470,7 +496,7 @@ impl UnifiedRegistry {
                     })
                     .await
                     .map_err(RegistryError)?;
-                Ok((resolved_version, manifest))
+                Ok((resolved_version, Arc::new(manifest)))
             }
         }
     }