diff --git a/Cargo.lock b/Cargo.lock index c228da3e01..7a64d38479 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -11794,6 +11794,7 @@ dependencies = [ "anyhow", "async-trait", "bytes", + "crossbeam-queue", "dashmap 6.1.0", "deno_semver", "dirs 5.0.1", diff --git a/crates/pm/src/helper/lock.rs b/crates/pm/src/helper/lock.rs index 98f5ebeba2..207facbc4c 100644 --- a/crates/pm/src/helper/lock.rs +++ b/crates/pm/src/helper/lock.rs @@ -481,17 +481,35 @@ pub async fn is_pkg_lock_outdated(root_path: &Path) -> Result { /// Save PackageLock to disk synchronously pub async fn save_package_lock(path: &Path, package_lock: &PackageLock) -> Result<()> { + let t_total = std::time::Instant::now(); let temp_path = path.join("package-lock.json.tmp"); let target_path = path.join("package-lock.json"); - // PackageLock now has all required fields (name, version, lockfile_version, requires, packages) + let t_serialize = std::time::Instant::now(); let content = serde_json::to_string_pretty(package_lock)?; + let serialize_ms = t_serialize.elapsed().as_millis(); + let bytes = content.len(); + + let t_write = std::time::Instant::now(); fs::write(&temp_path, content) .await .context("Failed to write temporary package-lock.json")?; + let write_ms = t_write.elapsed().as_millis(); + + let t_rename = std::time::Instant::now(); fs::rename(temp_path, target_path) .await .context("Failed to rename package-lock.json")?; + let rename_ms = t_rename.elapsed().as_millis(); + + tracing::info!( + "Save package-lock.json: total={}ms (serialize={}ms write={}ms rename={}ms, size={}KB)", + t_total.elapsed().as_millis(), + serialize_ms, + write_ms, + rename_ms, + bytes / 1024, + ); Ok(()) } 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/service/install.rs b/crates/pm/src/service/install.rs index 9f591ee85b..83131eda30 100644 --- a/crates/pm/src/service/install.rs +++ b/crates/pm/src/service/install.rs @@ -287,6 +287,18 @@ impl InstallService { let added = clone_count().saturating_sub(clone_baseline); let download_delta = download_stats() - download_baseline; print_install_counts(added, download_delta.reused, download_delta.downloaded); + + // Per-phase cumulative time (summed across concurrent tasks, not wall). + // Ratio to wall clock gives effective concurrency for that phase. + let phases = crate::util::downloader::phase_totals(); + let clone_us = crate::util::cloner::clone_micros(); + println!( + " download {:.2}s · extract {:.2}s · clone {:.2}s · {:.1} MB fetched", + phases.download_micros as f64 / 1e6, + phases.extract_micros as f64 / 1e6, + clone_us as f64 / 1e6, + phases.downloaded_bytes as f64 / (1024.0 * 1024.0), + ); Ok(()) } diff --git a/crates/pm/src/util/cloner.rs b/crates/pm/src/util/cloner.rs index 5d47c06eae..1fab6023fa 100644 --- a/crates/pm/src/util/cloner.rs +++ b/crates/pm/src/util/cloner.rs @@ -1,5 +1,6 @@ use std::path::{Path, PathBuf}; -use std::sync::atomic::{AtomicUsize, Ordering}; +use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering}; +use std::time::Instant; use anyhow::{Context, Result}; use once_cell::sync::Lazy; @@ -28,11 +29,19 @@ static CLONE_CACHE: Lazy> = Lazy::new(OnceMap::new); /// Mirrors pnpm's "added" semantic. static CLONE_COUNT: AtomicUsize = AtomicUsize::new(0); +/// Wall microseconds summed across every per-package clone invocation. +static CLONE_MICROS: AtomicU64 = AtomicU64::new(0); + /// Returns the number of fresh clones performed (pnpm "added" equivalent). pub fn clone_count() -> usize { CLONE_COUNT.load(Ordering::Relaxed) } +/// Cumulative wall microseconds spent in `clone_package` across the process. +pub fn clone_micros() -> u64 { + CLONE_MICROS.load(Ordering::Relaxed) +} + /// Normalize a target path into the canonical key used by `CLONE_CACHE`. #[cfg(windows)] fn cache_key(target_path: &Path) -> PathBuf { @@ -78,6 +87,7 @@ pub async fn clone_package_once( CLONE_CACHE .get_or_init(key, || async move { let cache_path = resolve_cache_path(&name, &version, &tarball_url).await?; + let t0 = Instant::now(); let fresh = clone_package(&cache_path, &target_path, &name, &version, !is_git) .await .inspect_err(|e| { @@ -90,6 +100,7 @@ pub async fn clone_package_once( ) }) .ok()?; + CLONE_MICROS.fetch_add(t0.elapsed().as_micros() as u64, Ordering::Relaxed); if fresh { CLONE_COUNT.fetch_add(1, Ordering::Relaxed); diff --git a/crates/pm/src/util/downloader.rs b/crates/pm/src/util/downloader.rs index 2f46a4058c..79e64f1bd8 100644 --- a/crates/pm/src/util/downloader.rs +++ b/crates/pm/src/util/downloader.rs @@ -1,6 +1,7 @@ use std::path::PathBuf; use std::sync::OnceLock; -use std::sync::atomic::{AtomicU32, AtomicUsize, Ordering}; +use std::sync::atomic::{AtomicU32, AtomicU64, AtomicUsize, Ordering}; +use std::time::Instant; use anyhow::{Context, Result}; use bytes::Bytes; @@ -30,6 +31,28 @@ static DOWNLOAD_SEMAPHORE: OnceLock = OnceLock::new(); static DOWNLOAD_COUNT: AtomicUsize = AtomicUsize::new(0); static REUSE_COUNT: AtomicUsize = AtomicUsize::new(0); +// Per-operation wall time summed across all packages. Sum is non-exclusive +// across cores — dividing by wall clock gives average concurrency for the phase. +static DOWNLOAD_MICROS: AtomicU64 = AtomicU64::new(0); +static EXTRACT_MICROS: AtomicU64 = AtomicU64::new(0); +static DOWNLOAD_BYTES: AtomicU64 = AtomicU64::new(0); + +/// Snapshot of per-phase totals accumulated during install. +#[derive(Debug, Clone, Copy, Default)] +pub struct PhaseTotals { + pub download_micros: u64, + pub extract_micros: u64, + pub downloaded_bytes: u64, +} + +pub fn phase_totals() -> PhaseTotals { + PhaseTotals { + download_micros: DOWNLOAD_MICROS.load(Ordering::Relaxed), + extract_micros: EXTRACT_MICROS.load(Ordering::Relaxed), + downloaded_bytes: DOWNLOAD_BYTES.load(Ordering::Relaxed), + } +} + /// Process-global counters for tarball outcomes, matching pnpm's /// vocabulary. Each unique `(name, version)` pair lands in exactly one /// bucket thanks to `DOWNLOAD_CACHE`'s `OnceMap` dedup; git/file/link @@ -181,16 +204,21 @@ pub async fn download_to_cache(name: &str, version: &str, tarball_url: &str) -> let semaphore = DOWNLOAD_SEMAPHORE .get_or_init(|| Semaphore::new(get_manifests_concurrency_limit_sync())); let _permit = semaphore.acquire().await.ok()?; + let t0 = Instant::now(); let bytes = download_bytes(&tarball_url) .await .inspect_err(|e| tracing::warn!("Download failed: {}@{}: {}", name, version, e)) .ok()?; + DOWNLOAD_MICROS.fetch_add(t0.elapsed().as_micros() as u64, Ordering::Relaxed); + DOWNLOAD_BYTES.fetch_add(bytes.len() as u64, Ordering::Relaxed); // Extract + let t1 = Instant::now(); extract_and_write(bytes, &cache_path) .await .inspect_err(|e| tracing::warn!("Extract failed: {}@{}: {}", name, version, e)) .ok()?; + EXTRACT_MICROS.fetch_add(t1.elapsed().as_micros() as u64, Ordering::Relaxed); DOWNLOAD_COUNT.fetch_add(1, Ordering::Relaxed); tracing::debug!("Downloaded: {}@{}", name, version); 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/logger.rs b/crates/pm/src/util/logger.rs index 757a81070b..7a0f4a846b 100644 --- a/crates/pm/src/util/logger.rs +++ b/crates/pm/src/util/logger.rs @@ -36,8 +36,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() @@ -188,15 +203,23 @@ impl utoo_ruborist::progress::EventReceiver for ProgressReceiver { fn on_event(&self, event: utoo_ruborist::progress::BuildEvent) { use utoo_ruborist::progress::BuildEvent; match event { - BuildEvent::PreloadStart { count } | BuildEvent::PreloadQueued { count } => { - PROGRESS_BAR.inc_length(count as u64); - } - BuildEvent::PreloadFetching { name } => { - log_progress(&format!("fetching {}", name)); - } - BuildEvent::PreloadProgress { name, .. } => { - PROGRESS_BAR.inc(1); - log_progress(&format!("resolved {}", name)); + BuildEvent::PreloadStart { .. } + | BuildEvent::PreloadQueued { .. } + | BuildEvent::PreloadFetching { .. } + | BuildEvent::PreloadProgress { .. } => { + // Hot path: 4571 dispatches + 4571 completions = ~9000 + // events landing on the main task during a 3-4 s phase. + // Every `PROGRESS_BAR.inc[_length]` takes indicatif's + // internal ProgressBar `Mutex`, contending with the + // steady_tick draw thread. Standalone manifest-bench + // sustains avg_conc=95 at cap=128; ruborist with these + // events stuck at 56 — the indicatif lock acquisitions + // cap the main loop's fill-and-drain rate. + // + // Drop all per-event progress bar updates during preload. + // The spinner still ticks via steady_tick so the user + // sees the phase is alive; the final summary prints + // success/fail counts at PreloadComplete. } BuildEvent::PreloadComplete { success, failed } => { PROGRESS_BAR.set_position(0); 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 53dca2826b..0ec4429dc4 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()), @@ -458,7 +467,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, @@ -707,11 +716,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 } @@ -729,12 +741,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 } @@ -759,12 +777,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, @@ -772,7 +796,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 @@ -786,12 +812,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; } @@ -804,18 +836,21 @@ async fn run_preload_phase( } tracing::debug!("Preload phase: {} initial dependencies", initial_deps.len()); - receiver.on_event(BuildEvent::PreloadStart { - count: initial_deps.len(), - }); + // Note: PreloadStart is emitted inside `preload_manifests` itself now. 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| { @@ -824,21 +859,22 @@ async fn run_preload_phase( ) .await; - tracing::debug!( - "Preload phase completed: {} success, {} failed", + let elapsed = start.elapsed(); + tracing::info!( + "Preload phase: {:.2?} ({} success, {} failed, {} initial deps)", + elapsed, stats.success_count, - stats.failed_count + stats.failed_count, + stats.total_processed, ); receiver.on_event(BuildEvent::PreloadComplete { success: stats.success_count, failed: stats.failed_count, }); - - tracing::debug!("Preload phase: {:?}", start.elapsed()); } /// 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, @@ -961,10 +997,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 } @@ -975,12 +1014,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/http.rs b/crates/ruborist/src/resolver/http.rs index 36cfd3580e..5bda3c0d57 100644 --- a/crates/ruborist/src/resolver/http.rs +++ b/crates/ruborist/src/resolver/http.rs @@ -98,7 +98,7 @@ use crate::model::manifest::CoreVersionManifest; use crate::service::fetch::{ FetchError, classify_reqwest_error, classify_status, is_retryable, retry_strategy, }; -use crate::service::http::get_client; +use crate::service::http::pick_client; use crate::traits::registry::ResolvedPackage; /// Session-scoped dedup cache: one fetch per URL even under concurrent BFS. @@ -153,7 +153,7 @@ async fn download_tarball(url: &str) -> Result { || { let url = url.to_string(); async move { - let resp = get_client() + let resp = pick_client() .map_err(FetchError::Permanent)? .get(&url) .send() 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 9fb5aff67b..da6f897b8e 100644 --- a/crates/ruborist/src/resolver/preload.rs +++ b/crates/ruborist/src/resolver/preload.rs @@ -1,21 +1,40 @@ -//! 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 std::time::Instant; -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::service::http::{ + finish_http_trace, finish_parse_trace, start_http_trace, start_parse_trace, +}; 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); @@ -71,79 +90,163 @@ 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>, + u64, +); + +/// 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; + let preload_wall_start = Instant::now(); + start_http_trace(); + start_parse_trace(); + + // 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 start = Instant::now(); + let result = resolve_package(&*registry, &name, &spec).await; + let elapsed_ms = start.elapsed().as_millis() as u64; + + 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, elapsed_ms)).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; + receiver.on_event(BuildEvent::PreloadStart { + count: initial_count, + }); + // Main task: drain completions, run user callbacks. + while let Some((name, result, elapsed_ms)) = result_rx.recv().await { if stats.success_count == 0 && stats.failed_count == 0 { stats.min_request_ms = elapsed_ms; stats.max_request_ms = elapsed_ms; @@ -163,11 +266,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) => { @@ -184,6 +284,12 @@ where failed: stats.failed_count, }); + let preload_wall_ms = preload_wall_start.elapsed().as_millis(); + let intervals = finish_http_trace(); + log_http_diagnostics(&intervals, preload_wall_ms); + let parses = finish_parse_trace(); + log_parse_diagnostics(&parses); + let total = stats.success_count + stats.failed_count; let avg = if total > 0 { stats.total_request_ms / total as u64 @@ -202,15 +308,156 @@ where stats } +/// Summarise captured HTTP intervals from the preload phase and log one +/// info-level line. Splits total preload wall into: +/// +/// - `wall` — first HTTP send → last HTTP body end (pure network window) +/// - `busy` — interval union (time at least one request was in-flight) +/// - `sum` — Σ per-request `(end − start)` (total req-level time) +/// - `cpu_tail` = preload_total − wall (our CPU work after last body) +/// - `avg_conc` = sum / busy (effective parallelism over busy window) +/// - percentiles of per-request latency +fn log_http_diagnostics(intervals: &[(Instant, Instant)], preload_wall_ms: u128) { + if intervals.is_empty() { + tracing::info!( + "Preload HTTP diag: no requests captured (total wall {}ms)", + preload_wall_ms + ); + return; + } + + let mut spans: Vec<(Instant, Instant)> = intervals.to_vec(); + spans.sort_by_key(|(s, _)| *s); + + let first_start = spans.first().unwrap().0; + let last_end = spans.iter().map(|(_, e)| *e).max().unwrap(); + let wall = last_end.duration_since(first_start).as_millis(); + + let sum: u128 = spans + .iter() + .map(|(s, e)| e.duration_since(*s).as_micros()) + .sum(); + + // Interval union: sweep sorted spans, merging overlaps. + let mut busy_us: u128 = 0; + let (mut cur_s, mut cur_e) = spans[0]; + for &(s, e) in &spans[1..] { + if s <= cur_e { + if e > cur_e { + cur_e = e; + } + } else { + busy_us += cur_e.duration_since(cur_s).as_micros(); + cur_s = s; + cur_e = e; + } + } + busy_us += cur_e.duration_since(cur_s).as_micros(); + + let mut per_req_us: Vec = spans + .iter() + .map(|(s, e)| e.duration_since(*s).as_micros()) + .collect(); + per_req_us.sort_unstable(); + let n = per_req_us.len(); + let p50 = per_req_us[n / 2]; + let p95 = per_req_us[(n * 95).div_ceil(100).saturating_sub(1)]; + let max = *per_req_us.last().unwrap(); + + let cpu_tail_ms = preload_wall_ms.saturating_sub(wall); + let avg_conc = if busy_us > 0 { + sum as f64 / busy_us as f64 + } else { + 0.0 + }; + + tracing::info!( + "Preload HTTP diag: n={} wall={}ms busy={}ms ({:.0}% of wall) sum={}ms avg_conc={:.1} p50={}ms p95={}ms max={}ms cpu_tail={}ms", + n, + wall, + busy_us / 1000, + if wall > 0 { + 100.0 * (busy_us as f64 / 1000.0) / wall as f64 + } else { + 0.0 + }, + sum / 1000, + avg_conc, + p50 / 1000, + p95 / 1000, + max / 1000, + cpu_tail_ms, + ); +} + +/// Summarise parse timing. +fn log_parse_diagnostics(parses: &[(Instant, Instant, Instant)]) { + if parses.is_empty() { + return; + } + + let mut queue_us: Vec = parses + .iter() + .map(|(q, s, _)| s.duration_since(*q).as_micros()) + .collect(); + let mut exec_us: Vec = parses + .iter() + .map(|(_, s, e)| e.duration_since(*s).as_micros()) + .collect(); + queue_us.sort_unstable(); + exec_us.sort_unstable(); + + let n = parses.len(); + let pct = |v: &[u128], p: usize| v[(n * p).div_ceil(100).saturating_sub(1)]; + let sum_queue: u128 = queue_us.iter().sum(); + let sum_exec: u128 = exec_us.iter().sum(); + + let mut spans: Vec<(Instant, Instant)> = parses.iter().map(|(_, s, e)| (*s, *e)).collect(); + spans.sort_by_key(|(s, _)| *s); + let (mut cur_s, mut cur_e) = spans[0]; + let mut exec_busy_us: u128 = 0; + for &(s, e) in &spans[1..] { + if s <= cur_e { + if e > cur_e { + cur_e = e; + } + } else { + exec_busy_us += cur_e.duration_since(cur_s).as_micros(); + cur_s = s; + cur_e = e; + } + } + exec_busy_us += cur_e.duration_since(cur_s).as_micros(); + let avg_exec_parallelism = if exec_busy_us > 0 { + sum_exec as f64 / exec_busy_us as f64 + } else { + 0.0 + }; + + tracing::info!( + "Preload parse diag: n={} queue(p50={}ms p95={}ms max={}ms sum={}ms) exec(p50={}ms p95={}ms max={}ms sum={}ms) exec_busy={}ms avg_parallel={:.1}", + n, + queue_us[n / 2] / 1000, + pct(&queue_us, 95) / 1000, + queue_us.last().unwrap() / 1000, + sum_queue / 1000, + exec_us[n / 2] / 1000, + pct(&exec_us, 95) / 1000, + exec_us.last().unwrap() / 1000, + sum_exec / 1000, + exec_busy_us / 1000, + avg_exec_parallelism, + ); +} + #[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 { @@ -237,30 +484,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( @@ -270,44 +517,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 6edeb9069c..807d3cf8f9 100644 --- a/crates/ruborist/src/resolver/registry.rs +++ b/crates/ruborist/src/resolver/registry.rs @@ -7,13 +7,13 @@ //! - For semver-supporting registries: directly fetches specific version //! - For traditional registries: fetches full manifest and resolves locally -use std::sync::Arc; - 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 std::sync::Arc; + +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, @@ -147,7 +147,6 @@ pub fn resolve_from_manifest( let resolved_version = resolve_target_version(&manifest.dist_tags, &version_list, spec) .map_err(|e| ResolveError::Version(format!("{}@{}: {}", manifest.name, spec, e)))?; - // Get manifest for resolved version (lazy: parse from Value on demand) let version_manifest = manifest .get_core_version(&resolved_version) .map(Arc::new) @@ -167,7 +166,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..7e52bd190a 100644 --- a/crates/ruborist/src/service/http.rs +++ b/crates/ruborist/src/service/http.rs @@ -76,47 +76,130 @@ //! WASM targets skip DNS entirely (browser handles it). use std::sync::LazyLock; -use std::time::Duration; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::time::Instant; use anyhow::{Context, Result, anyhow}; +use crossbeam_queue::SegQueue; -/// 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); +/// Diagnostic: per-HTTP-request `(send_start, body_end)` timestamps. +/// +/// When the flag is active, manifest fetch sites push `(Instant, Instant)` +/// pairs into the queue. Preload uses the collected intervals to compute +/// "pure network window", "busy window" (interval union), and per-request +/// stats — isolating network wait from our own CPU work. +/// +/// Flag is a single `AtomicBool` (relaxed) and only manifests as one +/// comparison + two `Instant::now()` per HTTP request when enabled; zero +/// cost on the disabled path. +static HTTP_TRACE_ACTIVE: AtomicBool = AtomicBool::new(false); +static HTTP_TRACE: LazyLock> = LazyLock::new(SegQueue::new); + +/// Activate per-request HTTP timing capture. Drains any prior trace. +pub fn start_http_trace() { + while HTTP_TRACE.pop().is_some() {} + HTTP_TRACE_ACTIVE.store(true, Ordering::Relaxed); +} + +/// Stop capturing and return the collected `(start, end)` intervals. +pub fn finish_http_trace() -> Vec<(Instant, Instant)> { + HTTP_TRACE_ACTIVE.store(false, Ordering::Relaxed); + let mut out = Vec::new(); + while let Some(v) = HTTP_TRACE.pop() { + out.push(v); + } + out +} + +/// Record one completed HTTP request's `(send_start, body_end)` timestamps. +/// No-op when the trace flag is off. Cheap relaxed-load check guards the +/// push so disabled callers pay almost nothing. +#[inline] +pub fn record_http_interval(start: Instant, end: Instant) { + if HTTP_TRACE_ACTIVE.load(Ordering::Relaxed) { + HTTP_TRACE.push((start, end)); + } +} + +/// Diagnostic: per-parse `(queued_at, exec_start, exec_end)` timestamps. +/// +/// `queued_at` is captured right before `spawn_blocking` is called; +/// `exec_start` is captured inside the closure (i.e. once the blocking +/// pool actually picks the task up). `queue_wait = exec_start − queued_at` +/// measures how long each parse sat idle in the blocking-pool queue. +/// When `queue_wait p50 ≫ exec p50` the pool is the bottleneck, and +/// `resolve_package` awaits stall the `FuturesUnordered` pipeline. +static PARSE_TRACE_ACTIVE: AtomicBool = AtomicBool::new(false); +static PARSE_TRACE: LazyLock> = LazyLock::new(SegQueue::new); + +pub fn start_parse_trace() { + while PARSE_TRACE.pop().is_some() {} + PARSE_TRACE_ACTIVE.store(true, Ordering::Relaxed); +} + +pub fn finish_parse_trace() -> Vec<(Instant, Instant, Instant)> { + PARSE_TRACE_ACTIVE.store(false, Ordering::Relaxed); + let mut out = Vec::new(); + while let Some(v) = PARSE_TRACE.pop() { + out.push(v); + } + out +} + +#[inline] +pub fn parse_trace_enabled() -> bool { + PARSE_TRACE_ACTIVE.load(Ordering::Relaxed) +} + +#[inline] +pub fn record_parse_interval(queued_at: Instant, exec_start: Instant, exec_end: Instant) { + if PARSE_TRACE_ACTIVE.load(Ordering::Relaxed) { + PARSE_TRACE.push((queued_at, exec_start, exec_end)); + } +} /// 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")) .map_err(|e| e.to_string()) }); -pub(crate) fn get_client() -> Result<&'static reqwest::Client> { +pub(crate) fn pick_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 +213,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 +242,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/manifest.rs b/crates/ruborist/src/service/manifest.rs index b9dbff7c8d..e31adc43b1 100644 --- a/crates/ruborist/src/service/manifest.rs +++ b/crates/ruborist/src/service/manifest.rs @@ -10,7 +10,7 @@ use tokio_retry::RetryIf; use super::fetch::{ FetchError, classify_reqwest_error, classify_status, is_retryable, retry_strategy, }; -use super::http::get_client; +use super::http::{parse_trace_enabled, pick_client, record_http_interval, record_parse_interval}; use crate::model::manifest::{CoreVersionManifest, FullManifest}; /// Result of a full manifest fetch with ETag support. @@ -60,7 +60,7 @@ pub async fn fetch_full_manifest(opts: FetchManifestOptions<'_>) -> Result) -> Result) -> Result = response .bytes() .await .map_err(|e| FetchError::Permanent(anyhow!("Response read error: {e}")))? - .to_vec(); - // Save raw bytes before simd_json mutates the parse buffer. + .into(); + record_http_interval(send_start, std::time::Instant::now()); + + // Inline parse on the async worker — simd_json is fast + // (1-5ms CPU per manifest) and the worker-pool preload + // architecture distributes 80+ concurrent fetches across + // tokio's worker_threads (= num_cpus). Earlier + // spawn_blocking offload backed up the 4-thread blocking + // pool: parse diag at cap=160 showed queue p95=200ms + // sum=70-89s, adding ~26ms per request — accounted for + // the entire ruborist-vs-standalone per-req gap (55ms vs + // 28ms). Inline parse trades a brief async-worker stall + // for zero queue wait, and worker-pool's independent + // task scheduling means one stalled worker doesn't + // starve the others. + let traced = parse_trace_enabled(); + let queued_at = traced.then(std::time::Instant::now); + let exec_start = queued_at.map(|_| std::time::Instant::now()); 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}")))?; manifest.raw = std::sync::Arc::from(raw_bytes); + if let (Some(q), Some(s)) = (queued_at, exec_start) { + record_parse_interval(q, s, std::time::Instant::now()); + } Ok(FetchManifestResult::Ok(manifest, new_etag)) } else { @@ -173,7 +196,8 @@ pub async fn fetch_version_manifest( || { let url = url.clone(); async move { - let response = get_client() + let send_start = std::time::Instant::now(); + let response = pick_client() .map_err(FetchError::Permanent)? .get(&url) .header("Accept", accept) @@ -182,13 +206,25 @@ pub async fn fetch_version_manifest( .map_err(classify_reqwest_error)?; if response.status().is_success() { - let mut bytes = response + let bytes: Vec = 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}"))) + .into(); + record_http_interval(send_start, std::time::Instant::now()); + // Inline parse — see fetch_full_manifest above for the + // rationale (blocking-pool queue saturation under + // worker-pool preload concurrency). + let traced = parse_trace_enabled(); + let queued_at = traced.then(std::time::Instant::now); + let exec_start = queued_at.map(|_| std::time::Instant::now()); + let mut buf = bytes; + let result = simd_json::serde::from_slice::(&mut buf) + .map_err(|e| FetchError::Permanent(anyhow!("JSON parse error: {e}"))); + if let (Some(q), Some(s)) = (queued_at, exec_start) { + record_parse_interval(q, s, std::time::Instant::now()); + } + result } else { Err(classify_status(response.status(), &url)) } diff --git a/crates/ruborist/src/traits/registry.rs b/crates/ruborist/src/traits/registry.rs index df8adac00f..033b91c31b 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?; let version_list: Vec = manifest.versions.clone(); @@ -194,7 +219,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); @@ -300,6 +328,7 @@ pub mod mock { use super::*; /// Internal package data for mock registry. + #[derive(Clone)] struct MockPackage { name: String, dist_tags: HashMap, @@ -307,6 +336,7 @@ pub mod mock { } /// Mock registry client that returns predefined packages. + #[derive(Clone)] pub struct MockRegistryClient { packages: HashMap, }