diff --git a/Cargo.lock b/Cargo.lock index c228da3e01..d67969e7de 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -11805,6 +11805,7 @@ dependencies = [ "parking_lot", "pathdiff", "petgraph 0.6.5", + "rayon", "reqwest 0.12.24", "rustls", "rustls-native-certs", diff --git a/crates/ruborist/Cargo.toml b/crates/ruborist/Cargo.toml index ffdbcfae77..066ee01efe 100644 --- a/crates/ruborist/Cargo.toml +++ b/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. diff --git a/crates/ruborist/src/model/manifest.rs b/crates/ruborist/src/model/manifest.rs index e56168ee05..37e95deb9c 100644 --- a/crates/ruborist/src/model/manifest.rs +++ b/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>)` — 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, + version: String, +) -> (String, Option>) { + #[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, diff --git a/crates/ruborist/src/resolver/preload.rs b/crates/ruborist/src/resolver/preload.rs index 14824b6cec..1230c5bf68 100644 --- a/crates/ruborist/src/resolver/preload.rs +++ b/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); diff --git a/crates/ruborist/src/service/dns.rs b/crates/ruborist/src/service/dns.rs index ac61fec1eb..3b0ebf29a9 100644 --- a/crates/ruborist/src/service/dns.rs +++ b/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>>, inflight: Arc>>>, 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 { + 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> 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) }) } diff --git a/crates/ruborist/src/service/http.rs b/crates/ruborist/src/service/http.rs index b58e60dbef..fd5ca63c48 100644 --- a/crates/ruborist/src/service/http.rs +++ b/crates/ruborist/src/service/http.rs @@ -159,7 +159,13 @@ pub fn client_builder() -> Result { 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")] { diff --git a/crates/ruborist/src/service/manifest.rs b/crates/ruborist/src/service/manifest.rs index 4e796d3f8e..74baf3b9ca 100644 --- a/crates/ruborist/src/service/manifest.rs +++ b/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(mut bytes: Vec) -> Result +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::(&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::(&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(bytes) + .await + .map_err(FetchError::Permanent) } else { Err(classify_status(response.status(), &url)) } diff --git a/crates/ruborist/src/service/registry.rs b/crates/ruborist/src/service/registry.rs index a978be50d8..be2a8b61f7 100644 --- a/crates/ruborist/src/service/registry.rs +++ b/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>, + /// 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>, } /// 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,7 +447,7 @@ impl UnifiedRegistry { &self, name: &str, spec: &str, - ) -> Result<(String, CoreVersionManifest), RegistryError> { + ) -> Result<(String, Arc), RegistryError> { match self.resolve_full_manifest(name).await? { FullManifestResult::Full(full) => { if full.versions.is_empty() { @@ -440,7 +455,18 @@ impl UnifiedRegistry { } 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`), 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))) } } }