Speed up hosted and vendored scans: concurrent API requests, parallel crawl, single-pass rewriters

.gitattributes

@@ -10,3 +10,9 @@ crates/socket-patch-core/tests/fixtures/pdm-native/*.lock -text
 # refuse CRLF by design (vendor_lockfile_crlf_unsupported), and the tests
 # derive their CRLF variants from the LF bytes themselves.
 crates/socket-patch-core/tests/fixtures/pnpm-hosted/** -text
+
+# The legacy vendor ledgers are what the base binary wrote, byte for byte:
+# vendor_ledger_schema_e2e replays them through `vendor --revert` and
+# compares the result byte for byte, so a CRLF checkout would change both
+# the replayed wiring files and the expected revert.
+crates/socket-patch-cli/tests/fixtures/legacy-ledgers/** -text

Cargo.toml

@@ -21,17 +21,20 @@ sha1 = "=0.10.6"
 hex = "=0.4.3"
 reqwest = { version = "=0.12.28", features = ["rustls-tls", "json"], default-features = false }
 tokio = { version = "=1.50.0", features = ["full"] }
+futures-util = { version = "=0.3.32", default-features = false, features = ["std"] }
 thiserror = "=2.0.18"
 walkdir = "=2.5.0"
 uuid = { version = "=1.21.0", features = ["v4"] }
 console = "=0.15.11"
 dialoguer = "=0.11.0"
 tempfile = "=3.26.0"
 regex = "=1.12.3"
+aho-corasick = "=1.1.4"
 glob = "=0.3.4"
 toml_edit = "=0.25.12"
 once_cell = "=1.21.3"
 qbsdiff = "=1.4.4"
+rayon = "=1.12.0"
 tar = "=0.4.46"
 flate2 = "=1.1.9"
 zip = { version = "=8.6.0", default-features = false, features = ["deflate"] }

README.md

@@ -531,6 +531,40 @@ warning — it never breaks a command or pollutes `--json` output. `socket-patch
 cloned repo must never be able to redirect where patches come from or spend your token.
 (Full rationale: [docs/design/configuration.md](docs/design/configuration.md).)
 
+One more env-only knob tunes *pacing* rather than routing. `scan` queries the patch API
+with several requests in flight: against the authenticated endpoint, a quarter of the
+requests a step has to make, between 8 and 32 (so a step with 128 or more requests runs
+32 at once, one with 32 or fewer runs 8); against the public proxy, which shares one
+server-side limit across anonymous callers, 4. The patch-record fetches behind `vex` and
+`scan --vex` run up to 10 at once (4 on the proxy).
+`SOCKET_API_CONCURRENCY=<n>` overrides that, clamped to `1`-`32`; on the public proxy it
+can only lower it. Set it when an endpoint in front of the API caps in-flight requests
+per client — a self-hosted `--api-url`, a corporate reverse proxy, a WAF or a CDN — and
+a scan starts reporting fewer patches than it should because some requests are being
+rejected. `SOCKET_API_CONCURRENCY=1` sends one request at a time, the slowest and most
+conservative setting. An unset, empty or non-numeric value leaves the defaults in place.
+
+A throttled patch API is retried, within bounds. An HTTP `429` or `503` answer to any
+patch-API query (batch search, patch lists, patch views, VEX record fetches, hosted
+package references) is retried up to 3 times, waiting as long as the server's
+`Retry-After` asks (seconds or an HTTP date; a request asked to wait more than 30 s gives
+up at once) or, without one, 0.5 s, 1 s, 2 s with jitter. All retries in one run must
+finish within 60 s of the run's first retry (wall-clock: requests waiting in parallel
+don't add up), so a heavily throttled run gives up instead of hanging. `SOCKET_API_MAX_RETRIES=<n>` changes
+the per-request count (`0`-`10`; `0` turns retries off). Other errors are never retried.
+A query still throttled after its retries is reported, never dropped: a failed batch
+prints `Warning: API batch <n> of <total> failed: …` (under `--json`, a top-level
+`warnings[]` entry with code `api_batch_failed`), a failed patch-list lookup prints
+`Warning: could not fetch details for <purl>: …` (`--json`: `patch_details_failed`), and
+if every query fails the scan exits 1 with an error, as before.
+
+The crawl has a pacing knob too. Its directory walks (`node_modules`, and the Maven
+repository with its POM parse) run on a small pool of threads: 4 by default (fewer on a machine with fewer performance cores), because the walk is bound
+by the kernel's directory cache and more threads only add system time.
+`SOCKET_WALK_THREADS=<n>` overrides that, clamped to `1`-`16` and to the machine's CPU
+count; an unset, empty or non-numeric value leaves the default in place. A soft open-file
+limit below 128 still runs the walk on one thread, whatever the knob says.
+
 The sections below list only each command's **command-specific** flags.
 
 ### `scan`

crates/socket-patch-cli/Cargo.toml

@@ -22,6 +22,7 @@ clap = { workspace = true }
 serde = { workspace = true }
 serde_json = { workspace = true }
 tokio = { workspace = true }
+futures-util = { workspace = true }
 console = { workspace = true }
 dialoguer = { workspace = true }
 uuid = { workspace = true }
@@ -62,6 +63,10 @@ docker-e2e = []
 setup-e2e = []
 
 [dev-dependencies]
+# vendor_crash_safety_e2e / vendor_group_commit_e2e crash the binary through
+# its failpoints; with this, `cargo test --release` (the test-release job)
+# builds them in too. Dev-only: resolver 2 keeps it out of normal builds.
+socket-patch-core = { workspace = true, features = ["failpoints"] }
 sha2 = { workspace = true }
 # docker_e2e_vendor_maven's host oracle recomputes the maven2 .jar.sha1 sidecar.
 sha1 = { workspace = true }

crates/socket-patch-cli/src/commands/fetch_stage.rs

@@ -10,13 +10,15 @@
 use std::collections::{HashMap, HashSet};
 use std::path::{Path, PathBuf};
 
+use futures_util::StreamExt;
 use socket_patch_core::api::blob_fetcher::{
     fetch_missing_blobs, fetch_missing_sources, get_missing_archives, get_missing_blobs,
     DownloadMode, FetchMissingBlobsResult,
 };
-use socket_patch_core::api::client::{get_api_client_with_overrides, ApiClient};
+use socket_patch_core::api::client::{get_api_client_with_overrides, hold_back_debug, ApiClient};
 use socket_patch_core::manifest::schema::{PatchManifest, PatchRecord};
 use socket_patch_core::patch::apply::{is_valid_blob_hash, PatchSources};
+use socket_patch_core::utils::concurrent::{api_concurrency_for, ordered_concurrent};
 use tempfile::TempDir;
 
 use super::get::base64_decode;
@@ -565,6 +567,15 @@ pub(crate) async fn stage_vendor_sources_in_memory(
             }
         };
         let mut failed: Vec<&str> = Vec::new();
+        // The views are fetched concurrently (at most `api_concurrency` in
+        // flight) but consumed in `to_fetch` order, each request's `--debug`
+        // lines released at its turn, so `mem`, `failed` and every error
+        // line fold exactly as the serial loop's did.
+        let mut views = std::pin::pin!(ordered_concurrent(
+            to_fetch.iter(),
+            api_concurrency_for(client.uses_public_proxy(), to_fetch.len()),
+            |(_, uuid)| async move { (*uuid, hold_back_debug(client.fetch_patch(uuid)).await) },
+        ));
         for (i, (purl, uuid)) in to_fetch.iter().enumerate() {
             if to_fetch.len() > 1 {
                 status.set(format!(
@@ -574,7 +585,19 @@ pub(crate) async fn stage_vendor_sources_in_memory(
                     to_fetch.len()
                 ));
             }
-            match client.fetch_patch(uuid).await {
+            let view = match views.next().await {
+                Some((planned, view)) if planned == *uuid => view.release(),
+                // Unreachable: the plan IS this list. Falling back to the
+                // live request keeps the staging COMPLETE if the two ever
+                // fall out of step — running dry here would otherwise
+                // return `Ready` with blobs missing and nothing in
+                // `failed`.
+                _ => {
+                    debug_assert!(false, "view prefetch plan out of step with the fetch list");
+                    client.fetch_patch(uuid).await
+                }
+            };
+            match view {
                 Ok(Some(patch)) => {
                     let mut complete = true;
                     for (file, info) in &patch.files {