diff --git a/crates/pm/src/main.rs b/crates/pm/src/main.rs index 8f3a3a3662..889030bc3a 100644 --- a/crates/pm/src/main.rs +++ b/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()); diff --git a/crates/pm/src/util/extractor.rs b/crates/pm/src/util/extractor.rs index 29b71149b0..ca25614814 100644 --- a/crates/pm/src/util/extractor.rs +++ b/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)] diff --git a/crates/pm/src/util/user_config.rs b/crates/pm/src/util/user_config.rs index 34ee45a340..b9f7d4b8d9 100644 --- a/crates/pm/src/util/user_config.rs +++ b/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> = - LazyLock::new(|| ConfigValue::new("manifests-concurrency-limit", 64)); + LazyLock::new(|| ConfigValue::new("manifests-concurrency-limit", 128)); pub fn set_manifests_concurrency_limit(value: Option) { MANIFESTS_CONCURRENCY_LIMIT.set(value); diff --git a/crates/ruborist/Cargo.toml b/crates/ruborist/Cargo.toml index ffdbcfae77..bdd11026ef 100644 --- a/crates/ruborist/Cargo.toml +++ b/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] diff --git a/crates/ruborist/src/resolver/builder.rs b/crates/ruborist/src/resolver/builder.rs index b0bf2794c6..a99e819e6a 100644 --- a/crates/ruborist/src/resolver/builder.rs +++ b/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( /// /// # Returns /// The result of processing (reused, created, or skipped) -pub async fn process_dependency( +pub async fn process_dependency( graph: &mut DependencyGraph, registry: &R, node_index: NodeIndex, @@ -667,11 +676,14 @@ pub async fn process_dependency( /// // Add initial dependency edges to root... /// build_deps(&mut graph, ®istry, PeerDeps::Include).await?; /// ``` -pub async fn build_deps( +pub async fn build_deps( graph: &mut DependencyGraph, registry: &R, peer_deps: PeerDeps, -) -> Result<(), ResolveError> { +) -> Result<(), ResolveError> +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( /// * `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( +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> { +) -> Result<(), ResolveError> +where + R::Error: MaybeSend, +{ let config = BuildDepsConfig::default().with_peer_deps(peer_deps); build_deps_with_config(graph, registry, config, receiver).await } @@ -719,12 +737,18 @@ pub async fn build_deps_with_receiver( /// /// build_deps_with_config(&mut graph, ®istry, config, &receiver).await?; /// ``` -pub async fn build_deps_with_config( +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> { +) -> Result<(), ResolveError> +where + R::Error: MaybeSend, +{ tracing::debug!( "Starting dependency tree build, peer_deps: {:?}, concurrency: {}, skip_preload: {}", config.peer_deps, @@ -732,7 +756,9 @@ pub async fn build_deps_with_config( 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( } /// Run the preload phase to warm up the cache with manifests. -async fn run_preload_phase( +#[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( } 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 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( ) .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( } /// Run the BFS traversal phase to build the dependency tree. -async fn run_bfs_phase( +async fn run_bfs_phase( 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, ®istry).await?; /// ``` -pub async fn resolve( +pub async fn resolve( pkg: &PackageJson, registry: &R, -) -> Result> { +) -> Result> +where + R::Error: MaybeSend, +{ resolve_with_options(pkg, registry, PeerDeps::Include, &NoopReceiver).await } @@ -935,12 +967,18 @@ pub async fn resolve( /// * `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( +pub async fn resolve_with_options< + R: RegistryClient + Clone + MaybeSend + MaybeSync + 'static, + E: EventReceiver, +>( pkg: &PackageJson, registry: &R, peer_deps: PeerDeps, receiver: &E, -) -> Result> { +) -> Result> +where + R::Error: MaybeSend, +{ // Create graph with root node let mut graph = DependencyGraph::from_package_json(PathBuf::from("."), pkg.clone()); diff --git a/crates/ruborist/src/resolver/mod.rs b/crates/ruborist/src/resolver/mod.rs index 582e03b316..38d7944d3e 100644 --- a/crates/ruborist/src/resolver/mod.rs +++ b/crates/ruborist/src/resolver/mod.rs @@ -7,6 +7,10 @@ pub mod edges; pub mod git; #[cfg(feature = "http-tarball")] pub mod http; +// Worker-pool preload uses `tokio::spawn` for background workers, which +// requires `Send` futures. wasm-bindgen-futures aren't `Send`, so we gate +// the whole module to native targets and skip the preload phase on wasm. +#[cfg(not(target_arch = "wasm32"))] pub mod preload; pub mod registry; pub mod runtime; diff --git a/crates/ruborist/src/resolver/preload.rs b/crates/ruborist/src/resolver/preload.rs index 14824b6cec..432ef69401 100644 --- a/crates/ruborist/src/resolver/preload.rs +++ b/crates/ruborist/src/resolver/preload.rs @@ -1,21 +1,36 @@ -//! Parallel manifest preloading for dependency resolution. +//! Parallel manifest preloading via worker pool. //! -//! Uses FuturesUnordered for true streaming concurrency: when a package resolves, -//! its transitive dependencies are immediately added to the queue. +//! Architecture: N long-lived `tokio::spawn` workers pulling work from a +//! shared `SegQueue`. Replaces the prior `FuturesUnordered` design that +//! had main task own the futures and poll them cooperatively, which +//! capped effective parallelism at ~55-60 even when standalone +//! manifest-bench (same reqwest stack, no resolver overhead) sustained +//! 90+ concurrent at the same cap. The deeper `await` chain inside +//! `resolve_package` (registry cache check + `OnceMap::get_or_init` + +//! `RetryIf` + `request.send()` + `bytes()` + parse-spawn_blocking) +//! made every yielded poll round-trip through the main task — +//! starving the dispatch refill. Worker tasks run on tokio's global +//! pool so each future progresses independently. -use std::collections::{HashSet, VecDeque}; use std::sync::Arc; +use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; -use futures::stream::{FuturesUnordered, StreamExt}; +use crossbeam_queue::SegQueue; +use dashmap::DashSet; +use tokio::sync::{Notify, mpsc}; use crate::model::manifest::CoreVersionManifest; use crate::model::node::PeerDeps; +use crate::resolver::registry::ResolveError; use crate::resolver::registry::resolve_package; use crate::traits::progress::{BuildEvent, EventReceiver}; use crate::traits::registry::RegistryClient; -/// Default concurrency limit for manifest fetching -pub const DEFAULT_CONCURRENCY: usize = 64; +/// Default concurrency limit for manifest fetching. +/// +/// Preload now runs N long-lived worker tasks; this is N. Each worker +/// processes one resolve_package at a time on tokio's global pool. +pub const DEFAULT_CONCURRENCY: usize = 256; /// A dependency spec: (name, version_spec) pub type Dep = (String, String); @@ -44,9 +59,6 @@ pub struct PreloadStats { pub success_count: usize, pub failed_count: usize, pub total_processed: usize, - pub min_request_ms: u64, - pub max_request_ms: u64, - pub total_request_ms: u64, } /// Collect dependencies from any deps map, filtering out non-registry specs. @@ -71,88 +83,157 @@ fn extract_transitive_deps(manifest: &CoreVersionManifest, config: &PreloadConfi deps } -/// Preload all package manifests in parallel with streaming concurrency. +/// Result message sent from worker to main task. +type Completion = ( + String, + Result>, +); + +/// Preload all package manifests in parallel via a tokio worker pool. +/// +/// `registry` is moved in and shared across workers via `Arc`. Each +/// worker is a long-lived spawned task that pulls work items from a +/// shared lock-free `SegQueue` until both the queue and the in-flight +/// counter reach zero, signalling end-of-phase. +/// +/// `receiver` and `on_manifest` run on the main task only — they do +/// not need to be `Send`/`Sync`. pub async fn preload_manifests( initial_deps: Vec, - registry: &R, + registry: Arc, config: PreloadConfig, receiver: &E, mut on_manifest: F, ) -> PreloadStats where - R: RegistryClient, + R: RegistryClient + Send + Sync + 'static, + R::Error: Send, E: EventReceiver, F: FnMut(&str, Arc), { let mut stats = PreloadStats::default(); - let mut processed: HashSet = HashSet::new(); - let mut pending: VecDeque = initial_deps.into(); - let concurrency = config.concurrency; + + // Shared work queue and dedup set. + let pending: Arc> = Arc::new(SegQueue::new()); + let processed: Arc> = Arc::new(DashSet::new()); + + // Counters for global termination — when `dispatched == completed` + // and the queue is empty, the phase is done. + let dispatched: Arc = Arc::new(AtomicUsize::new(0)); + let completed: Arc = Arc::new(AtomicUsize::new(0)); + let shutdown: Arc = Arc::new(AtomicBool::new(false)); + + // Wake-up signal for workers parked on an empty queue. + let notify: Arc = Arc::new(Notify::new()); + + // Result channel — workers send completions, main task drains. + let (result_tx, mut result_rx) = mpsc::unbounded_channel::>(); + + // Seed the queue with initial deps (deduped via DashSet). + for dep in initial_deps { + let key = format!("{}@{}", dep.0, dep.1); + if processed.insert(key) { + pending.push(dep); + dispatched.fetch_add(1, Ordering::Relaxed); + } + } + let initial_count = dispatched.load(Ordering::Relaxed); + let concurrency = config.concurrency.max(1); tracing::debug!( - "Preload: {} initial deps, concurrency={}", - pending.len(), + "Preload: {} initial deps, concurrency={}, mode=worker-pool", + initial_count, concurrency ); - let mut futures = FuturesUnordered::new(); - let mut in_flight = 0usize; - let mut started = false; - - loop { - // Fill up to concurrency limit - while in_flight < concurrency { - let item = loop { - let Some((name, spec)) = pending.pop_front() else { - break None; - }; - let key = format!("{}@{}", name, spec); - if !processed.contains(&key) { - processed.insert(key); - break Some((name, spec)); + // Spawn N long-lived workers. Each loops: pop -> resolve -> push transitives. + for _ in 0..concurrency { + let pending = Arc::clone(&pending); + let processed = Arc::clone(&processed); + let dispatched = Arc::clone(&dispatched); + let completed = Arc::clone(&completed); + let shutdown = Arc::clone(&shutdown); + let notify = Arc::clone(¬ify); + let result_tx = result_tx.clone(); + let registry = Arc::clone(®istry); + let config_for_worker = config.clone(); + + tokio::spawn(async move { + loop { + // Try fetching work first — fast path when queue is hot. + if let Some((name, spec)) = pending.pop() { + let result = resolve_package(&*registry, &name, &spec).await; + + if let Ok(resolved) = &result { + let mut new_added = 0usize; + for tdep in extract_transitive_deps(&resolved.manifest, &config_for_worker) + { + let key = format!("{}@{}", tdep.0, tdep.1); + if processed.insert(key) { + pending.push(tdep); + new_added += 1; + } + } + if new_added > 0 { + dispatched.fetch_add(new_added, Ordering::Release); + // Wake parked workers so they pick up the new work + // before checking the termination condition. + notify.notify_waiters(); + } + } + + if result_tx.send((name, result)).is_err() { + // Main task dropped the receiver — done collecting. + break; + } + let done = completed.fetch_add(1, Ordering::AcqRel) + 1; + + // After completion, check global done condition. The + // `Acquire` on dispatched pairs with the `Release` in + // the producer add above, so we won't miss late + // transitives queued by a sibling worker. + if done == dispatched.load(Ordering::Acquire) && pending.is_empty() { + shutdown.store(true, Ordering::Release); + notify.notify_waiters(); + break; + } + continue; } - }; - let Some((name, spec)) = item else { - break; - }; - - if !started { - receiver.on_event(BuildEvent::PreloadStart { count: 1 }); - started = true; - } else { - receiver.on_event(BuildEvent::PreloadQueued { count: 1 }); + // Queue empty — register interest before re-checking, then + // park. The Notify+enable() pattern guarantees we won't + // miss a `notify_waiters()` racing with our park. + if shutdown.load(Ordering::Acquire) { + break; + } + let notified = notify.notified(); + tokio::pin!(notified); + notified.as_mut().enable(); + if !pending.is_empty() { + continue; + } + if shutdown.load(Ordering::Acquire) { + break; + } + if completed.load(Ordering::Acquire) == dispatched.load(Ordering::Acquire) { + shutdown.store(true, Ordering::Release); + notify.notify_waiters(); + break; + } + notified.await; } + }); + } + // Drop the original sender so when all worker clones drop on exit, the + // result channel closes and the main loop terminates. + drop(result_tx); - receiver.on_event(BuildEvent::PreloadFetching { name: &name }); - - futures.push(async move { - let start = tokio::time::Instant::now(); - let result = resolve_package(registry, &name, &spec).await; - let elapsed_ms = start.elapsed().as_millis() as u64; - (name, result, elapsed_ms) - }); - in_flight += 1; - } - - if in_flight == 0 { - break; - } - - let Some((name, result, elapsed_ms)) = futures.next().await else { - break; - }; - in_flight -= 1; - - if stats.success_count == 0 && stats.failed_count == 0 { - stats.min_request_ms = elapsed_ms; - stats.max_request_ms = elapsed_ms; - } else { - stats.min_request_ms = stats.min_request_ms.min(elapsed_ms); - stats.max_request_ms = stats.max_request_ms.max(elapsed_ms); - } - stats.total_request_ms += elapsed_ms; + receiver.on_event(BuildEvent::PreloadStart { + count: initial_count, + }); + // Main task: drain completions, run user callbacks. + while let Some((name, result)) = result_rx.recv().await { match result { Ok(resolved) => { stats.success_count += 1; @@ -162,11 +243,8 @@ where version: &resolved.version, current: stats.success_count, }); - - // Send PackageResolved event for pipeline downloading receiver.on_event(BuildEvent::PackageResolved((&*resolved.manifest).into())); - pending.extend(extract_transitive_deps(&resolved.manifest, &config)); on_manifest(&name, resolved.manifest); } Err(e) => { @@ -183,33 +261,16 @@ where failed: stats.failed_count, }); - let total = stats.success_count + stats.failed_count; - let avg = if total > 0 { - stats.total_request_ms / total as u64 - } else { - 0 - }; - tracing::debug!( - "Preload stats: {} requests, min={}ms, max={}ms, avg={}ms, total={}ms", - total, - stats.min_request_ms, - stats.max_request_ms, - avg, - stats.total_request_ms - ); - stats } - #[cfg(test)] mod tests { use super::*; use crate::model::manifest::CoreVersionManifest; use crate::traits::progress::NoopReceiver; use crate::traits::registry::mock::MockRegistryClient; - use std::cell::RefCell; use std::collections::HashMap; - use std::rc::Rc; + use std::sync::Mutex; fn manifest(name: &str, version: &str) -> CoreVersionManifest { CoreVersionManifest { @@ -236,30 +297,30 @@ mod tests { } } - #[tokio::test] + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn test_preload_single() { let mut registry = MockRegistryClient::new(); registry.add_package("lodash", "4.17.21", manifest("lodash", "4.17.21")); - let cache: Rc>>> = Default::default(); - let cache_clone = Rc::clone(&cache); + let cache: Arc>>> = Default::default(); + let cache_clone = Arc::clone(&cache); let stats = preload_manifests( vec![("lodash".into(), "^4.17.0".into())], - ®istry, + Arc::new(registry), PreloadConfig::default(), &NoopReceiver, - |name, m| { - cache_clone.borrow_mut().insert(name.into(), m); + move |name, m| { + cache_clone.lock().unwrap().insert(name.into(), m); }, ) .await; assert_eq!(stats.success_count, 1); - assert!(cache.borrow().contains_key("lodash")); + assert!(cache.lock().unwrap().contains_key("lodash")); } - #[tokio::test] + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn test_preload_transitive() { let mut registry = MockRegistryClient::new(); registry.add_package( @@ -269,44 +330,44 @@ mod tests { ); registry.add_package("b", "1.0.0", manifest("b", "1.0.0")); - let cache: Rc>>> = Default::default(); - let cache_clone = Rc::clone(&cache); + let cache: Arc>>> = Default::default(); + let cache_clone = Arc::clone(&cache); let stats = preload_manifests( vec![("a".into(), "^1.0.0".into())], - ®istry, + Arc::new(registry), PreloadConfig::default(), &NoopReceiver, - |name, m| { - cache_clone.borrow_mut().insert(name.into(), m); + move |name, m| { + cache_clone.lock().unwrap().insert(name.into(), m); }, ) .await; assert_eq!(stats.success_count, 2); - assert!(cache.borrow().contains_key("a")); - assert!(cache.borrow().contains_key("b")); + assert!(cache.lock().unwrap().contains_key("a")); + assert!(cache.lock().unwrap().contains_key("b")); } - #[tokio::test] + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn test_preload_missing() { let registry = MockRegistryClient::new(); - let cache: Rc>>> = Default::default(); - let cache_clone = Rc::clone(&cache); + let cache: Arc>>> = Default::default(); + let cache_clone = Arc::clone(&cache); let stats = preload_manifests( vec![("nonexistent".into(), "^1.0.0".into())], - ®istry, + Arc::new(registry), PreloadConfig::default(), &NoopReceiver, - |name, m| { - cache_clone.borrow_mut().insert(name.into(), m); + move |name, m| { + cache_clone.lock().unwrap().insert(name.into(), m); }, ) .await; assert_eq!(stats.failed_count, 1); - assert!(cache.borrow().is_empty()); + assert!(cache.lock().unwrap().is_empty()); } #[test] diff --git a/crates/ruborist/src/resolver/registry.rs b/crates/ruborist/src/resolver/registry.rs index 145923160c..da9ce6ee85 100644 --- a/crates/ruborist/src/resolver/registry.rs +++ b/crates/ruborist/src/resolver/registry.rs @@ -13,7 +13,7 @@ use crate::model::manifest::FullManifest; use crate::model::node::EdgeType; use crate::resolver::semver::normalize_spec; use crate::resolver::version::resolve_target_version; -use crate::traits::registry::{RegistryClient, ResolvedPackage}; +use crate::traits::registry::{MaybeSync, RegistryClient, ResolvedPackage}; /// Error type for package resolution. #[derive(Debug)] @@ -114,7 +114,7 @@ impl std::error::Error for ResolveError { /// let resolved = resolve_package(®istry, "lodash", "^4.0.0").await?; /// println!("Resolved to {}@{}", resolved.name, resolved.version); /// ``` -pub async fn resolve_package( +pub async fn resolve_package( registry: &R, name: &str, spec: &str, @@ -165,7 +165,7 @@ pub fn resolve_from_manifest( /// /// For optional dependencies, returns `Ok(None)` on resolution failure /// instead of propagating the error. -pub async fn resolve_registry_dep( +pub async fn resolve_registry_dep( registry: &R, name: &str, spec: &str, diff --git a/crates/ruborist/src/resolver/semver.rs b/crates/ruborist/src/resolver/semver.rs index 6e81d5c10c..5380ca81cc 100644 --- a/crates/ruborist/src/resolver/semver.rs +++ b/crates/ruborist/src/resolver/semver.rs @@ -39,7 +39,12 @@ use deno_semver::{Version, VersionReq}; /// assert_eq!(name, "lodash"); /// assert_eq!(spec, "^4.0.0"); /// ``` -pub fn normalize_spec(name: &str, spec: &str) -> (String, String) { +pub fn normalize_spec<'a>( + name: &'a str, + spec: &'a str, +) -> (std::borrow::Cow<'a, str>, std::borrow::Cow<'a, str>) { + use std::borrow::Cow; + // Handle npm: alias - fetch the aliased package instead if let Some(npm_spec) = spec.strip_prefix("npm:") { // Skip "npm:" @@ -48,21 +53,26 @@ pub fn normalize_spec(name: &str, spec: &str) -> (String, String) { // Make sure we don't split on the @ of a scoped package if last_at_index > 0 { let (pkg_name, version) = npm_spec.split_at(last_at_index); - return (pkg_name.to_string(), version[1..].to_string()); + return (Cow::Borrowed(pkg_name), Cow::Borrowed(&version[1..])); } } // No version specified - return (npm_spec.to_string(), "*".to_string()); + return (Cow::Borrowed(npm_spec), Cow::Borrowed("*")); } // Handle workspace: prefix - keep original name, extract version if let Some(workspace_spec) = spec.strip_prefix("workspace:") { // Skip "workspace:" - return (name.to_string(), workspace_spec.to_string()); + return (Cow::Borrowed(name), Cow::Borrowed(workspace_spec)); } - // No special prefix, return as-is - (name.to_string(), spec.to_string()) + // No special prefix, return borrowed — the common case (~99 % of + // deps in a typical npm workload). Previously returned + // `(name.to_string(), spec.to_string())` which allocated two + // Strings per `resolve_package` call on the resolver hot path + // (~5460 allocs on ant-design preload, all from cooperative-poll + // future bodies that the main task drives). + (Cow::Borrowed(name), Cow::Borrowed(spec)) } /// Check if a version matches a semver range. @@ -208,51 +218,34 @@ mod tests { #[test] fn test_normalize_spec() { - // npm alias - assert_eq!( - normalize_spec("string-width-cjs", "npm:string-width@^4.2.0"), - ("string-width".to_string(), "^4.2.0".to_string()) - ); + let assert_eq_norm = |actual: (std::borrow::Cow<'_, str>, std::borrow::Cow<'_, str>), + expected: (&str, &str)| { + assert_eq!(actual.0.as_ref(), expected.0); + assert_eq!(actual.1.as_ref(), expected.1); + }; - // npm alias without version - assert_eq!( - normalize_spec("lodash-cjs", "npm:lodash"), - ("lodash".to_string(), "*".to_string()) + assert_eq_norm( + normalize_spec("string-width-cjs", "npm:string-width@^4.2.0"), + ("string-width", "^4.2.0"), ); - - // scoped npm alias - assert_eq!( + assert_eq_norm(normalize_spec("lodash-cjs", "npm:lodash"), ("lodash", "*")); + assert_eq_norm( normalize_spec("my-pkg", "npm:@scope/pkg@^1.0.0"), - ("@scope/pkg".to_string(), "^1.0.0".to_string()) + ("@scope/pkg", "^1.0.0"), ); - - // scoped npm alias without version - assert_eq!( + assert_eq_norm( normalize_spec("my-pkg", "npm:@scope/pkg"), - ("@scope/pkg".to_string(), "*".to_string()) + ("@scope/pkg", "*"), ); - - // workspace reference - assert_eq!( - normalize_spec("my-lib", "workspace:*"), - ("my-lib".to_string(), "*".to_string()) - ); - - assert_eq!( + assert_eq_norm(normalize_spec("my-lib", "workspace:*"), ("my-lib", "*")); + assert_eq_norm( normalize_spec("my-lib", "workspace:^1.0.0"), - ("my-lib".to_string(), "^1.0.0".to_string()) + ("my-lib", "^1.0.0"), ); - - // regular spec (unchanged) - assert_eq!( - normalize_spec("lodash", "^4.0.0"), - ("lodash".to_string(), "^4.0.0".to_string()) - ); - - // scoped package regular spec - assert_eq!( + assert_eq_norm(normalize_spec("lodash", "^4.0.0"), ("lodash", "^4.0.0")); + assert_eq_norm( normalize_spec("@types/node", "^18.0.0"), - ("@types/node".to_string(), "^18.0.0".to_string()) + ("@types/node", "^18.0.0"), ); } } diff --git a/crates/ruborist/src/resolver/workspace.rs b/crates/ruborist/src/resolver/workspace.rs index 6b550255e9..3e9b2cc715 100644 --- a/crates/ruborist/src/resolver/workspace.rs +++ b/crates/ruborist/src/resolver/workspace.rs @@ -68,42 +68,57 @@ impl WorkspaceDiscovery { None => return Ok(workspaces), }; + // Collect all workspace directories from every pattern first, then + // read every package.json concurrently. Ant-design has ~200 + // workspace packages; the previous serial `for path in + // matched_paths { read_package_json(...).await }` paid 200× + // single-file FS round-trips on the resolver hot path + // (~150-200 ms of sequential I/O during p1_resolve, the largest + // unmeasured chunk between preload completion and lockfile write). + let mut workspace_paths: Vec = Vec::new(); for pattern_str in patterns { - // Prepare glob pattern for package.json files let package_json_pattern = root_path.join(pattern_str).join("package.json"); - - // Match workspace directories using Glob::glob let matched_paths = self .glob .glob(&package_json_pattern) .await .map_err(|e| anyhow::anyhow!("Failed to glob pattern: {}", e))?; - for path in matched_paths { - let workspace_path = match path.parent() { - Some(p) => p.to_path_buf(), - None => continue, - }; - - // Read workspace package.json - let workspace_pkg: PackageJson = match read_package_json(&workspace_path).await { - Ok(pkg) => pkg, - Err(e) => { - tracing::debug!("Failed to read workspace package.json: {}", e); - continue; - } - }; - - tracing::debug!( - "Found workspace: {} at {:?}", - workspace_pkg.name, - workspace_path - ); - workspaces.push(WorkspacePackage { - name: workspace_pkg.name.clone(), - path: workspace_path, - package_json: workspace_pkg, - }); + if let Some(p) = path.parent() { + workspace_paths.push(p.to_path_buf()); + } + } + } + + // Fan out: dispatch every package.json read in parallel via + // `FuturesUnordered`. Each read is small (typical workspace + // package.json < 4 KB) so completion order doesn't matter. + use futures::stream::{FuturesUnordered, StreamExt}; + let mut futs: FuturesUnordered<_> = workspace_paths + .into_iter() + .map(|workspace_path| async move { + let pkg = read_package_json(&workspace_path).await; + (workspace_path, pkg) + }) + .collect(); + + while let Some((workspace_path, pkg)) = futs.next().await { + match pkg { + Ok(workspace_pkg) => { + tracing::debug!( + "Found workspace: {} at {:?}", + workspace_pkg.name, + workspace_path + ); + workspaces.push(WorkspacePackage { + name: workspace_pkg.name.clone(), + path: workspace_path, + package_json: workspace_pkg, + }); + } + Err(e) => { + tracing::debug!("Failed to read workspace package.json: {}", e); + } } } diff --git a/crates/ruborist/src/service/dns.rs b/crates/ruborist/src/service/dns.rs index ac61fec1eb..969530a2b0 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,59 @@ 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, keeping each address + /// family (IPv4 / IPv6) rotated independently and preserving the + /// system resolver's original family ordering between them. + /// + /// Why per-family: `getaddrinfo` typically returns IPv6 first + /// (e.g. 10 v6 + 12 v4 = 22 entries for `registry.npmjs.org`). A + /// flat rotation by `offset % 22` means offsets 0..10 all start + /// inside the IPv6 range and — on hosts without usable IPv6 + /// routing like GitHub Actions runners — every single one of them + /// falls through to the *same* first IPv4 address once Happy + /// Eyeballs gives up on v6. Pcap comparison with bun confirmed + /// this bug: 66 of utoo's 128 connections (51 %) were landing on + /// a single Cloudflare edge IP while the other 11 IPs saw only + /// 5-6 connections each. bun distributes cleanly 64 per IP across + /// 4 IPs, which is the behaviour we want. + /// + /// With per-family rotation, every new connection's first-reachable + /// IPv4 cycles through all v4 addresses (and same for v6 if it + /// works), giving ~11 conns per IP instead of 66 on one. + fn rotate_addrs(addrs: &[SocketAddr], offset: usize) -> Vec { + if addrs.is_empty() { + return Vec::new(); + } + let rotate = |slice: &[SocketAddr]| -> Vec { + if slice.is_empty() { + return Vec::new(); + } + let start = offset % slice.len(); + slice[start..] + .iter() + .chain(&slice[..start]) + .copied() + .collect() + }; + let v6: Vec = addrs.iter().filter(|a| a.is_ipv6()).copied().collect(); + let v4: Vec = addrs.iter().filter(|a| a.is_ipv4()).copied().collect(); + let v6_rot = rotate(&v6); + let v4_rot = rotate(&v4); + // Preserve v6-first ordering if that's what the resolver gave us; + // Happy Eyeballs will still prefer v6 when it's reachable. + let v6_first = addrs.first().map(|a| a.is_ipv6()).unwrap_or(true); + if v6_first { + v6_rot.into_iter().chain(v4_rot).collect() + } else { + v4_rot.into_iter().chain(v6_rot).collect() + } + } + /// Perform a system DNS lookup and cache the result. async fn do_lookup( hostname: &str, @@ -132,10 +188,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 +212,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 +225,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..4ef79eec15 100644 --- a/crates/ruborist/src/service/http.rs +++ b/crates/ruborist/src/service/http.rs @@ -76,19 +76,18 @@ //! WASM targets skip DNS entirely (browser handles it). use std::sync::LazyLock; -use std::time::Duration; use anyhow::{Context, Result, anyhow}; -/// Cuts hung TCP/TLS handshakes that would otherwise pin a conn-slot -/// indefinitely — `service::fetch`'s retry layer only fires on a reqwest -/// error, which silently-stalled sockets never raise. -const CONNECT_TIMEOUT: Duration = Duration::from_secs(5); - /// Global HTTP client with connection pooling and DNS caching. /// /// Stores `Result` so that proxy-configuration errors are /// surfaced to callers instead of panicking or calling `process::exit`. +/// +/// Multi-client experiments (2 and 4 isolated clients + preheat) both +/// regressed p1_resolve on CI — preheat TLS handshake cost exceeded the +/// phase-lock smoothing benefit, and uv (which uses a single client) +/// proves one pool is the right default. static HTTP_CLIENT: LazyLock> = LazyLock::new(|| { client_builder() .and_then(|b| b.build().context("Failed to build reqwest client")) @@ -99,24 +98,29 @@ pub(crate) fn get_client() -> Result<&'static reqwest::Client> { HTTP_CLIENT.as_ref().map_err(|e| anyhow!("{e}")) } -/// Override reqwest's default `ring` with `aws-lc-rs` on macOS. Local -/// M-series benchmarks show ~40 % faster cold resolves; CI bench-phases -/// regresses on Linux/x86_64, so non-macOS targets keep ring via -/// `rustls-tls-native-roots`. +/// Build a `rustls::ClientConfig` using the `aws-lc-rs` crypto provider +/// instead of reqwest's default `ring`. Measured on CI (4-core runner) +/// against npmjs.org, ring's per-TLS-handshake client-side CPU cost +/// (ECDHE key derivation + cert verification + Finished MAC) serialised +/// across 128 parallel handshakes into a 154 ms "CCS → first AppData" +/// span — the HTTP requests couldn't fire until all TLS crypto drained +/// through 4 async workers. aws-lc-rs uses BoringSSL's assembly-optimised +/// primitives and is roughly 3× faster at handshake work. #[cfg(target_os = "macos")] fn build_rustls_config() -> Result { - // `install_default` consumes the provider by value, so we can't share - // it with the `Arc` below — `default_provider()` is cheap (a struct of - // function pointers), so the duplicate call is fine. Idempotent across - // the process; only the first call wins. Sets the default for any - // other rustls consumer (e.g. `gix`). + // Install aws-lc-rs as the default for any other rustls consumer in + // the process. Idempotent — only the first call per process wins. let _ = rustls::crypto::aws_lc_rs::default_provider().install_default(); + // Load OS root certs. On Linux this hits /etc/ssl/certs/, + // on macOS it queries the Security framework keychain. Called once + // per process via `HTTP_CLIENT`'s `LazyLock`. let roots = rustls_native_certs::load_native_certs(); let mut root_store = rustls::RootCertStore::empty(); for cert in roots.certs { - // Tolerate individual bad roots so one broken cert in the OS - // trust store doesn't brick all requests. + // Best-effort: skip any cert rustls refuses (same tolerance + // native-tls shows). A hard fail here would brick every + // request on a box with one bad root in its trust store. let _ = root_store.add(cert); } if !roots.errors.is_empty() { @@ -130,7 +134,7 @@ fn build_rustls_config() -> Result { rustls::crypto::aws_lc_rs::default_provider(), )) .with_safe_default_protocol_versions() - .context("rustls safe_default_protocol_versions")? + .map_err(|e| anyhow!("rustls safe_default_protocol_versions: {e}"))? .with_root_certificates(root_store) .with_no_client_auth(); @@ -159,7 +163,16 @@ pub fn client_builder() -> Result { let mut builder = builder .no_proxy() .dns_resolver(shared_resolver()) - .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. + // Pool size matches `preload::DEFAULT_CONCURRENCY` so the + // per-host idle pool can absorb every in-flight fetch without + // churning connections. + .http1_only() + .pool_max_idle_per_host(256); #[cfg(target_os = "macos")] { diff --git a/crates/ruborist/src/service/registry.rs b/crates/ruborist/src/service/registry.rs index a978be50d8..028bff570c 100644 --- a/crates/ruborist/src/service/registry.rs +++ b/crates/ruborist/src/service/registry.rs @@ -369,9 +369,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(), @@ -432,7 +431,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,6 +439,14 @@ impl UnifiedRegistry { } let resolved_version = resolve_target_version((&*full).into(), spec) .map_err(|e| RegistryError(anyhow!("{}@{}: {}", name, spec, e)))?; + // Short-circuit: a sibling spec may have already resolved to + // this same version and warmed the cache via the dual-key + // write at the call-site. Skip the `get_core_version` reparse + // (the dominant CPU cost on resolve hot path) and share the + // existing `Arc`. + if let Some(cached) = self.cache.get_version_manifest(name, &resolved_version) { + return Ok((resolved_version, cached)); + } let core = full.get_core_version(&resolved_version).ok_or_else(|| { RegistryError(anyhow!( "Version {} not found in manifest for {}", @@ -447,7 +454,7 @@ impl UnifiedRegistry { name )) })?; - Ok((resolved_version, core)) + Ok((resolved_version, Arc::new(core))) } FullManifestResult::NotModified => { // 304 fallback: ETag matched, full payload not refetched. @@ -470,7 +477,7 @@ impl UnifiedRegistry { }) .await .map_err(RegistryError)?; - Ok((resolved_version, manifest)) + Ok((resolved_version, Arc::new(manifest))) } } } diff --git a/crates/ruborist/src/traits/registry.rs b/crates/ruborist/src/traits/registry.rs index 48e29ad920..f71cdd6e58 100644 --- a/crates/ruborist/src/traits/registry.rs +++ b/crates/ruborist/src/traits/registry.rs @@ -8,6 +8,28 @@ use crate::model::manifest::{CoreVersionManifest, FullManifest}; use crate::resolver::semver::normalize_spec; use crate::resolver::version::resolve_target_version; +// `Send` is required so the native preload worker-pool can move resolve +// futures onto `tokio::spawn`. `wasm-bindgen-futures` produce +// `Rc>`-backed futures that aren't `Send`/`Sync`, so on wasm we +// fall back to a no-op blanket impl. +#[cfg(not(target_arch = "wasm32"))] +pub trait MaybeSend: Send {} +#[cfg(not(target_arch = "wasm32"))] +impl MaybeSend for T {} +#[cfg(target_arch = "wasm32")] +pub trait MaybeSend {} +#[cfg(target_arch = "wasm32")] +impl MaybeSend for T {} + +#[cfg(not(target_arch = "wasm32"))] +pub trait MaybeSync: Sync {} +#[cfg(not(target_arch = "wasm32"))] +impl MaybeSync for T {} +#[cfg(target_arch = "wasm32")] +pub trait MaybeSync {} +#[cfg(target_arch = "wasm32")] +impl MaybeSync for T {} + /// Check if a registry URL is the official npm registry. /// /// The official npm registry (registry.npmjs.org/com) does not support: @@ -141,7 +163,7 @@ pub trait RegistryClient { fn fetch_full_manifest( &self, name: &str, - ) -> impl Future, Self::Error>>; + ) -> impl Future, Self::Error>> + MaybeSend; /// Fetch specific version manifest from registry. /// @@ -154,7 +176,10 @@ pub trait RegistryClient { &self, name: &str, spec: &str, - ) -> impl Future, Self::Error>> { + ) -> impl Future, Self::Error>> + MaybeSend + where + Self: MaybeSync, + { async move { let manifest = self.fetch_full_manifest(name).await?; @@ -193,7 +218,10 @@ pub trait RegistryClient { &self, name: &str, spec: &str, - ) -> impl Future> { + ) -> impl Future> + MaybeSend + where + Self: MaybeSync, + { async move { // Normalize spec (handles npm: alias and workspace: prefix) let (fetch_name, fetch_spec) = normalize_spec(name, spec); @@ -298,6 +326,7 @@ pub mod mock { use super::*; /// Internal package data for mock registry. + #[derive(Clone)] struct MockPackage { name: String, dist_tags: HashMap, @@ -305,6 +334,7 @@ pub mod mock { } /// Mock registry client that returns predefined packages. + #[derive(Clone)] pub struct MockRegistryClient { packages: HashMap, }