diff --git a/.github/actions/setup-utoo-next-baseline/action.yml b/.github/actions/setup-utoo-next-baseline/action.yml new file mode 100644 index 0000000000..9e8a82b1a6 --- /dev/null +++ b/.github/actions/setup-utoo-next-baseline/action.yml @@ -0,0 +1,23 @@ +name: setup-utoo-next-baseline +description: Download the utoo binary built from origin/next HEAD (by build-next-* jobs) into a temp directory and export $UTOO_NEXT_BIN. Used by bench-phases to add a `utoo-next` baseline column — the merged-but-not-published code — so PRs can show their delta vs the trunk baseline (not just vs the older npm-published version). + +inputs: + artifact: + description: Artifact name uploaded by the matching build-next-* job (e.g. utoo-next-linux-x64) + required: true + +runs: + using: composite + steps: + - name: Download utoo-next binary + uses: actions/download-artifact@v4 + with: + name: ${{ inputs.artifact }} + path: /tmp/utoo-next-pkg + - name: Set UTOO_NEXT_BIN + shell: bash + run: | + UTOO_NEXT_BIN="/tmp/utoo-next-pkg/utoo" + chmod +x "$UTOO_NEXT_BIN" + "$UTOO_NEXT_BIN" --version + echo "UTOO_NEXT_BIN=$UTOO_NEXT_BIN" >> $GITHUB_ENV diff --git a/.github/scripts/render-bench-phases-comment.sh b/.github/scripts/render-bench-phases-comment.sh index 7cc2340dad..a2ac74e211 100755 --- a/.github/scripts/render-bench-phases-comment.sh +++ b/.github/scripts/render-bench-phases-comment.sh @@ -35,7 +35,7 @@ mkdir -p /tmp/pm-bench-output echo "" echo "[Workflow run]($RUN_URL) — ant-design" echo "" - echo "_PMs: \`utoo\` (this branch) · \`utoo-npm\` (latest published) · \`bun\` (latest)_" + echo "_PMs: \`utoo\` (this branch) · \`utoo-next\` (next-branch baseline) · \`utoo-npm\` (latest published) · \`bun\` (latest)_" echo "" } > "$OUT" diff --git a/.github/workflows/pm-e2e-bench.yml b/.github/workflows/pm-e2e-bench.yml index a2a3f9c348..5744ad00d0 100644 --- a/.github/workflows/pm-e2e-bench.yml +++ b/.github/workflows/pm-e2e-bench.yml @@ -174,6 +174,89 @@ jobs: path: target/aarch64-apple-darwin/release/utoo retention-days: 1 + # ============================================================ + # NEXT-BRANCH BASELINE BUILDS + # Build utoo from origin/next HEAD so bench-phases can include a + # `utoo-next` column alongside `utoo` (this branch HEAD) — gives + # a clean A/B for the perf delta this PR contributes vs the merged + # baseline. Only fires when bench-phases will run. + # ============================================================ + + build-next-linux: + if: > + (github.event_name == 'pull_request' && contains(github.event.pull_request.labels.*.name, 'benchmark')) || + (github.event_name == 'workflow_dispatch' && inputs.target == 'pm-bench-phases') + runs-on: ubuntu-latest + container: + image: ubuntu:20.04 + steps: + - name: Install dependencies + timeout-minutes: 6 + env: + DEBIAN_FRONTEND: noninteractive + run: | + install_pkgs() { + timeout 60 apt-get update \ + && timeout 60 apt-get install -y build-essential curl pkg-config git + } + install_pkgs && exit 0 + dpkg --configure -a || true + sed -i 's|http://archive\.ubuntu\.com|http://mirrors.edge.kernel.org|g; s|http://security\.ubuntu\.com|http://mirrors.edge.kernel.org|g' /etc/apt/sources.list + install_pkgs + - uses: actions/checkout@v4 + with: + ref: next + - name: Configure git safe directory + run: git config --global --add safe.directory /__w/utoo/utoo + - name: Init git submodules + run: git submodule update --init --recursive --depth 1 + - name: Setup Rust toolchain + uses: dtolnay/rust-toolchain@stable + with: + toolchain: nightly-2026-04-02 + targets: x86_64-unknown-linux-gnu + - name: Cache cargo + uses: Swatinem/rust-cache@v2 + with: + shared-key: pm-build-next-linux + - name: Build utoo-pm from next + run: cargo build --release --target x86_64-unknown-linux-gnu --bin utoo -p utoo-pm + - name: Upload utoo-next binary + uses: actions/upload-artifact@v4 + with: + name: utoo-next-linux-x64 + path: target/x86_64-unknown-linux-gnu/release/utoo + retention-days: 1 + + build-next-mac-arm64: + if: > + (github.event_name == 'pull_request' && contains(github.event.pull_request.labels.*.name, 'benchmark')) || + (github.event_name == 'workflow_dispatch' && inputs.target == 'pm-bench-phases') + runs-on: macos-latest + steps: + - uses: actions/checkout@v4 + with: + ref: next + - name: Init git submodules + run: git submodule update --init --recursive --depth 1 + - name: Setup Rust toolchain + uses: dtolnay/rust-toolchain@stable + with: + toolchain: nightly-2026-04-02 + targets: aarch64-apple-darwin + - name: Cache cargo + uses: Swatinem/rust-cache@v2 + with: + shared-key: pm-build-next-mac-arm64 + - name: Build utoo-pm from next + run: cargo build --release --target aarch64-apple-darwin --bin utoo -p utoo-pm + - name: Upload utoo-next binary + uses: actions/upload-artifact@v4 + with: + name: utoo-next-mac-arm64 + path: target/aarch64-apple-darwin/release/utoo + retention-days: 1 + build-mac-x64: # Mac x86_64 serves: e2e-mac-x64 only. Cross-compiled on macos-latest # (arm64 host) via `rustup target add x86_64-apple-darwin`. @@ -441,7 +524,7 @@ jobs: # ============================================================ bench-phases-linux: - needs: [build-linux] + needs: [build-linux, build-next-linux] if: > (github.event_name == 'pull_request' && contains(github.event.pull_request.labels.*.name, 'benchmark')) || (github.event_name == 'workflow_dispatch' && inputs.target == 'pm-bench-phases') @@ -467,18 +550,22 @@ jobs: with: platform: linux - uses: ./.github/actions/install-utoo-npm-baseline + - uses: ./.github/actions/setup-utoo-next-baseline + with: + artifact: utoo-next-linux-x64 - name: Verify tools run: | hyperfine --version utoo --version "$UTOO_NPM_BIN" --version + "$UTOO_NEXT_BIN" --version bun --version - name: Define cache wipe function run: | cat > /tmp/wipe-bench-state.sh <<'WIPE' #!/usr/bin/env bash rm -rf /tmp/pm-bench /tmp/pm-bench-results /tmp/pm-bench-output \ - /tmp/utoo-bench-cache /tmp/utoo-npm-bench-cache /tmp/bun-bench-cache \ + /tmp/utoo-bench-cache /tmp/utoo-npm-bench-cache /tmp/utoo-next-bench-cache /tmp/bun-bench-cache \ ~/.bun/install/cache ~/.cache/nm 2>/dev/null || true mkdir -p /tmp/pm-bench-output WIPE @@ -490,7 +577,7 @@ jobs: PROJECT: ${{ github.event.inputs.project || 'ant-design' }} REGISTRY: 'https://registry.npmjs.org' BENCH_RUNS: ${{ github.event.inputs.bench_runs || '3' }} - PM_LIST: 'utoo,utoo-npm,bun' + PM_LIST: 'utoo,utoo-next,utoo-npm,bun' run: | chmod +x bench/pm-bench-phases.sh bash bench/pm-bench-phases.sh 2>&1 | tee /tmp/pm-bench-output/bench-phases-npmjs.log @@ -506,7 +593,7 @@ jobs: PROJECT: ${{ github.event.inputs.project || 'ant-design' }} REGISTRY: 'https://registry.npmmirror.com' BENCH_RUNS: ${{ github.event.inputs.bench_runs || '3' }} - PM_LIST: 'utoo,utoo-npm,bun' + PM_LIST: 'utoo,utoo-next,utoo-npm,bun' run: | mkdir -p /tmp/pm-bench-output bash bench/pm-bench-phases.sh 2>&1 | tee /tmp/pm-bench-output/bench-phases-npmmirror.log @@ -525,7 +612,7 @@ jobs: github-token: ${{ secrets.GITHUB_TOKEN }} bench-phases-mac: - needs: [build-mac-arm64] + needs: [build-mac-arm64, build-next-mac-arm64] if: > (github.event_name == 'pull_request' && contains(github.event.pull_request.labels.*.name, 'benchmark')) || (github.event_name == 'workflow_dispatch' && inputs.target == 'pm-bench-phases') @@ -551,18 +638,22 @@ jobs: with: platform: mac - uses: ./.github/actions/install-utoo-npm-baseline + - uses: ./.github/actions/setup-utoo-next-baseline + with: + artifact: utoo-next-mac-arm64 - name: Verify tools run: | hyperfine --version utoo --version "$UTOO_NPM_BIN" --version + "$UTOO_NEXT_BIN" --version bun --version - name: Define cache wipe function run: | cat > /tmp/wipe-bench-state.sh <<'WIPE' #!/usr/bin/env bash rm -rf /tmp/pm-bench /tmp/pm-bench-results /tmp/pm-bench-output \ - /tmp/utoo-bench-cache /tmp/utoo-npm-bench-cache /tmp/bun-bench-cache \ + /tmp/utoo-bench-cache /tmp/utoo-npm-bench-cache /tmp/utoo-next-bench-cache /tmp/bun-bench-cache \ ~/.bun/install/cache ~/.cache/nm 2>/dev/null || true mkdir -p /tmp/pm-bench-output WIPE @@ -574,7 +665,7 @@ jobs: PROJECT: ${{ github.event.inputs.project || 'ant-design' }} REGISTRY: 'https://registry.npmjs.org' BENCH_RUNS: ${{ github.event.inputs.bench_runs || '3' }} - PM_LIST: 'utoo,utoo-npm,bun' + PM_LIST: 'utoo,utoo-next,utoo-npm,bun' run: | chmod +x bench/pm-bench-phases.sh bash bench/pm-bench-phases.sh 2>&1 | tee /tmp/pm-bench-output/bench-phases-npmjs.log @@ -590,7 +681,7 @@ jobs: PROJECT: ${{ github.event.inputs.project || 'ant-design' }} REGISTRY: 'https://registry.npmmirror.com' BENCH_RUNS: ${{ github.event.inputs.bench_runs || '3' }} - PM_LIST: 'utoo,utoo-npm,bun' + PM_LIST: 'utoo,utoo-next,utoo-npm,bun' run: | mkdir -p /tmp/pm-bench-output bash bench/pm-bench-phases.sh 2>&1 | tee /tmp/pm-bench-output/bench-phases-npmmirror.log diff --git a/Cargo.lock b/Cargo.lock index 46d2141e2e..b691a1f46d 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -11346,6 +11346,8 @@ version = "0.1.0" dependencies = [ "anyhow", "bytes", + "crossbeam-queue", + "dashmap 6.1.0", "deno_semver", "dirs 5.0.1", "futures", @@ -11357,6 +11359,8 @@ dependencies = [ "pathdiff", "petgraph 0.6.5", "reqwest 0.12.24", + "rustls", + "rustls-native-certs", "serde", "serde_json", "sha2", @@ -11368,6 +11372,7 @@ dependencies = [ "tokio-fs-ext", "tokio-retry", "tracing", + "wasm-bindgen-futures", ] [[package]] diff --git a/bench/pm-bench-phases.sh b/bench/pm-bench-phases.sh index 4ca4cc4780..265f32d678 100755 --- a/bench/pm-bench-phases.sh +++ b/bench/pm-bench-phases.sh @@ -18,23 +18,27 @@ mkdir -p "$BENCH_DIR" "$RESULTS_DIR" # image, and previously-set user config). UTOO_CACHE="${UTOO_CACHE:-/tmp/utoo-bench-cache}" UTOO_NPM_CACHE="${UTOO_NPM_CACHE:-/tmp/utoo-npm-bench-cache}" +UTOO_NEXT_CACHE="${UTOO_NEXT_CACHE:-/tmp/utoo-next-bench-cache}" BUN_CACHE="${BUN_CACHE:-/tmp/bun-bench-cache}" export BUN_INSTALL_CACHE_DIR="$BUN_CACHE" -# Drop `utoo-npm` from the PM list when no published-baseline binary is -# wired up — UTOO_NPM_BIN is set by CI's "Install utoo@npm" step. Local -# runs without it just skip the comparison instead of erroring. -if [ -z "${UTOO_NPM_BIN:-}" ]; then - FILTERED=() - for pm in "${PACKAGE_MANAGERS[@]}"; do - if [ "$pm" = "utoo-npm" ]; then - echo "skip utoo-npm: UTOO_NPM_BIN not set" >&2 - continue - fi - FILTERED+=("$pm") - done - PACKAGE_MANAGERS=("${FILTERED[@]}") -fi +# Drop `utoo-npm` / `utoo-next` from the PM list when their binaries +# aren't wired up. CI sets UTOO_NPM_BIN (latest published) and +# UTOO_NEXT_BIN (next-branch HEAD); local runs without them just skip +# the comparison instead of erroring. +FILTERED=() +for pm in "${PACKAGE_MANAGERS[@]}"; do + if [ "$pm" = "utoo-npm" ] && [ -z "${UTOO_NPM_BIN:-}" ]; then + echo "skip utoo-npm: UTOO_NPM_BIN not set" >&2 + continue + fi + if [ "$pm" = "utoo-next" ] && [ -z "${UTOO_NEXT_BIN:-}" ]; then + echo "skip utoo-next: UTOO_NEXT_BIN not set" >&2 + continue + fi + FILTERED+=("$pm") +done +PACKAGE_MANAGERS=("${FILTERED[@]}") GREEN='\033[0;32m' YELLOW='\033[1;33m' @@ -122,9 +126,10 @@ capture_footprint() { local phase=$1 pm=$2 out=$3 local cache case "$pm" in - utoo) cache=$UTOO_CACHE ;; - utoo-npm) cache=$UTOO_NPM_CACHE ;; - bun) cache=$BUN_CACHE ;; + utoo) cache=$UTOO_CACHE ;; + utoo-npm) cache=$UTOO_NPM_CACHE ;; + utoo-next) cache=$UTOO_NEXT_CACHE ;; + bun) cache=$BUN_CACHE ;; esac printf '{"cache":%d,"node_modules":%d,"lockfile":%d}\n' \ "$(du_bytes "$cache")" \ @@ -148,9 +153,10 @@ write_prepare() { local path=$1 phase=$2 pm=$3 local cache case "$pm" in - utoo) cache=$UTOO_CACHE ;; - utoo-npm) cache=$UTOO_NPM_CACHE ;; - bun) cache=$BUN_CACHE ;; + utoo) cache=$UTOO_CACHE ;; + utoo-npm) cache=$UTOO_NPM_CACHE ;; + utoo-next) cache=$UTOO_NEXT_CACHE ;; + bun) cache=$BUN_CACHE ;; esac cat > "$path" <> "$path" <> "$path" <> "$path" <> "$path" <> "$path" <> "$path" <> "$path" < "$RESULTS_DIR/seed_${phase}_${pm}.log" 2>&1 fi ;; + p3_*:utoo-next|p4_*:utoo-next) + if [ ! -f package-lock.json ]; then + echo -e " ${CYAN}seed: running \`utoo-next deps\` to generate package-lock.json${NC}" + "$UTOO_NEXT_BIN" deps --registry="$REGISTRY" --cache-dir="$UTOO_NEXT_CACHE" > "$RESULTS_DIR/seed_${phase}_${pm}.log" 2>&1 + fi + ;; p3_*:bun|p4_*:bun) if [ ! -f bun.lock ]; then echo -e " ${CYAN}seed: running \`bun install --lockfile-only\` to generate bun.lock${NC}" @@ -264,17 +287,19 @@ seed_for_phase() { if [[ "$phase" == p4_* ]]; then local cache case "$pm" in - utoo) cache=$UTOO_CACHE ;; - utoo-npm) cache=$UTOO_NPM_CACHE ;; - bun) cache=$BUN_CACHE ;; + utoo) cache=$UTOO_CACHE ;; + utoo-npm) cache=$UTOO_NPM_CACHE ;; + utoo-next) cache=$UTOO_NEXT_CACHE ;; + bun) cache=$BUN_CACHE ;; esac if [ ! -d "$cache" ] || [ -z "$(ls -A "$cache" 2>/dev/null)" ]; then echo -e " ${CYAN}seed: warming $pm cache via full install${NC}" rm -rf node_modules case "$pm" in - utoo) utoo install --ignore-scripts --registry="$REGISTRY" --cache-dir="$UTOO_CACHE" > "$RESULTS_DIR/seed_warmup_${pm}.log" 2>&1 ;; - utoo-npm) "$UTOO_NPM_BIN" install --ignore-scripts --registry="$REGISTRY" --cache-dir="$UTOO_NPM_CACHE" > "$RESULTS_DIR/seed_warmup_${pm}.log" 2>&1 ;; - bun) bun install --ignore-scripts --registry="$REGISTRY" > "$RESULTS_DIR/seed_warmup_${pm}.log" 2>&1 ;; + utoo) utoo install --ignore-scripts --registry="$REGISTRY" --cache-dir="$UTOO_CACHE" > "$RESULTS_DIR/seed_warmup_${pm}.log" 2>&1 ;; + utoo-npm) "$UTOO_NPM_BIN" install --ignore-scripts --registry="$REGISTRY" --cache-dir="$UTOO_NPM_CACHE" > "$RESULTS_DIR/seed_warmup_${pm}.log" 2>&1 ;; + utoo-next) "$UTOO_NEXT_BIN" install --ignore-scripts --registry="$REGISTRY" --cache-dir="$UTOO_NEXT_CACHE" > "$RESULTS_DIR/seed_warmup_${pm}.log" 2>&1 ;; + bun) bun install --ignore-scripts --registry="$REGISTRY" > "$RESULTS_DIR/seed_warmup_${pm}.log" 2>&1 ;; esac rm -rf node_modules fi @@ -283,17 +308,19 @@ seed_for_phase() { install_cmd() { case "$1" in - utoo) echo "utoo install --ignore-scripts --registry=$REGISTRY --cache-dir=$UTOO_CACHE" ;; - utoo-npm) echo "$UTOO_NPM_BIN install --ignore-scripts --registry=$REGISTRY --cache-dir=$UTOO_NPM_CACHE" ;; - bun) echo "bun install --ignore-scripts --registry=$REGISTRY" ;; + utoo) echo "utoo install --ignore-scripts --registry=$REGISTRY --cache-dir=$UTOO_CACHE" ;; + utoo-npm) echo "$UTOO_NPM_BIN install --ignore-scripts --registry=$REGISTRY --cache-dir=$UTOO_NPM_CACHE" ;; + utoo-next) echo "$UTOO_NEXT_BIN install --ignore-scripts --registry=$REGISTRY --cache-dir=$UTOO_NEXT_CACHE" ;; + bun) echo "bun install --ignore-scripts --registry=$REGISTRY" ;; esac } resolve_cmd() { case "$1" in - utoo) echo "utoo deps --registry=$REGISTRY --cache-dir=$UTOO_CACHE" ;; - utoo-npm) echo "$UTOO_NPM_BIN deps --registry=$REGISTRY --cache-dir=$UTOO_NPM_CACHE" ;; - bun) echo "bun install --lockfile-only --registry=$REGISTRY" ;; + utoo) echo "utoo deps --registry=$REGISTRY --cache-dir=$UTOO_CACHE" ;; + utoo-npm) echo "$UTOO_NPM_BIN deps --registry=$REGISTRY --cache-dir=$UTOO_NPM_CACHE" ;; + utoo-next) echo "$UTOO_NEXT_BIN deps --registry=$REGISTRY --cache-dir=$UTOO_NEXT_CACHE" ;; + bun) echo "bun install --lockfile-only --registry=$REGISTRY" ;; esac } diff --git a/crates/pm/src/cmd/view.rs b/crates/pm/src/cmd/view.rs index d8355e9922..282ee46cee 100644 --- a/crates/pm/src/cmd/view.rs +++ b/crates/pm/src/cmd/view.rs @@ -27,7 +27,7 @@ pub async fn view(package_spec: &str) -> Result<()> { tracing::debug!("Fetched package info: {full_manifest:?}"); // Resolve version and get full VersionManifest (with all display fields) - let version_list: Vec = full_manifest.versions.clone(); + let version_list: Vec = full_manifest.versions.keys.clone(); let resolved_version = utoo_ruborist::resolver::version::resolve_target_version( &full_manifest.dist_tags, &version_list, @@ -94,7 +94,7 @@ fn print_package_header(full_manifest: &FullManifest, version_manifest: &Version .as_ref() .map(|d| d.len()) .unwrap_or(0); - let versions_count = full_manifest.versions.len(); + let versions_count = full_manifest.versions.keys.len(); let deps_str = if deps_count == 0 { "none".to_string() 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/ruborist/Cargo.toml b/crates/ruborist/Cargo.toml index 4f63f3ec7f..c22059c5e1 100644 --- a/crates/ruborist/Cargo.toml +++ b/crates/ruborist/Cargo.toml @@ -7,57 +7,76 @@ license = "MIT" # tombi: format.rules.table-keys-order.disabled = true [dependencies] -anyhow = { workspace = true } -bytes = "1" -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 } +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 } # HTTP client - works on both native and WASM -reqwest = { version = "0.12", default-features = false, features = [ +# +# We use `rustls-tls-native-roots-no-provider` (instead of the default +# `rustls-tls-native-roots`) so reqwest doesn't pin the crypto provider +# to `ring`. We then build our own `rustls::ClientConfig` with +# `aws-lc-rs` (BoringSSL's crypto primitives) in `service/http.rs` and +# pass it via `Client::use_preconfigured_tls`. Measured on CI against +# npmjs.org: `ring` took 78 ms p50 / 154 ms max from CCS → first +# AppData per handshake (CPU-bound; 128 parallel handshakes serialise +# across 4 cores). aws-lc-rs's per-handshake CPU is roughly 3× faster. +reqwest = { version = "0.12", default-features = false, features = [ + "brotli", + "gzip", "http2", "json", - "rustls-tls", - "rustls-tls-native-roots", + "rustls-tls-native-roots-no-provider", "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-only dependencies (not compiled for WASM) [target.'cfg(not(target_arch = "wasm32"))'.dependencies] -libc = "0.2" +libc = "0.2" +# Our own rustls + aws-lc-rs crypto provider — native-only because +# `aws-lc-sys` builds BoringSSL via a `cc` C build that doesn't +# support `wasm32-unknown-unknown`. The wasm reqwest path uses the +# browser fetch API which ignores rustls anyway. +rustls = { version = "0.23", default-features = false, features = ["aws-lc-rs", "logging", "std", "tls12"] } +rustls-native-certs = "0.8" # WASM-only dependencies [target.'cfg(target_arch = "wasm32")'.dependencies] -js-sys = "0.3" # For time in WASM +js-sys = "0.3" # For time in WASM +wasm-bindgen-futures = "0.4" # spawn_local for fire-and-forget disk writes [features] default = [] diff --git a/crates/ruborist/src/lib.rs b/crates/ruborist/src/lib.rs index 890d6c2798..b7d0753071 100644 --- a/crates/ruborist/src/lib.rs +++ b/crates/ruborist/src/lib.rs @@ -101,6 +101,4 @@ pub mod http { } /// Utility functions. -pub mod util { - pub use crate::model::util::{PackageNameStr, parse_package_spec, read_package_json}; -} +pub mod util; diff --git a/crates/ruborist/src/model/manifest.rs b/crates/ruborist/src/model/manifest.rs index 6ec7495bad..16be3c3787 100644 --- a/crates/ruborist/src/model/manifest.rs +++ b/crates/ruborist/src/model/manifest.rs @@ -37,10 +37,24 @@ pub struct FullManifest { #[serde(rename = "dist-tags")] pub dist_tags: HashMap, - #[serde(default, deserialize_with = "deserialize_version_keys")] - pub versions: Vec, + /// Version keys (preserved order) and pre-parsed `CoreVersionManifest`s, + /// populated in a single pass by [`Versions`]'s custom `Deserialize`. + /// + /// Replaces the previous `versions: Vec` (populated by + /// `IgnoredAny` visitor) + per-resolve `extract_version` that + /// re-parsed the full raw JSON via `simd_json::to_borrowed_value` + /// for every `resolve_package` call. That re-parse was uninstrumented + /// CPU on the async worker — ~0.4ms avg × 4567 resolves = 1.85s + /// serial load on the 4-core tokio runtime, preventing the pipeline + /// from maintaining 64 in-flight fetches (observed ~38 effective). + /// + /// With eager parse, `get_core_version` becomes an O(1) map lookup. + pub versions: Versions, - /// Raw HTTP response bytes. Injected post-parse; used for on-demand version extraction. + /// Raw HTTP response bytes. Retained only for `get_full_version` + /// (cold path — `ut view` command) which still needs on-demand + /// extraction of the full `VersionManifest`. Hot-path resolve uses + /// [`Self::versions`] which is pre-parsed. #[serde(skip)] pub raw: Arc<[u8]>, @@ -76,11 +90,61 @@ pub struct FullManifest { } impl FullManifest { - /// Extract a single version from raw bytes on demand. + /// On-demand `CoreVersionManifest` extraction from a pre-parsed + /// `simd_json::OwnedValue` subtree. + /// + /// History: we've been through three designs here. + /// 1. **Eager struct parse (previous)**: `Versions::deserialize` + /// built `Arc` for every version up + /// front. Lookup was O(1) but fetch-time spawn_blocking had to + /// materialise ~500 structs per manifest × 2730 manifests = + /// ~1.37M CoreVersionManifest builds, 99 % of which the + /// resolver never reads. Blocking pool saturation at cap=128 + /// showed up as `parse_us` avg 20 ms with long tails. + /// 2. **Lazy `Arc` (tried in c5cd8318, + /// reverted fe8365d7)**: stored `serde_json::Value` subtrees + /// and called `from_value(value.clone())` on demand. The + /// deep clone + serde_json walk ran on async worker and + /// measured `core_version_us` 11-18 ms avg — worse than eager. + /// 3. **This design — lazy simd_json `OwnedValue` + memoisation**: + /// Store each version as `Arc` (the + /// native simd_json tree form — no serde_json round-trip). + /// On demand call `from_refvalue(&OwnedValue)` which + /// zero-copies through `&Value` implementing `Deserializer`. + /// First hit for a given (manifest, version) pays ~50-200 μs + /// of subtree walking; subsequent hits are `Arc::clone` via + /// the `DashMap` memoisation cache. /// - /// Copies raw bytes, parses with `simd_json::to_borrowed_value`, navigates - /// to `versions[ver]`, then converts to the target type via `serde_json::Value`. - fn extract_version Deserialize<'de>>(&self, version: &str) -> Option { + /// Why this is better than design 1: a typical resolve touches + /// 1-3 of a manifest's ~500 versions. Building the other 497+ + /// was pure waste on the blocking pool's critical path. + /// + /// Why this is better than design 2: `simd_json::OwnedValue`'s + /// `&Value: Deserializer` zero-copies through the tree without + /// allocating an intermediate `serde_json::Value`. And moving + /// the conversion out of `spawn_blocking` (it's small, runs on + /// async worker) removes the 200 μs dispatch overhead that was + /// ~500 ms of the previous lazy attempt. + pub fn get_core_version(&self, version: &str) -> Option> { + // Fast path: memoised conversion. + if let Some(cached) = self.versions.cache.get(version) { + return Some(cached.clone()); + } + let tree = self.versions.trees.get(version)?; + // `&simd_json::OwnedValue` impls `Deserializer<'_>` directly + // (see simd_json::serde::value::owned::de), so this is a + // zero-allocation tree walk — no Value clone, no bytes copy. + let core = CoreVersionManifest::deserialize(tree.as_ref()).ok()?; + let arc = Arc::new(core); + self.versions.cache.insert(version.to_string(), arc.clone()); + Some(arc) + } + + /// Parse a single version on demand into full `VersionManifest` + /// (cold path — `ut view`). Uses the retained raw bytes because + /// `VersionManifest` carries display fields that `CoreVersionManifest` + /// drops, and `ut view` is infrequent. + pub fn get_full_version(&self, version: &str) -> Option { use simd_json::prelude::ValueObjectAccess; let mut buf = self.raw.to_vec(); let parsed = simd_json::to_borrowed_value(&mut buf).ok()?; @@ -88,45 +152,95 @@ impl FullManifest { let value = serde_json::to_value(version_obj).ok()?; serde_json::from_value(value).ok() } +} - /// Parse a single version on demand into CoreVersionManifest (hot path). - pub fn get_core_version(&self, version: &str) -> Option { - self.extract_version(version) - } +/// Version entries of a `FullManifest`: preserves key insertion order +/// (for `VersionsInfo`/semver callers that iterate the list) alongside +/// per-version pre-parsed JSON subtrees and a memoisation cache for +/// strongly-typed `CoreVersionManifest` conversions. +/// +/// Fetch-time parse builds only `Arc` per +/// version — cheaper than constructing a full `CoreVersionManifest` +/// because `OwnedValue` is the native simd_json tree without +/// field-by-field `#[serde(default)]` / `skip_on_error` validation. +/// The real `CoreVersionManifest` is materialised on demand inside +/// `FullManifest::get_core_version`, so the ~99 % of versions the +/// resolver never touches don't pay for field-level parsing. +#[derive(Debug, Clone, Default)] +pub struct Versions { + /// Version strings in insertion order — used by + /// `resolve_target_version` and cached into `VersionsInfo`. + pub keys: Vec, + /// Per-version pre-parsed JSON subtrees. `Arc` because the + /// `FullManifest` is cloned across cache tiers (memory / project + /// cache) and we want Arc-level sharing of the underlying bytes. + pub trees: HashMap>, + /// Memoisation cache populated on first `get_core_version(v)` + /// call. `DashMap` instead of `Mutex` because + /// multiple resolves for the same (name, version) can land + /// concurrently on different async workers; DashMap avoids + /// serialising them on a global mutex. + pub cache: Arc>>, +} - /// Parse a single version on demand into full VersionManifest (cold path, e.g. `ut view`). - pub fn get_full_version(&self, version: &str) -> Option { - self.extract_version(version) +impl Versions { + pub fn is_empty(&self) -> bool { + self.keys.is_empty() } } -/// Deserialize a versions map by extracting only the keys, skipping all values. -/// -/// Uses `IgnoredAny` to skip over version manifest JSON objects without allocating, -/// only collecting the version number strings. -fn deserialize_version_keys<'de, D>(deserializer: D) -> Result, D::Error> -where - D: Deserializer<'de>, -{ - struct VersionKeysVisitor; - impl<'de> serde::de::Visitor<'de> for VersionKeysVisitor { - type Value = Vec; - fn expecting(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result { - f.write_str("a map of version strings to objects") +impl Serialize for Versions { + fn serialize(&self, serializer: S) -> Result { + // Serialize the same shape the registry returns: a JSON object + // mapping version string -> raw JSON tree. Order follows + // `keys` for determinism (HashMap iteration order is arbitrary + // otherwise). + use serde::ser::SerializeMap; + let mut map = serializer.serialize_map(Some(self.keys.len()))?; + for k in &self.keys { + if let Some(v) = self.trees.get(k) { + map.serialize_entry(k, v.as_ref())?; + } } - fn visit_map>( - self, - mut map: M, - ) -> Result, M::Error> { - let mut keys = Vec::with_capacity(map.size_hint().unwrap_or(0)); - while let Some(key) = map.next_key::()? { - map.next_value::()?; - keys.push(key); + map.end() + } +} + +impl<'de> Deserialize<'de> for Versions { + fn deserialize>(deserializer: D) -> Result { + struct VersionsVisitor; + impl<'de> serde::de::Visitor<'de> for VersionsVisitor { + type Value = Versions; + fn expecting(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result { + f.write_str("a map of version strings to manifest objects") + } + fn visit_map>( + self, + mut map: M, + ) -> Result { + let cap = map.size_hint().unwrap_or(0); + let mut keys = Vec::with_capacity(cap); + let mut trees = HashMap::with_capacity(cap); + while let Some(key) = map.next_key::()? { + // Deserialize into `simd_json::OwnedValue` — the + // native simd_json tree. Cheaper than building + // `CoreVersionManifest` because it skips field + // validation / `skip_on_error` round-trips; those + // only run on demand for versions the resolver + // actually reads. + let tree: simd_json::OwnedValue = map.next_value()?; + trees.insert(key.clone(), Arc::new(tree)); + keys.push(key); + } + Ok(Versions { + keys, + trees, + cache: Arc::new(dashmap::DashMap::new()), + }) } - Ok(keys) } + deserializer.deserialize_map(VersionsVisitor) } - deserializer.deserialize_map(VersionKeysVisitor) } /// Version-specific manifest from npm registry. @@ -302,6 +416,19 @@ pub struct VersionManifest { /// Contains only the ~13 fields needed for dependency resolution, installation, /// and lockfile serialization. Display-only fields (description, author, homepage, /// keywords, bugs, repository, npm_user, etc.) are omitted. +/// +/// `skip_on_error` is kept on every field that's typed as `HashMap`, +/// `String`, or `bool` — real registry data regularly ships values +/// that don't match the expected shape (e.g. `engines` with a null +/// value, `hasInstallScript` as a string, legacy `license` as an +/// array-of-object). Dropping the wrapper on those fields broke the +/// lodash fetch at `"ExpectedMap at character 0"` in CI. +/// +/// The three `Option` fields (`bin`, `os`, `cpu`) are the only +/// safe cases: `Value` absorbs any JSON, so the `Value::deserialize` +/// round-trip in `skip_on_error` provides zero robustness. Dropping +/// it there saves one allocation + recursive walk per occurrence +/// without correctness risk. #[derive(Debug, Clone, Default, Serialize, Deserialize)] #[serde(default)] pub struct CoreVersionManifest { @@ -337,10 +464,7 @@ pub struct CoreVersionManifest { pub dist: Dist, - #[serde( - deserialize_with = "skip_on_error", - skip_serializing_if = "Option::is_none" - )] + #[serde(skip_serializing_if = "Option::is_none")] pub bin: Option, #[serde( @@ -349,16 +473,10 @@ pub struct CoreVersionManifest { )] pub engines: Option>, - #[serde( - deserialize_with = "skip_on_error", - skip_serializing_if = "Option::is_none" - )] + #[serde(skip_serializing_if = "Option::is_none")] pub os: Option, - #[serde( - deserialize_with = "skip_on_error", - skip_serializing_if = "Option::is_none" - )] + #[serde(skip_serializing_if = "Option::is_none")] pub cpu: Option, #[serde( diff --git a/crates/ruborist/src/resolver/builder.rs b/crates/ruborist/src/resolver/builder.rs index b03db0fc43..b074dfaf4e 100644 --- a/crates/ruborist/src/resolver/builder.rs +++ b/crates/ruborist/src/resolver/builder.rs @@ -458,7 +458,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,7 +707,7 @@ 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, @@ -729,7 +729,10 @@ 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 + crate::util::maybe_send::MaybeSync, + E: EventReceiver, +>( graph: &mut DependencyGraph, registry: &R, peer_deps: PeerDeps, @@ -759,7 +762,10 @@ 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 + crate::util::maybe_send::MaybeSync, + E: EventReceiver, +>( graph: &mut DependencyGraph, registry: &R, config: BuildDepsConfig, @@ -786,7 +792,10 @@ 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< + R: RegistryClient + crate::util::maybe_send::MaybeSync, + E: EventReceiver, +>( graph: &DependencyGraph, registry: &R, config: &BuildDepsConfig, @@ -838,7 +847,7 @@ async fn run_preload_phase( } /// Run the BFS traversal phase to build the dependency tree. -async fn run_bfs_phase( +async fn run_bfs_phase( graph: &mut DependencyGraph, registry: &R, config: &BuildDepsConfig, @@ -961,7 +970,7 @@ 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> { @@ -975,7 +984,10 @@ 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 + crate::util::maybe_send::MaybeSync, + E: EventReceiver, +>( pkg: &PackageJson, registry: &R, peer_deps: PeerDeps, diff --git a/crates/ruborist/src/resolver/preload.rs b/crates/ruborist/src/resolver/preload.rs index 9fb5aff67b..7f7e2521da 100644 --- a/crates/ruborist/src/resolver/preload.rs +++ b/crates/ruborist/src/resolver/preload.rs @@ -80,7 +80,7 @@ pub async fn preload_manifests( mut on_manifest: F, ) -> PreloadStats where - R: RegistryClient, + R: RegistryClient + crate::util::maybe_send::MaybeSync, E: EventReceiver, F: FnMut(&str, Arc), { diff --git a/crates/ruborist/src/resolver/registry.rs b/crates/ruborist/src/resolver/registry.rs index 6edeb9069c..0edde99325 100644 --- a/crates/ruborist/src/resolver/registry.rs +++ b/crates/ruborist/src/resolver/registry.rs @@ -7,8 +7,6 @@ //! - 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; @@ -114,7 +112,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, @@ -137,7 +135,7 @@ pub fn resolve_from_manifest( manifest: &FullManifest, spec: &str, ) -> Result> { - let version_list: Vec = manifest.versions.clone(); + let version_list: Vec = manifest.versions.keys.clone(); if version_list.is_empty() { return Err(ResolveError::NoVersions(manifest.name.clone())); @@ -147,10 +145,9 @@ 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) + // O(1) lookup from pre-parsed `Versions` map. let version_manifest = manifest .get_core_version(&resolved_version) - .map(Arc::new) .ok_or_else(|| ResolveError::ManifestNotFound { name: manifest.name.clone(), version: resolved_version.clone(), @@ -167,7 +164,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/service/cache.rs b/crates/ruborist/src/service/cache.rs index d11ac6e4cf..14b1f71f53 100644 --- a/crates/ruborist/src/service/cache.rs +++ b/crates/ruborist/src/service/cache.rs @@ -53,7 +53,7 @@ //! It pre-populates memory cache to skip network on subsequent runs. //! ``` -use std::collections::HashMap; +use std::collections::{HashMap, HashSet}; use std::path::{Path, PathBuf}; use anyhow::{Context, Result}; @@ -91,7 +91,16 @@ use parking_lot::RwLock; pub struct MemoryCache(Arc); struct MemoryCacheInner { - full_manifests: RwLock>, + /// `Arc` so `get` returns a cheap atomic-bump clone + /// instead of deep-cloning the full versions HashMap (~100-500 + /// `simd_json::OwnedValue` entries per package). The deep clone + /// was running synchronously on the main task during preload — + /// `FuturesUnordered` polls each future on the owning task, and + /// `resolve_package`'s cache check (line 226 of registry.rs) is + /// the first thing every future does. Standalone manifest-bench + /// hit avg_conc=95 at cap=128; ruborist stalled at avg_conc=56 + /// because the per-future deep clone serialised the main loop. + full_manifests: RwLock>>, versions_info: RwLock>, version_manifests: RwLock>>, } @@ -119,11 +128,11 @@ impl MemoryCache { } // Full manifests - pub fn get_full_manifest(&self, name: &str) -> Option { + pub fn get_full_manifest(&self, name: &str) -> Option> { self.0.full_manifests.read().get(name).cloned() } - pub fn set_full_manifest(&self, name: String, manifest: FullManifest) { + pub fn set_full_manifest(&self, name: String, manifest: Arc) { self.0.full_manifests.write().insert(name, manifest); } @@ -214,6 +223,22 @@ pub struct PackageCache { memory: MemoryCache, /// Disk cache directory (None = no disk cache) cache_dir: Option, + /// Lazily-populated set of package names with existing disk cache. + /// + /// Built once per process from a single `read_dir(cache_dir)` + + /// per-scope `read_dir` for `@scope/pkg` entries. `get_versions_from_disk` + /// checks this set first and short-circuits to `None` when the package + /// isn't on disk — avoiding the 2730 per-package `try_exists` + /// spawn_blocking calls that dominated cold-preload critical path + /// (avg 16 ms per call × 128 parallel futures = measurable wall drag). + /// + /// Mid-run writes via `set_versions_to_disk` are not reflected in this + /// index — that's intentional. Any package we wrote in the current run + /// has already been populated into the memory cache, which takes + /// precedence over disk; the disk path is only consulted for packages + /// *not yet seen this run* with *existing disk cache from a prior run*. + #[cfg_attr(target_arch = "wasm32", allow(dead_code))] + disk_index: Arc>>, } impl PackageCache { @@ -222,6 +247,7 @@ impl PackageCache { Self { memory: MemoryCache::global(), cache_dir: None, + disk_index: Arc::new(tokio::sync::OnceCell::new()), } } @@ -230,9 +256,27 @@ impl PackageCache { Self { memory: MemoryCache::global(), cache_dir: Some(cache_dir), + disk_index: Arc::new(tokio::sync::OnceCell::new()), } } + /// Return `true` if the package has a disk cache entry (based on the + /// lazily-populated bulk-readdir index). Returns `false` if no + /// cache_dir is configured or the index says the package isn't + /// present. First call per process does a single directory scan. + #[cfg(not(target_arch = "wasm32"))] + async fn has_disk_entry(&self, name: &str) -> bool { + let Some(cache_dir) = self.cache_dir.as_ref() else { + return false; + }; + let cache_dir = cache_dir.clone(); + let index = self + .disk_index + .get_or_init(|| async move { build_disk_index(&cache_dir).await }) + .await; + index.contains(name) + } + /// Get the cache directory. pub fn cache_dir(&self) -> Option<&Path> { self.cache_dir.as_deref() @@ -240,8 +284,10 @@ impl PackageCache { // === Memory cache operations (sync) === - /// Get full manifest from memory cache. - pub fn get_full_manifest(&self, name: &str) -> Option { + /// Get full manifest from memory cache. Returns `Arc` so callers + /// pay an atomic-bump clone instead of deep-cloning the full + /// versions HashMap. + pub fn get_full_manifest(&self, name: &str) -> Option> { let result = self.memory.get_full_manifest(name); if result.is_some() { tracing::debug!("Memory cache hit for full manifest: {name}"); @@ -249,8 +295,10 @@ impl PackageCache { result } - /// Set full manifest in memory cache. - pub fn set_full_manifest(&self, name: String, manifest: FullManifest) { + /// Set full manifest in memory cache. Takes `Arc` so callers can + /// share ownership with their own copy without paying for a + /// deep clone at the boundary. + pub fn set_full_manifest(&self, name: String, manifest: Arc) { tracing::debug!("Caching full manifest in memory: {name}"); self.memory.set_full_manifest(name, manifest); } @@ -297,18 +345,21 @@ impl PackageCache { // === Disk cache operations (async, uses tokio-fs-ext) === /// Load versions info from disk cache. + /// + /// Short-circuits via the bulk-readdir index when the package isn't + /// present on disk — avoids the per-package `try_exists` syscall + /// that was 16 ms avg / 230 ms max on CI's blocking pool under load. pub async fn get_versions_from_disk(&self, name: &str) -> Option { let cache_dir = self.cache_dir.as_ref()?; - let path = get_versions_cache_path(cache_dir, name); - - if !super::fs::exists(&path).await { + #[cfg(not(target_arch = "wasm32"))] + if !self.has_disk_entry(name).await { return None; } + let path = get_versions_cache_path(cache_dir, name); match super::fs::read_json::(&path).await { Ok(info) => { tracing::debug!("Disk cache hit for versions: {name}"); - // Also cache in memory self.memory.set_versions(name.to_string(), info.clone()); Some(info) } @@ -319,30 +370,56 @@ impl PackageCache { } } - /// Save versions info to disk cache. - pub async fn set_versions_to_disk(&self, name: &str, info: &VersionsInfo) { - let Some(cache_dir) = &self.cache_dir else { + /// Save versions info to disk cache (fire-and-forget). + /// + /// Spawns the serialisation + write onto the tokio runtime so the caller + /// returns immediately. The resolve pipeline doesn't depend on the write + /// completing — disk cache is best-effort for the *next* run. pcap-driven + /// tuning showed the previous inline `.await` + `serde_json::to_string_pretty` + /// burned ~1–3 ms per call on the hot path, stalling the main preload + /// task and causing the 24..62 active-stream dip observed on CI. + pub fn set_versions_to_disk(&self, name: &str, info: &VersionsInfo) { + let Some(cache_dir) = self.cache_dir.clone() else { return; }; - - let path = get_versions_cache_path(cache_dir, name); - - // Ensure parent directory exists - if let Some(parent) = path.parent() { - let _ = tokio_fs_ext::create_dir_all(parent).await; - } - - match serde_json::to_string_pretty(info) { - Ok(content) => { - if let Err(e) = tokio_fs_ext::write(&path, content.as_bytes()).await { - tracing::debug!("Failed to write versions cache for {name}: {e}"); - } else { - tracing::debug!("Wrote versions to disk cache: {name}"); - } + let name = name.to_string(); + let info = info.clone(); + + // Fire-and-forget on native via `tokio::spawn`. wasm-bindgen- + // futures' `JsFuture` is `!Send` so `tokio::spawn` won't compile + // on wasm32; the wasm OPFS path is structurally serial anyway, + // so just await inline (best-effort, errors logged). + #[cfg(not(target_arch = "wasm32"))] + tokio::spawn(async move { + let path = get_versions_cache_path(&cache_dir, &name); + if let Some(parent) = path.parent() { + let _ = tokio_fs_ext::create_dir_all(parent).await; } - Err(e) => { - tracing::debug!("Failed to serialize versions for {name}: {e}"); + match serde_json::to_vec(&info) { + Ok(bytes) => { + if let Err(e) = tokio_fs_ext::write(&path, bytes).await { + tracing::debug!("Failed to write versions cache for {name}: {e}"); + } + } + Err(e) => { + tracing::debug!("Failed to serialize versions for {name}: {e}"); + } } + }); + #[cfg(target_arch = "wasm32")] + { + // Single-threaded wasm path — caller awaits via the next + // .await point in the resolver loop. Drop into a local + // future via wasm_bindgen_futures::spawn_local. + wasm_bindgen_futures::spawn_local(async move { + let path = get_versions_cache_path(&cache_dir, &name); + if let Some(parent) = path.parent() { + let _ = tokio_fs_ext::create_dir_all(parent).await; + } + if let Ok(bytes) = serde_json::to_vec(&info) { + let _ = tokio_fs_ext::write(&path, bytes).await; + } + }); } } @@ -353,6 +430,10 @@ impl PackageCache { version: &str, ) -> Option> { let cache_dir = self.cache_dir.as_ref()?; + #[cfg(not(target_arch = "wasm32"))] + if !self.has_disk_entry(name).await { + return None; + } let path = get_manifest_cache_path(cache_dir, name, version); if !super::fs::exists(&path).await { @@ -378,35 +459,53 @@ impl PackageCache { } } - /// Save version manifest to disk cache. - pub async fn set_version_manifest_to_disk( + /// Save version manifest to disk cache (fire-and-forget). + /// + /// Takes `Arc` so the shared reference moves into the + /// spawned task without deep-cloning the manifest (they can be 10–50 KB + /// each). Same rationale as [`Self::set_versions_to_disk`]: the resolve + /// pipeline was stalling on inline `.await` + pretty JSON serialisation. + pub fn set_version_manifest_to_disk( &self, name: &str, version: &str, - manifest: &CoreVersionManifest, + manifest: Arc, ) { - let Some(cache_dir) = &self.cache_dir else { + let Some(cache_dir) = self.cache_dir.clone() else { return; }; - - let path = get_manifest_cache_path(cache_dir, name, version); - - // Ensure parent directory exists - if let Some(parent) = path.parent() { - let _ = tokio_fs_ext::create_dir_all(parent).await; - } - - match serde_json::to_string_pretty(manifest) { - Ok(content) => { - if let Err(e) = tokio_fs_ext::write(&path, content.as_bytes()).await { - tracing::debug!("Failed to write manifest cache for {name}@{version}: {e}"); - } else { - tracing::debug!("Wrote manifest to disk cache: {name}@{version}"); - } + let name = name.to_string(); + let version = version.to_string(); + + // See `set_versions_to_disk` above for the cfg-split rationale. + #[cfg(not(target_arch = "wasm32"))] + tokio::spawn(async move { + let path = get_manifest_cache_path(&cache_dir, &name, &version); + if let Some(parent) = path.parent() { + let _ = tokio_fs_ext::create_dir_all(parent).await; } - Err(e) => { - tracing::debug!("Failed to serialize manifest for {name}@{version}: {e}"); + match serde_json::to_vec(&*manifest) { + Ok(bytes) => { + if let Err(e) = tokio_fs_ext::write(&path, bytes).await { + tracing::debug!("Failed to write manifest cache for {name}@{version}: {e}"); + } + } + Err(e) => { + tracing::debug!("Failed to serialize manifest for {name}@{version}: {e}"); + } } + }); + #[cfg(target_arch = "wasm32")] + { + wasm_bindgen_futures::spawn_local(async move { + let path = get_manifest_cache_path(&cache_dir, &name, &version); + if let Some(parent) = path.parent() { + let _ = tokio_fs_ext::create_dir_all(parent).await; + } + if let Ok(bytes) = serde_json::to_vec(&*manifest) { + let _ = tokio_fs_ext::write(&path, bytes).await; + } + }); } } @@ -429,6 +528,51 @@ impl PackageCache { } } +/// Scan `cache_dir` once to build a set of package names with existing +/// disk cache entries. +/// +/// Layout: `{cache_dir}//versions.json` for bare packages, +/// `{cache_dir}/@scope//versions.json` for scoped packages. One +/// top-level `read_dir` plus one `read_dir` per scope directory is +/// enough to enumerate every possible name. Unreadable entries are +/// silently skipped (cache dir doesn't exist on cold runs → empty +/// index → every lookup short-circuits to `None`). +#[cfg(not(target_arch = "wasm32"))] +async fn build_disk_index(cache_dir: &Path) -> HashSet { + let mut set = HashSet::new(); + let Ok(mut entries) = tokio_fs_ext::read_dir(cache_dir).await else { + return set; + }; + while let Ok(Some(entry)) = entries.next_entry().await { + let raw_name = entry.file_name(); + let Some(name) = raw_name.to_str() else { + continue; + }; + if let Some(stripped) = name.strip_prefix('@') { + // Scoped: read one level deeper. The `@` prefix distinguishes + // scopes from bare package names unambiguously (npm package + // names cannot start with `@` unless scoped). + let scope_path = entry.path(); + let Ok(mut sub) = tokio_fs_ext::read_dir(&scope_path).await else { + continue; + }; + while let Ok(Some(sub_entry)) = sub.next_entry().await { + if let Some(pkg) = sub_entry.file_name().to_str() { + set.insert(format!("@{stripped}/{pkg}")); + } + } + } else { + set.insert(name.to_string()); + } + } + tracing::debug!( + "Built disk cache index: {} package(s) at {}", + set.len(), + cache_dir.display() + ); + set +} + /// Cache statistics. #[derive(Debug, Clone)] pub struct CacheStats { @@ -578,7 +722,7 @@ mod tests { ..Default::default() }; - cache.set_full_manifest("test".to_string(), manifest.clone()); + cache.set_full_manifest("test".to_string(), Arc::new(manifest)); let retrieved = cache.get_full_manifest("test").unwrap(); assert_eq!(retrieved.name, "test"); 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 b959020973..f686bf7965 100644 --- a/crates/ruborist/src/service/http.rs +++ b/crates/ruborist/src/service/http.rs @@ -76,13 +76,101 @@ //! WASM targets skip DNS entirely (browser handles it). use std::sync::LazyLock; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::time::Instant; use anyhow::{Context, Result, anyhow}; +use crossbeam_queue::SegQueue; + +/// 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. +#[allow(dead_code)] +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. +#[allow(dead_code)] +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); + +#[allow(dead_code)] +pub fn start_parse_trace() { + while PARSE_TRACE.pop().is_some() {} + PARSE_TRACE_ACTIVE.store(true, Ordering::Relaxed); +} + +#[allow(dead_code)] +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")) @@ -93,6 +181,49 @@ pub(crate) fn get_client() -> Result<&'static reqwest::Client> { HTTP_CLIENT.as_ref().map_err(|e| anyhow!("{e}")) } +/// 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(not(target_arch = "wasm32"))] +fn build_rustls_config() -> Result { + // 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 { + // 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() { + tracing::debug!( + "rustls-native-certs reported {} non-fatal load issues", + roots.errors.len() + ); + } + + let config = rustls::ClientConfig::builder_with_provider(std::sync::Arc::new( + rustls::crypto::aws_lc_rs::default_provider(), + )) + .with_safe_default_protocol_versions() + .map_err(|e| anyhow!("rustls safe_default_protocol_versions: {e}"))? + .with_root_certificates(root_store) + .with_no_client_auth(); + + Ok(config) +} + /// Create a [`reqwest::ClientBuilder`] with TLS, DNS caching, and proxy /// from environment variables. /// @@ -112,10 +243,21 @@ pub fn client_builder() -> Result { let builder = { use crate::service::dns::shared_resolver; + let tls_config = build_rustls_config()?; let mut builder = builder - .use_rustls_tls() + .use_preconfigured_tls(tls_config) .no_proxy() - .dns_resolver(shared_resolver()); + .dns_resolver(shared_resolver()) + // 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); match env_var("ALL_PROXY") { Some(url) => { diff --git a/crates/ruborist/src/service/manifest.rs b/crates/ruborist/src/service/manifest.rs index ad682d6424..ca16bfb776 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::{get_client, parse_trace_enabled, record_http_interval, record_parse_interval}; use crate::model::manifest::{CoreVersionManifest, FullManifest}; /// Result of a full manifest fetch with ETag support. @@ -68,10 +68,14 @@ pub async fn fetch_full_manifest(opts: FetchManifestOptions<'_>) -> 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,6 +196,7 @@ pub async fn fetch_version_manifest( || { let url = url.clone(); async move { + let send_start = std::time::Instant::now(); let response = get_client() .map_err(FetchError::Permanent)? .get(&url) @@ -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/service/registry.rs b/crates/ruborist/src/service/registry.rs index 772cac3f79..b52bffcb66 100644 --- a/crates/ruborist/src/service/registry.rs +++ b/crates/ruborist/src/service/registry.rs @@ -44,6 +44,8 @@ use crate::model::manifest::{CoreVersionManifest, FullManifest}; use crate::resolver::semver::normalize_spec; use crate::resolver::version::resolve_target_version; use crate::traits::registry::{RegistryClient, RegistryError, ResolvedPackage, is_npm_registry}; +#[cfg(not(target_arch = "wasm32"))] +use crate::util::oncemap::OnceMap; /// Unified registry client that works on both native and WASM. /// @@ -68,6 +70,14 @@ pub struct UnifiedRegistry { registry_url: String, cache: Arc, supports_semver: bool, + /// Dedupes concurrent `resolve_full_manifest` fetches for the same + /// package name. First caller hits the network and stores the result; + /// other callers wait on `Notify` and read the shared `Arc`. Built on + /// `DashMap` + `tokio::sync::Notify` so the fast path (cache hit) is + /// lock-free, avoiding the serialisation the previous per-name + /// `tokio::sync::Mutex<()>` gate imposed on the hot dispatch path. + #[cfg(not(target_arch = "wasm32"))] + inflight: Arc>, } /// Builder for `UnifiedRegistry`. @@ -137,6 +147,8 @@ impl UnifiedRegistryBuilder { registry_url, cache, supports_semver, + #[cfg(not(target_arch = "wasm32"))] + inflight: Arc::new(OnceMap::new()), } } } @@ -153,6 +165,8 @@ impl Clone for UnifiedRegistry { registry_url: self.registry_url.clone(), cache: Arc::clone(&self.cache), supports_semver: self.supports_semver, + #[cfg(not(target_arch = "wasm32"))] + inflight: Arc::clone(&self.inflight), } } } @@ -162,10 +176,16 @@ impl Clone for UnifiedRegistry { /// Separates the 200 (full data) and 304 (use cache) cases at the type level, /// so callers can pattern-match instead of string-matching error messages. /// Transient return value, immediately destructured — Box not needed. -#[allow(clippy::large_enum_variant)] +/// +/// `Clone` is required so multiple `resolve_full_manifest` callers that +/// coalesce through `OnceMap` can each take an owned copy of the shared +/// `Arc`. `Arc` keeps that clone an +/// atomic-bump rather than a deep copy of the per-version `OwnedValue` +/// HashMap (~100-500 entries per package). +#[derive(Clone)] enum FullManifestResult { /// Fresh manifest fetched from the network (HTTP 200). - Full(FullManifest), + Full(Arc), /// ETag matched, versions cache is valid (HTTP 304). /// Caller should resolve from the in-memory versions cache and /// fetch individual version manifests as needed. @@ -202,17 +222,61 @@ impl UnifiedRegistry { /// 4. On 304: use disk cache data /// 5. On 200: update memory + disk cache async fn resolve_full_manifest(&self, name: &str) -> Result { - // 1. Check memory cache first + // Fast path: memory cache hit — lock-free read from parking_lot::RwLock. if let Some(manifest) = self.cache.get_full_manifest(name) { tracing::debug!("Memory cache hit for full manifest: {}", name); return Ok(FullManifestResult::Full(manifest)); } - // 2. Load etag from disk cache (if available) + // Coalesce concurrent callers for the same name via OnceMap. + // First caller runs the fetch closure; others await the shared + // result on the OnceMap's `Notify` and clone the cached value. + // Wasm runs single-threaded so coalescing is unnecessary; + // skip OnceMap and call the network path directly. + #[cfg(not(target_arch = "wasm32"))] + { + let shared = self + .inflight + .get_or_init(name.to_string(), || async { + self.fetch_full_manifest_network(name).await.ok() + }) + .await; + + match shared { + Some(arc) => Ok((*arc).clone()), + None => { + // OnceMap clears the key on None, so the next caller + // retries the fetch. Retry once here with a fresh + // error so we surface a useful message to this caller. + self.fetch_full_manifest_network(name).await + } + } + } + #[cfg(target_arch = "wasm32")] + { + self.fetch_full_manifest_network(name).await + } + } + + /// Perform the actual network fetch + cache update. Separated from + /// `resolve_full_manifest` so the OnceMap closure can invoke it + /// without re-entering the dedup layer. + async fn fetch_full_manifest_network( + &self, + name: &str, + ) -> Result { + // Disk ETag probe. The `PackageCache::get_versions_from_disk` + // call first consults a bulk-readdir index built lazily on + // first access — cold runs with an empty (or nonexistent) + // cache_dir short-circuit without per-package syscalls, which + // was the cold-path regression the earlier temporary removal + // (46cb8031) was meant to avoid. Warm runs pay the read + + // JSON parse once per previously-cached manifest and reuse + // its ETag to get a 1-packet `304 Not Modified` instead of + // re-downloading the full manifest body. let disk_versions = self.cache.get_versions_from_disk(name).await; let etag = disk_versions.as_ref().and_then(|v| v.etag.clone()); - // 3. Fetch from network with ETag for validation match manifest::fetch_full_manifest(manifest::FetchManifestOptions { registry_url: &self.registry_url, name, @@ -223,39 +287,35 @@ impl UnifiedRegistry { .map_err(RegistryError)? { manifest::FetchManifestResult::Ok(manifest, new_etag) => { - // 4. Cache full manifest in memory - self.cache - .set_full_manifest(name.to_string(), manifest.clone()); - - // 5. Extract and cache versions info (lightweight) let versions_info = VersionsInfo { versions: Versions { - version_list: manifest.versions.clone(), + version_list: manifest.versions.keys.clone(), dist_tags: manifest.dist_tags.clone(), }, etag: new_etag.clone(), last_updated: current_timestamp_secs(), }; + let manifest = Arc::new(manifest); + self.cache + .set_full_manifest(name.to_string(), Arc::clone(&manifest)); self.cache .set_versions(name.to_string(), versions_info.clone()); - - // 6. Write versions to disk (non-blocking for native, blocking for WASM) - self.cache.set_versions_to_disk(name, &versions_info).await; + // Fire-and-forget disk cache write. + self.cache.set_versions_to_disk(name, &versions_info); Ok(FullManifestResult::Full(manifest)) } manifest::FetchManifestResult::NotModified => { - // 304 Not Modified - use disk cache versions info tracing::debug!("ETag cache hit (304) for: {}", name); if let Some(versions_info) = disk_versions { - // Cache versions in memory for later use self.cache .set_versions(name.to_string(), versions_info.clone()); - Ok(FullManifestResult::NotModified) } else { - // Disk cache corrupted, fetch fresh (without etag) + // Disk cache disappeared between index build and now + // (mid-run eviction or concurrent cleanup). Fall + // back to a fresh fetch without etag. let (manifest, new_etag) = manifest::fetch_full_manifest_fresh( &self.registry_url, name, @@ -264,20 +324,20 @@ impl UnifiedRegistry { .await .map_err(RegistryError)?; - self.cache - .set_full_manifest(name.to_string(), manifest.clone()); - let versions_info = VersionsInfo { versions: Versions { - version_list: manifest.versions.clone(), + version_list: manifest.versions.keys.clone(), dist_tags: manifest.dist_tags.clone(), }, etag: new_etag.clone(), last_updated: current_timestamp_secs(), }; + let manifest = Arc::new(manifest); + self.cache + .set_full_manifest(name.to_string(), Arc::clone(&manifest)); self.cache .set_versions(name.to_string(), versions_info.clone()); - self.cache.set_versions_to_disk(name, &versions_info).await; + self.cache.set_versions_to_disk(name, &versions_info); Ok(FullManifestResult::Full(manifest)) } @@ -341,8 +401,7 @@ impl UnifiedRegistry { // 5. Write to disk cache (only for non-semver registries) if !self.supports_semver { self.cache - .set_version_manifest_to_disk(name, spec, &manifest) - .await; + .set_version_manifest_to_disk(name, spec, manifest.clone()); } Ok(manifest) @@ -357,7 +416,10 @@ impl RegistryClient for UnifiedRegistry { } fn get_cached_full_manifest(&self, name: &str) -> Option { - self.cache.get_full_manifest(name) + // Trait still returns owned for backward compat with non-resolver + // callers (notably `ut view`). Resolver hot paths read the + // `Arc` directly via `cache.get_full_manifest`. + self.cache.get_full_manifest(name).map(|arc| (*arc).clone()) } fn get_cached_versions(&self, name: &str) -> Option { @@ -376,7 +438,7 @@ impl RegistryClient for UnifiedRegistry { async fn fetch_full_manifest(&self, name: &str) -> Result { match self.resolve_full_manifest(name).await? { - FullManifestResult::Full(manifest) => Ok(manifest), + FullManifestResult::Full(manifest) => Ok((*manifest).clone()), FullManifestResult::NotModified => { // 304 in trait context: caller doesn't have versions cache access, // so we return an error indicating the manifest is unchanged. @@ -435,8 +497,7 @@ impl RegistryClient for UnifiedRegistry { // 5. Write to disk cache (only for non-semver registries) if !self.supports_semver { self.cache - .set_version_manifest_to_disk(name, spec, &manifest) - .await; + .set_version_manifest_to_disk(name, spec, manifest.clone()); } Ok(manifest) @@ -481,18 +542,19 @@ impl RegistryClient for UnifiedRegistry { fetch_name, fetch_spec ); - tracing::debug!( - "Using cached full manifest for {}@{}", - fetch_name, - fetch_spec - ); - let version_list: Vec = full_manifest.versions.clone(); - let resolved_version = - resolve_target_version(&full_manifest.dist_tags, &version_list, &fetch_spec) - .map_err(|e| RegistryError(anyhow!("{}@{}: {}", name, spec, e)))?; + // Borrow `keys` directly — `resolve_target_version` only needs + // `&[String]`. The previous `keys.clone()` rebuilt a 100-500 + // entry `Vec` per cache hit (≈1800 hits during a cold + // ant-design preload), bloating per-future allocator pressure + // by ~360k String allocs on shared resolver threads. + let resolved_version = resolve_target_version( + &full_manifest.dist_tags, + &full_manifest.versions.keys, + &fetch_spec, + ) + .map_err(|e| RegistryError(anyhow!("{}@{}: {}", name, spec, e)))?; let version_manifest = full_manifest .get_core_version(&resolved_version) - .map(Arc::new) .ok_or_else(|| { RegistryError(anyhow!( "Version {} not found in manifest for {}", @@ -508,9 +570,11 @@ impl RegistryClient for UnifiedRegistry { ); // Write to disk cache for non-semver registries if !self.supports_semver { - self.cache - .set_version_manifest_to_disk(&fetch_name, &resolved_version, &version_manifest) - .await; + self.cache.set_version_manifest_to_disk( + &fetch_name, + &resolved_version, + version_manifest.clone(), + ); } return Ok(ResolvedPackage { name: name.to_string(), @@ -586,7 +650,7 @@ impl RegistryClient for UnifiedRegistry { let resolve_result = match self.resolve_full_manifest(&fetch_name).await? { FullManifestResult::Full(full_manifest) => { // Got full manifest, resolve from it - let version_list: Vec = full_manifest.versions.clone(); + let version_list: Vec = full_manifest.versions.keys.clone(); if version_list.is_empty() { return Err(RegistryError(anyhow!( @@ -604,7 +668,6 @@ impl RegistryClient for UnifiedRegistry { let version_manifest = full_manifest .get_core_version(&resolved_version) - .map(Arc::new) .ok_or_else(|| { RegistryError(anyhow!( "Version {} not found in manifest for {}", @@ -650,9 +713,11 @@ impl RegistryClient for UnifiedRegistry { // Write to disk cache (only for non-semver registries) if !self.supports_semver { - self.cache - .set_version_manifest_to_disk(&fetch_name, &resolved_version, &version_manifest) - .await; + self.cache.set_version_manifest_to_disk( + &fetch_name, + &resolved_version, + version_manifest.clone(), + ); } Ok(ResolvedPackage { diff --git a/crates/ruborist/src/traits/registry.rs b/crates/ruborist/src/traits/registry.rs index 0760ff9b19..2383664fb9 100644 --- a/crates/ruborist/src/traits/registry.rs +++ b/crates/ruborist/src/traits/registry.rs @@ -140,7 +140,7 @@ pub trait RegistryClient { fn fetch_full_manifest( &self, name: &str, - ) -> impl Future>; + ) -> impl Future> + crate::util::maybe_send::MaybeSend; /// Fetch specific version manifest from registry. /// @@ -153,26 +153,27 @@ pub trait RegistryClient { &self, name: &str, spec: &str, - ) -> impl Future, Self::Error>> { + ) -> impl Future, Self::Error>> + + crate::util::maybe_send::MaybeSend + where + Self: crate::util::maybe_send::MaybeSync, + { async move { let manifest = self.fetch_full_manifest(name).await?; - let version_list: Vec = manifest.versions.clone(); + let version_list: Vec = manifest.versions.keys.clone(); // Resolve version using shared logic let resolved_version = resolve_target_version(&manifest.dist_tags, &version_list, spec) .map_err(|e| RegistryError(anyhow::anyhow!("{}@{}: {}", name, spec, e)))?; - manifest - .get_core_version(&resolved_version) - .map(Arc::new) - .ok_or_else(|| { - RegistryError(anyhow::anyhow!( - "Version {} not found in manifest for {}", - resolved_version, - name - )) - .into() - }) + manifest.get_core_version(&resolved_version).ok_or_else(|| { + RegistryError(anyhow::anyhow!( + "Version {} not found in manifest for {}", + resolved_version, + name + )) + .into() + }) } } @@ -193,7 +194,10 @@ pub trait RegistryClient { &self, name: &str, spec: &str, - ) -> impl Future> { + ) -> impl Future> + crate::util::maybe_send::MaybeSend + where + Self: crate::util::maybe_send::MaybeSync, + { async move { // Normalize spec (handles npm: alias and workspace: prefix) let (fetch_name, fetch_spec) = normalize_spec(name, spec); @@ -226,7 +230,7 @@ pub trait RegistryClient { fetch_spec ); let full_manifest = self.fetch_full_manifest(&fetch_name).await?; - let version_list: Vec = full_manifest.versions.clone(); + let version_list: Vec = full_manifest.versions.keys.clone(); if version_list.is_empty() { return Err(RegistryError(anyhow::anyhow!( @@ -242,7 +246,6 @@ pub trait RegistryClient { let version_manifest = full_manifest .get_core_version(&resolved_version) - .map(Arc::new) .ok_or_else(|| { RegistryError(anyhow::anyhow!( "Version {} not found in manifest for {}", @@ -271,7 +274,7 @@ pub trait RegistryClient { async move { let manifest = self.fetch_full_manifest(name).await?; Ok(VersionsInfo { - version_list: manifest.versions.clone(), + version_list: manifest.versions.keys.clone(), dist_tags: manifest.dist_tags, }) } @@ -320,6 +323,7 @@ pub mod mock { use super::*; /// Internal package data for mock registry. + #[derive(Clone)] struct MockPackage { name: String, dist_tags: HashMap, @@ -327,6 +331,7 @@ pub mod mock { } /// Mock registry client that returns predefined packages. + #[derive(Clone)] pub struct MockRegistryClient { packages: HashMap, } @@ -398,21 +403,21 @@ pub mod mock { .get(name) .ok_or_else(|| MockError(format!("Package not found: {}", name)))?; - // Build JSON and serialize to raw bytes for on-demand extraction + // Build JSON matching the registry wire format and round-trip + // it through the real `FullManifest` deserializer so tests + // exercise the same `Versions` parse path as production. let json = serde_json::json!({ "name": &pkg.name, "dist-tags": &pkg.dist_tags, + "_id": &pkg.name, "versions": &pkg.versions, }); let raw = serde_json::to_vec(&json).expect("mock JSON serialization"); - - Ok(FullManifest { - name: pkg.name.clone(), - dist_tags: pkg.dist_tags.clone(), - versions: pkg.versions.keys().cloned().collect(), - raw: Arc::from(raw), - ..Default::default() - }) + let mut parse_buf = raw.clone(); + let mut manifest: FullManifest = simd_json::serde::from_slice(&mut parse_buf) + .map_err(|e| MockError(format!("mock manifest parse error: {e}")))?; + manifest.raw = Arc::from(raw); + Ok(manifest) } } } diff --git a/crates/ruborist/src/util/maybe_send.rs b/crates/ruborist/src/util/maybe_send.rs new file mode 100644 index 0000000000..cb9abb3d00 --- /dev/null +++ b/crates/ruborist/src/util/maybe_send.rs @@ -0,0 +1,30 @@ +//! Cross-target `Send`/`Sync` shims. +//! +//! On native targets `MaybeSend` is `Send` and `MaybeSync` is `Sync`. +//! On `wasm32-unknown-unknown` they are vacuous marker traits +//! implemented for every type — wasm-bindgen futures use +//! `Rc>` internally and cannot satisfy real `Send`/`Sync`. +//! +//! Using these in trait bounds lets the same trait surface compile +//! against both native (where `tokio::spawn` requires `Send`) and the +//! single-threaded wasm target. + +#[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 {} diff --git a/crates/ruborist/src/util/mod.rs b/crates/ruborist/src/util/mod.rs new file mode 100644 index 0000000000..7d9eac190d --- /dev/null +++ b/crates/ruborist/src/util/mod.rs @@ -0,0 +1,15 @@ +//! Shared utilities usable across ruborist and its downstream crates. + +// OnceMap uses `dashmap` + `tokio::sync::Notify`. Both nominally compile +// for wasm, but the only consumers so far (pm-side downloader / cloner, +// and the ruborist preload dedup) are native-only paths. Gate it to keep +// the wasm surface minimal until a wasm consumer actually needs it. +#[cfg(not(target_arch = "wasm32"))] +pub mod oncemap; + +pub mod maybe_send; + +// Previously these lived in an inline `pub mod util { ... }` re-export +// block in `lib.rs`; keep the same public path to avoid breaking +// downstream consumers. +pub use crate::model::util::{PackageNameStr, parse_package_spec, read_package_json}; diff --git a/crates/ruborist/src/util/oncemap.rs b/crates/ruborist/src/util/oncemap.rs new file mode 100644 index 0000000000..c189f61cc9 --- /dev/null +++ b/crates/ruborist/src/util/oncemap.rs @@ -0,0 +1,465 @@ +#![allow(dead_code)] +//! OnceMap: A concurrent map that ensures each key's work is done exactly once. +//! +//! Based on the pattern from uv package manager: +//! +//! +//! When multiple tasks request the same resource concurrently: +//! - First caller executes the work +//! - Other callers wait and receive the shared result +//! - No duplicate work is performed +//! +//! # Typical Use Case in Package Manager +//! +//! ```text +//! Dependency Graph +//! +//! ┌─────────┐ +//! │ app │ +//! └────┬────┘ +//! ┌──────┼──────┐ +//! ▼ ▼ ▼ +//! ┌───────┐ ┌───┐ ┌───────┐ +//! │ lib-a │ │...│ │ lib-z │ +//! └───┬───┘ └───┘ └───┬───┘ +//! │ │ +//! └───────┬───────┘ +//! ▼ +//! ┌─────────┐ +//! │ react │ ◄── requested by multiple deps +//! └─────────┘ +//! +//! Without OnceMap: +//! lib-a fetches react ──► network request +//! lib-z fetches react ──► network request (duplicate!) +//! +//! With OnceMap: +//! lib-a fetches react ──► network request +//! lib-z waits on lib-a ──► shares result (no duplicate!) +//! ``` + +use dashmap::{DashMap, mapref::entry::Entry}; +use std::future::Future; +use std::hash::Hash; +use std::sync::Arc; +use tokio::sync::Notify; + +/// The state of a value in the OnceMap. +enum Value { + /// Work is in progress, waiters can subscribe to the notify. + Waiting(Arc), + /// Work is complete, result is available. + Done(Arc), +} + +/// A concurrent map that ensures each key's work is done exactly once. +/// +/// # Example +/// ```ignore +/// let map: OnceMap> = OnceMap::new(); +/// +/// // Multiple tasks can call get_or_init concurrently +/// // Only one will actually fetch, others will wait +/// let result = map.get_or_init("react", || async { +/// fetch_package("react").await +/// }).await; +/// ``` +pub struct OnceMap { + map: DashMap>, + /// Waiters for keys that don't exist yet. + /// When get_or_init/complete finishes a key, it notifies and removes the entry. + waiters: DashMap>, +} + +impl Default for OnceMap +where + K: Eq + Hash + Clone, +{ + fn default() -> Self { + Self::new() + } +} + +impl OnceMap +where + K: Eq + Hash + Clone, +{ + /// Create a new empty OnceMap. + pub fn new() -> Self { + Self { + map: DashMap::new(), + waiters: DashMap::new(), + } + } + + /// Register a key as pending (Waiting state) without starting work. + /// Returns the Notify handle if newly registered, None if already exists. + pub fn register(&self, key: K) -> Option> { + use dashmap::mapref::entry::Entry; + match self.map.entry(key) { + Entry::Occupied(_) => None, + Entry::Vacant(vacant) => { + let notify = Arc::new(Notify::new()); + vacant.insert(Value::Waiting(Arc::clone(¬ify))); + Some(notify) + } + } + } + + /// Wait for a key to complete. + /// + /// - If the key exists and is Done, returns immediately. + /// - If the key exists and is Waiting, waits for it to complete. + /// - If the key doesn't exist yet, waits for it to be created and completed. + /// + /// This enables parent-ordering in the clone pipeline: a child package + /// can wait for its parent's clone to finish even if the parent's + /// `get_or_init` hasn't been called yet. + pub async fn wait_if_pending(&self, key: &K) { + // Fast path: key already exists + if let Some(entry) = self.map.get(key) { + match entry.value() { + Value::Done(_) => return, + Value::Waiting(notify) => { + let notify = Arc::clone(notify); + let notified = notify.notified(); + drop(entry); + + if let Some(entry) = self.map.get(key) + && matches!(entry.value(), Value::Done(_)) + { + return; + } + + notified.await; + return; + } + } + } + + // Key doesn't exist yet — register a creation waiter. + // get_or_init/complete will notify this when the key reaches Done (or fails). + let creation_notify = self + .waiters + .entry(key.clone()) + .or_insert_with(|| Arc::new(Notify::new())) + .clone(); + + // Register notified BEFORE double-checking to prevent missed notifications. + let notified = creation_notify.notified(); + + // Double-check: key might have been created and completed between + // the initial map.get and registering here. + if let Some(entry) = self.map.get(key) + && matches!(entry.value(), Value::Done(_)) + { + return; + } + + // Wait for the key to be created and completed + notified.await; + } + + /// Complete a pre-registered key with a value. + /// + /// This is used in conjunction with `register()` for the pre-registration pattern: + /// 1. Call `register(key)` synchronously to claim the key + /// 2. Do async work + /// 3. Call `complete(key, value, notify)` to store result and notify waiters + /// + /// The notify handle must be the one returned from `register()`. + pub fn complete(&self, key: K, value: Option, notify: Arc) { + match value { + Some(v) => { + self.map.insert(key.clone(), Value::Done(Arc::new(v))); + } + None => { + self.map.remove(&key); + } + } + self.notify_all(&key, ¬ify); + } + + /// Notify both the work-notify and any creation waiters for a key. + fn notify_all(&self, key: &K, notify: &Notify) { + notify.notify_waiters(); + if let Some((_, cn)) = self.waiters.remove(key) { + cn.notify_waiters(); + } + } + + /// Get or initialize a value for the given key. + /// + /// If the key doesn't exist, the provided async closure will be called to compute the value. + /// If another task is already computing the value, this will wait for it to complete. + /// + /// Returns `Some(Arc)` if the computation succeeded, `None` if it failed. + pub async fn get_or_init(&self, key: K, init: F) -> Option> + where + F: FnOnce() -> Fut, + Fut: Future>, + { + // Fast path: check if already done + if let Some(entry) = self.map.get(&key) + && let Value::Done(result) = entry.value() + { + return Some(Arc::clone(result)); + } + + // Try to register as the worker + let notify = Arc::new(Notify::new()); + let entry = self.map.entry(key.clone()); + + match entry { + Entry::Occupied(occupied) => { + // Someone else is working on it or it's done + match occupied.get() { + Value::Done(result) => { + return Some(Arc::clone(result)); + } + Value::Waiting(existing_notify) => { + let existing_notify = Arc::clone(existing_notify); + + // Register the waiter BEFORE releasing the lock. + // This prevents a race condition where the worker completes + // and calls notify_waiters() between dropping the lock and + // registering the waiter - which would cause us to miss the + // notification and wait forever. + let notified = existing_notify.notified(); + + // Now safe to release the lock + drop(occupied); + + // Double-check: the value might have been inserted between + // releasing the lock and here. If so, return it directly. + if let Some(entry) = self.map.get(&key) + && let Value::Done(result) = entry.value() + { + return Some(Arc::clone(result)); + } + + // Wait for the worker to complete + notified.await; + + // Check the result + if let Some(entry) = self.map.get(&key) + && let Value::Done(result) = entry.value() + { + return Some(Arc::clone(result)); + } + return None; + } + } + } + Entry::Vacant(vacant) => { + // We are the worker + vacant.insert(Value::Waiting(Arc::clone(¬ify))); + } + } + + // Execute the work + let result = init().await; + + // Update the map with the result + let arc_value = result.map(|v| { + let arc = Arc::new(v); + self.map.insert(key.clone(), Value::Done(Arc::clone(&arc))); + arc + }); + if arc_value.is_none() { + self.map.remove(&key); + } + self.notify_all(&key, ¬ify); + arc_value + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::sync::atomic::{AtomicUsize, Ordering}; + use std::time::Duration; + + #[tokio::test] + async fn test_single_execution() { + let map: OnceMap = OnceMap::new(); + let call_count = Arc::new(AtomicUsize::new(0)); + + let call_count_clone = Arc::clone(&call_count); + let result = map + .get_or_init("key".to_string(), || async move { + call_count_clone.fetch_add(1, Ordering::SeqCst); + Some(42) + }) + .await; + + assert_eq!(*result.unwrap(), 42); + assert_eq!(call_count.load(Ordering::SeqCst), 1); + + // Second call should return cached result + let call_count_clone = Arc::clone(&call_count); + let result = map + .get_or_init("key".to_string(), || async move { + call_count_clone.fetch_add(1, Ordering::SeqCst); + Some(100) + }) + .await; + + assert_eq!(*result.unwrap(), 42); // Still 42, not 100 + assert_eq!(call_count.load(Ordering::SeqCst), 1); // Still 1, not called again + } + + #[tokio::test] + async fn test_concurrent_requests() { + let map = Arc::new(OnceMap::::new()); + let call_count = Arc::new(AtomicUsize::new(0)); + + let mut handles = vec![]; + + // Spawn 10 concurrent tasks requesting the same key + for _ in 0..10 { + let map = Arc::clone(&map); + let call_count = Arc::clone(&call_count); + + handles.push(tokio::spawn(async move { + map.get_or_init("shared_key".to_string(), || async move { + // Simulate slow work + tokio::time::sleep(Duration::from_millis(50)).await; + call_count.fetch_add(1, Ordering::SeqCst); + Some(42) + }) + .await + })); + } + + // Wait for all tasks + let results: Vec<_> = futures::future::join_all(handles) + .await + .into_iter() + .map(|r| r.unwrap()) + .collect(); + + // All should get the same result + for result in results { + assert_eq!(*result.unwrap(), 42); + } + + // Work should only be done once + assert_eq!(call_count.load(Ordering::SeqCst), 1); + } + + #[tokio::test] + async fn test_failed_work_allows_retry() { + let map: OnceMap = OnceMap::new(); + let attempt = Arc::new(AtomicUsize::new(0)); + + // First attempt fails + let attempt_clone = Arc::clone(&attempt); + let result = map + .get_or_init("key".to_string(), || async move { + attempt_clone.fetch_add(1, Ordering::SeqCst); + None // Fail + }) + .await; + assert!(result.is_none()); + + // Second attempt should be allowed and succeed + let attempt_clone = Arc::clone(&attempt); + let result = map + .get_or_init("key".to_string(), || async move { + attempt_clone.fetch_add(1, Ordering::SeqCst); + Some(42) + }) + .await; + assert_eq!(*result.unwrap(), 42); + assert_eq!(attempt.load(Ordering::SeqCst), 2); + } + + /// Test that waiters don't miss notifications due to race conditions. + /// + /// This test verifies the fix for a race condition where a waiter could + /// miss the notification if the worker completed between the waiter + /// releasing the lock and registering for notifications. + /// + /// Timeline of the race condition (before fix): + /// ```text + /// Worker Waiter + /// ────── ────── + /// 1. register key + /// insert Waiting(notify) + /// start work... + /// 2. sees Waiting state + /// clone notify + /// drop(lock) ← release lock + /// + /// ┌─────────────────┐ + /// │ DANGER WINDOW │ + /// │ (not listening │ + /// │ for notify) │ + /// └─────────────────┘ + /// + /// 3. work done! + /// insert Done(value) + /// notify_waiters() ← sent! + /// (but no one listening) + /// + /// 4. notified = notify.notified() + /// ← too late! already notified + /// + /// 5. notified.await + /// ← waits forever, deadlock! + /// ``` + /// + /// The fix: register `notified()` BEFORE releasing the lock, then + /// double-check if the value was inserted. + #[tokio::test] + async fn test_no_missed_notifications() { + use tokio::sync::Barrier; + + let map = Arc::new(OnceMap::::new()); + let barrier = Arc::new(Barrier::new(2)); + + // Spawn the worker task + let map_clone = Arc::clone(&map); + let barrier_clone = Arc::clone(&barrier); + let worker = tokio::spawn(async move { + map_clone + .get_or_init("key".to_string(), || async move { + // Wait for waiter to be ready + barrier_clone.wait().await; + // Small delay to let waiter release lock + tokio::time::sleep(Duration::from_millis(5)).await; + Some(42) + }) + .await + }); + + // Spawn waiter task + let map_clone = Arc::clone(&map); + let barrier_clone = Arc::clone(&barrier); + let waiter = tokio::spawn(async move { + // Small delay to ensure worker registers first + tokio::time::sleep(Duration::from_millis(1)).await; + + // Signal worker to proceed + barrier_clone.wait().await; + + // This should not hang - if it does, the race condition exists + map_clone + .get_or_init("key".to_string(), || async move { + panic!("Waiter should not execute work"); + }) + .await + }); + + // Use timeout to detect deadlock + let timeout = Duration::from_secs(2); + let results = tokio::time::timeout(timeout, futures::future::join(worker, waiter)) + .await + .expect("Test timed out - possible deadlock due to missed notification"); + + // Both should get the same result + assert_eq!(*results.0.unwrap().unwrap(), 42); + assert_eq!(*results.1.unwrap().unwrap(), 42); + } +}