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/main.rs b/crates/pm/src/main.rs index 8f3a3a3662..2c72863c05 100644 --- a/crates/pm/src/main.rs +++ b/crates/pm/src/main.rs @@ -352,6 +352,14 @@ fn main() { let result = tokio::runtime::Builder::new_multi_thread() .enable_all() .worker_threads(worker_threads) + // Cap the blocking pool to reduce thread-creation thrash on + // runs that burst-dispatch ~128 spawn_blocking calls (JSON + // parse for manifest fetches). Default 512 leaves room for + // unbounded growth; ~worker_threads count is enough to + // saturate a 4-10 core machine on short CPU-bound tasks while + // keeping the pool warm. Default pool gave bimodal walls + // (2.7s fast / 6.9s thrash on M2); capping eliminates thrash. + .max_blocking_threads(worker_threads) .build() .expect("failed to build tokio runtime") .block_on(async_main()); 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 6ba124bb24..b4fd0de813 100644 --- a/crates/pm/src/util/logger.rs +++ b/crates/pm/src/util/logger.rs @@ -53,8 +53,23 @@ pub fn init_tracing(verbose: bool) -> Result<(PathBuf, WorkerGuard)> { let console_filter = EnvFilter::try_from_default_env() .unwrap_or_else(|_| EnvFilter::new(format!("utoo={console_level}"))); - // File filter: always capture debug+ for troubleshooting - let file_filter = EnvFilter::new("utoo=debug"); + // File filter: capture info+ by default. Hot-path debug!() calls + // (cache hits, per-package preload events, BFS dispatch) emit + // ~5-10 events per resolved manifest. With 2730+ manifests during + // a cold preload that's 15-30k events that — even routed through + // the non_blocking appender's channel — pay format/serialise CPU + // on the resolving thread. CI standalone manifest-bench reaches + // avg_conc=92 at cap=128 with the same reqwest stack; ruborist + // sat at avg_conc=56 after every other Mutex/clone hot-path was + // eliminated. Tracing was the last remaining shared cost. + // + // Override via `UTOO_FILE_LOG=debug` (or any RUST_LOG-style spec) + // to bring the troubleshooting captures back when actually + // diagnosing. + let file_filter = match env::var("UTOO_FILE_LOG") { + Ok(spec) if !spec.is_empty() => EnvFilter::new(spec), + _ => EnvFilter::new("utoo=info"), + }; // 2. Create log file let timestamp = SystemTime::now() diff --git a/crates/pm/src/util/user_config.rs b/crates/pm/src/util/user_config.rs index 34ee45a340..ef461cbb74 100644 --- a/crates/pm/src/util/user_config.rs +++ b/crates/pm/src/util/user_config.rs @@ -132,9 +132,23 @@ pub fn get_install_scope() -> InstallScope { INSTALL_SCOPE.get().copied().unwrap_or_default() } -// Manifest fetch concurrency configuration +// Manifest fetch concurrency configuration. +// +// Sweep history under worker-pool + HTTP/1.1 + DNS per-family rotation: +// cap=64 wall=3.10s avg_conc=56 (FuturesUnordered ceiling) +// cap=96 wall=2.87-3.08s ratio 0.83-1.03 (s9-s10 with H1 stack) +// cap=128 wall=2.15s avg_conc=84 (PR #2818 historical sweet spot, inline parse) +// cap=160 wall=2.14s avg_conc=119 (Cloudflare per-IP throttle bites) +// +// 96 was chosen empirically when burst aggression on H2 was triggering +// Cloudflare throttle (pre-H1 era). Once `.http1_only()` landed and +// requests stopped multiplexing through a single TCP connection, cap=128 +// becomes the right setting again — independent TCP streams distribute +// across the rotated DNS pool without tripping per-IP throttle, and the +// extra 33% concurrency saturates more of the Cloudflare connection +// budget. static MANIFESTS_CONCURRENCY_LIMIT: LazyLock> = - 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/lib.rs b/crates/ruborist/src/lib.rs index 0ea3b3b42f..2896da21b4 100644 --- a/crates/ruborist/src/lib.rs +++ b/crates/ruborist/src/lib.rs @@ -22,6 +22,7 @@ //! }).await?; //! ``` +pub mod maybe_send; pub mod model; pub mod resolver; pub mod service; diff --git a/crates/ruborist/src/maybe_send.rs b/crates/ruborist/src/maybe_send.rs new file mode 100644 index 0000000000..309f901ea0 --- /dev/null +++ b/crates/ruborist/src/maybe_send.rs @@ -0,0 +1,27 @@ +//! `Send`/`Sync` shim that is `Send`/`Sync` on native targets and +//! no-op on `wasm32`. Lets the resolver share a single trait surface +//! across multi-thread tokio (where futures must be `Send` to be +//! `tokio::spawn`-able) and wasm-bindgen (where `JsFuture` is `!Send` +//! and tasks run via `wasm_bindgen_futures::spawn_local`). + +/// `Send` on native, no-op on wasm32. +#[cfg(not(target_arch = "wasm32"))] +pub trait MaybeSend: Send {} +#[cfg(not(target_arch = "wasm32"))] +impl MaybeSend for T {} + +#[cfg(target_arch = "wasm32")] +pub trait MaybeSend {} +#[cfg(target_arch = "wasm32")] +impl MaybeSend for T {} + +/// `Sync` on native, no-op on wasm32. +#[cfg(not(target_arch = "wasm32"))] +pub trait MaybeSync: Sync {} +#[cfg(not(target_arch = "wasm32"))] +impl MaybeSync for T {} + +#[cfg(target_arch = "wasm32")] +pub trait MaybeSync {} +#[cfg(target_arch = "wasm32")] +impl MaybeSync for T {} diff --git a/crates/ruborist/src/resolver/builder.rs b/crates/ruborist/src/resolver/builder.rs index 53dca2826b..8d05ccb597 100644 --- a/crates/ruborist/src/resolver/builder.rs +++ b/crates/ruborist/src/resolver/builder.rs @@ -34,7 +34,7 @@ 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::{PreloadRegistry, 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. @@ -458,13 +458,16 @@ 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, edge_info: &DependencyEdgeInfo, config: &BuildDepsConfig, -) -> Result> { +) -> Result> +where + R::Error: crate::maybe_send::MaybeSend, +{ tracing::debug!( "Processing dependency {}@{} from [{:?}]", edge_info.name, @@ -707,11 +710,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: crate::maybe_send::MaybeSend, +{ let config = BuildDepsConfig::default().with_peer_deps(peer_deps); build_deps_with_config(graph, registry, config, &NoopReceiver).await } @@ -729,12 +735,15 @@ 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( graph: &mut DependencyGraph, registry: &R, peer_deps: PeerDeps, receiver: &E, -) -> Result<(), ResolveError> { +) -> Result<(), ResolveError> +where + R::Error: crate::maybe_send::MaybeSend, +{ let config = BuildDepsConfig::default().with_peer_deps(peer_deps); build_deps_with_config(graph, registry, config, receiver).await } @@ -759,12 +768,15 @@ 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( graph: &mut DependencyGraph, registry: &R, config: BuildDepsConfig, receiver: &E, -) -> Result<(), ResolveError> { +) -> Result<(), ResolveError> +where + R::Error: crate::maybe_send::MaybeSend, +{ tracing::debug!( "Starting dependency tree build, peer_deps: {:?}, concurrency: {}, skip_preload: {}", config.peer_deps, @@ -786,12 +798,14 @@ pub async fn build_deps_with_config( } /// Run the preload phase to warm up the cache with manifests. -async fn run_preload_phase( +async fn run_preload_phase( graph: &DependencyGraph, registry: &R, config: &BuildDepsConfig, receiver: &E, -) { +) where + R::Error: crate::maybe_send::MaybeSend, +{ if config.skip_preload { return; } @@ -804,18 +818,22 @@ async fn run_preload_phase( } tracing::debug!("Preload phase: {} initial dependencies", initial_deps.len()); - receiver.on_event(BuildEvent::PreloadStart { - count: initial_deps.len(), - }); + // PreloadStart / PreloadComplete are emitted inside `preload_manifests`. 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 as an independent `tokio::spawn` task on + // the multi-thread runtime. UnifiedRegistry's Clone is shallow + // (Arc-based), so this clone is cheap. + let registry_arc = std::sync::Arc::new(registry.clone()); + let stats = preload_manifests( initial_deps, - registry, + registry_arc, preload_config, receiver, |_name, _manifest| { @@ -829,21 +847,20 @@ async fn run_preload_phase( stats.success_count, stats.failed_count ); - 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, receiver: &E, -) -> Result<(), ResolveError> { +) -> Result<(), ResolveError> +where + R::Error: crate::maybe_send::MaybeSend, +{ let start = tokio::time::Instant::now(); let mut current_level = vec![graph.root_index]; @@ -961,10 +978,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: crate::maybe_send::MaybeSend, +{ resolve_with_options(pkg, registry, PeerDeps::Include, &NoopReceiver).await } @@ -975,12 +995,15 @@ 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( pkg: &PackageJson, registry: &R, peer_deps: PeerDeps, receiver: &E, -) -> Result> { +) -> Result> +where + R::Error: crate::maybe_send::MaybeSend, +{ // Create graph with root node let mut graph = DependencyGraph::from_package_json(PathBuf::from("."), pkg.clone()); diff --git a/crates/ruborist/src/resolver/preload.rs b/crates/ruborist/src/resolver/preload.rs index 9fb5aff67b..040ec86c3c 100644 --- a/crates/ruborist/src/resolver/preload.rs +++ b/crates/ruborist/src/resolver/preload.rs @@ -1,21 +1,43 @@ -//! Parallel manifest preloading for dependency resolution. +//! Parallel manifest preloading via multi-thread worker pool. //! -//! Uses FuturesUnordered for true streaming concurrency: when a package resolves, -//! its transitive dependencies are immediately added to the queue. +//! N long-lived `tokio::spawn` workers pull work items from a shared +//! lock-free `SegQueue`. Each spawned task is independent on tokio's +//! multi-thread runtime, so when one worker is parsing manifest JSON +//! (CPU-bound, simd_json), other workers can still drive their network +//! IO. Replaces the prior `FuturesUnordered` design where the main task +//! owned all preload futures and polled them cooperatively — every +//! per-future await continuation (cache check / OnceMap / RetryIf / +//! reqwest send / bytes / parse / transitive dispatch) ran on the same +//! single task, plus all parses serialised in that task's polling. +//! +//! Wasm32 fallback uses `wasm_bindgen_futures::spawn_local` since +//! `JsFuture` is `!Send`. Workers still run independently on the JS +//! event loop; the queue + Notify + mpsc termination story is identical. -use std::collections::{HashSet, VecDeque}; +use std::collections::HashSet; 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::maybe_send::{MaybeSend, MaybeSync}; 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; +use crate::traits::registry::{RegistryClient, ResolvedPackage}; -/// Default concurrency limit for manifest fetching -pub const DEFAULT_CONCURRENCY: usize = 64; +/// Default concurrency limit for manifest fetching. +/// +/// Number of long-lived `tokio::spawn` workers. Each processes one +/// `resolve_package` at a time on tokio's multi-thread runtime. +/// 128 matches the manifests-concurrency-limit sweet spot — high +/// enough to saturate independent npmjs connections, low enough to +/// stay under the Cloudflare per-source-IP throttle threshold. +pub const DEFAULT_CONCURRENCY: usize = 128; /// A dependency spec: (name, version_spec) pub type Dep = (String, String); @@ -71,79 +93,179 @@ 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: name, resolve result, +/// per-request elapsed ms, count of new transitives queued by this worker. +type Completion = (String, Result>, u64, usize); + +/// Preload all package manifests in parallel via a multi-thread 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. Each spawned task is +/// independent on tokio's multi-thread runtime — IO and CPU work +/// (simd_json parse) parallelise across cores. +/// +/// `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 + MaybeSend + MaybeSync + 'static, + R::Error: MaybeSend, 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; - - tracing::debug!( - "Preload: {} initial deps, concurrency={}", - pending.len(), - 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)); - } - }; + // Shared work queue and dedup set (lock-free, multi-thread safe). + let pending: Arc> = Arc::new(SegQueue::new()); + let processed: Arc> = Arc::new(DashSet::new()); - let Some((name, spec)) = item else { - break; - }; + // 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)); - if !started { - receiver.on_event(BuildEvent::PreloadStart { count: 1 }); - started = true; - } else { - receiver.on_event(BuildEvent::PreloadQueued { count: 1 }); - } + // Wake-up signal for workers parked on an empty queue. + let notify: Arc = Arc::new(Notify::new()); - receiver.on_event(BuildEvent::PreloadFetching { name: &name }); + // Result channel — workers send completions, main task drains. + let (result_tx, mut result_rx) = mpsc::unbounded_channel::>(); - 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; + // 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); - if in_flight == 0 { - break; - } + tracing::debug!( + "Preload: {} initial deps, concurrency={}, mode=mt-worker-pool", + initial_count, + concurrency + ); - let Some((name, result, elapsed_ms)) = futures.next().await else { - break; + // Short-circuit if nothing was seeded. + if initial_count == 0 { + receiver.on_event(BuildEvent::PreloadStart { count: 0 }); + receiver.on_event(BuildEvent::PreloadComplete { + success: 0, + failed: 0, + }); + return stats; + } + + 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 config_for_worker = config.clone(); + let registry = Arc::clone(®istry); + + let worker = async move { + loop { + // Try fetching work first — fast path when queue is hot. + if let Some((name, spec)) = pending.pop() { + let start = tokio::time::Instant::now(); + let result = resolve_package(&*registry, &name, &spec).await; + let elapsed_ms = start.elapsed().as_millis() as u64; + + let new_added = if let Ok(resolved) = &result { + let mut count = 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); + count += 1; + } + } + if count > 0 { + dispatched.fetch_add(count, Ordering::Release); + // Wake parked workers so they pick up the new + // work before checking the termination condition. + notify.notify_waiters(); + } + count + } else { + 0 + }; + + if result_tx + .send((name, result, elapsed_ms, new_added)) + .is_err() + { + // Main task dropped the receiver — done. + 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; + } + + // 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; + } }; - in_flight -= 1; + #[cfg(not(target_arch = "wasm32"))] + tokio::spawn(worker); + #[cfg(target_arch = "wasm32")] + wasm_bindgen_futures::spawn_local(worker); + } + // 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::PreloadStart { + count: initial_count, + }); + + // Main task: drain completions, run user callbacks. Receiver/callback + // run on this task only — they don't need to be Send/Sync. + while let Some((name, result, elapsed_ms, new_added)) = 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; @@ -153,6 +275,10 @@ where } stats.total_request_ms += elapsed_ms; + if new_added > 0 { + receiver.on_event(BuildEvent::PreloadQueued { count: new_added }); + } + match result { Ok(resolved) => { stats.success_count += 1; @@ -163,11 +289,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) => { @@ -177,7 +300,14 @@ where } } - stats.total_processed = processed.len(); + // Snapshot the dedup set's size for the historical observable. + stats.total_processed = { + let mut set: HashSet = HashSet::with_capacity(processed.len()); + for entry in processed.iter() { + set.insert(entry.key().clone()); + } + set.len() + }; receiver.on_event(BuildEvent::PreloadComplete { success: stats.success_count, @@ -208,9 +338,8 @@ mod tests { 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 +366,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 +399,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..19bbfc9ffe 100644 --- a/crates/ruborist/src/resolver/registry.rs +++ b/crates/ruborist/src/resolver/registry.rs @@ -114,11 +114,14 @@ 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, -) -> Result> { +) -> Result> +where + R::Error: crate::maybe_send::MaybeSend, +{ // Normalize spec first to handle npm: alias and workspace: prefix // This ensures correct behavior even if RegistryClient::resolve_package is overridden // e.g., "wrap-ansi-cjs" + "npm:wrap-ansi@^7.0.0" -> fetch "wrap-ansi" @ "^7.0.0" @@ -167,12 +170,15 @@ 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, edge_type: &EdgeType, -) -> Result, ResolveError> { +) -> Result, ResolveError> +where + R::Error: crate::maybe_send::MaybeSend, +{ match resolve_package(registry, name, spec).await { Ok(resolved) => Ok(Some(resolved)), Err(e) => { 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..de05559192 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,11 @@ mod native { cache: Arc>>, inflight: Arc>>>, ttl: Duration, + /// Round-robin counter incremented on every `resolve` call. Drives + /// per-family DNS rotation so successive connections distribute + /// across all returned Cloudflare edge IPs instead of all hitting + /// the first reachable v4 address. + rr_counter: AtomicUsize, } impl CachingResolver { @@ -79,10 +85,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 +187,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 +211,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 +224,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..0439d360ef 100644 --- a/crates/ruborist/src/service/http.rs +++ b/crates/ruborist/src/service/http.rs @@ -159,7 +159,17 @@ pub fn client_builder() -> Result { let mut builder = builder .no_proxy() .dns_resolver(shared_resolver()) - .connect_timeout(CONNECT_TIMEOUT); + .connect_timeout(CONNECT_TIMEOUT) + // Force HTTP/1.1. 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 is left at reqwest's unlimited default — + // upstream `MANIFESTS_CONCURRENCY_LIMIT` (128) already + // bounds the in-flight count, so an explicit pool ceiling + // is redundant. + .http1_only(); #[cfg(target_os = "macos")] { diff --git a/crates/ruborist/src/traits/registry.rs b/crates/ruborist/src/traits/registry.rs index df8adac00f..db7b700bce 100644 --- a/crates/ruborist/src/traits/registry.rs +++ b/crates/ruborist/src/traits/registry.rs @@ -141,7 +141,7 @@ pub trait RegistryClient { fn fetch_full_manifest( &self, name: &str, - ) -> impl Future, Self::Error>>; + ) -> impl Future, Self::Error>> + crate::maybe_send::MaybeSend; /// Fetch specific version manifest from registry. /// @@ -154,7 +154,11 @@ pub trait RegistryClient { &self, name: &str, spec: &str, - ) -> impl Future, Self::Error>> { + ) -> impl Future, Self::Error>> + crate::maybe_send::MaybeSend + where + Self: crate::maybe_send::MaybeSync, + Self::Error: crate::maybe_send::MaybeSend, + { async move { let manifest = self.fetch_full_manifest(name).await?; let version_list: Vec = manifest.versions.clone(); @@ -194,7 +198,11 @@ pub trait RegistryClient { &self, name: &str, spec: &str, - ) -> impl Future> { + ) -> impl Future> + crate::maybe_send::MaybeSend + where + Self: crate::maybe_send::MaybeSync, + Self::Error: crate::maybe_send::MaybeSend, + { async move { // Normalize spec (handles npm: alias and workspace: prefix) let (fetch_name, fetch_spec) = normalize_spec(name, spec); @@ -268,7 +276,11 @@ pub trait RegistryClient { fn fetch_versions_info( &self, name: &str, - ) -> impl Future> { + ) -> impl Future> + crate::maybe_send::MaybeSend + where + Self: crate::maybe_send::MaybeSync, + Self::Error: crate::maybe_send::MaybeSend, + { async move { let manifest = self.fetch_full_manifest(name).await?; Ok(VersionsInfo { @@ -294,12 +306,41 @@ pub trait RegistryClient { } } +/// Convenience bound for registries usable by the parallel preload worker pool. +/// +/// `Clone + 'static` lets callers share one owned handle across N spawned +/// workers (`Arc` on native, `Rc` on wasm). `MaybeSend + MaybeSync` +/// expand to `Send + Sync` on native (so the spawned worker future can be +/// scheduled across tokio's multi-thread runtime) and to no-op shims on +/// wasm (where `wasm_bindgen_futures::spawn_local` runs everything on the +/// JS event loop and `JsFuture` is `!Send`). Both `UnifiedRegistry` (real +/// client; `Clone` is shallow Arc-based) and `MockRegistryClient` (test +/// client) satisfy this trivially. +pub trait PreloadRegistry: + RegistryClient + Clone + crate::maybe_send::MaybeSend + crate::maybe_send::MaybeSync + 'static +where + Self::Error: crate::maybe_send::MaybeSend, +{ +} + +impl PreloadRegistry for T +where + T: RegistryClient + + Clone + + crate::maybe_send::MaybeSend + + crate::maybe_send::MaybeSync + + 'static, + T::Error: crate::maybe_send::MaybeSend, +{ +} + /// A simple in-memory registry client for testing. #[cfg(test)] pub mod mock { use super::*; /// Internal package data for mock registry. + #[derive(Clone)] struct MockPackage { name: String, dist_tags: HashMap, @@ -307,6 +348,7 @@ pub mod mock { } /// Mock registry client that returns predefined packages. + #[derive(Clone)] pub struct MockRegistryClient { packages: HashMap, }