From 6141dba774d47f4fc8e399c5459769628b0d5dd2 Mon Sep 17 00:00:00 2001 From: kridaydave Date: Sun, 11 Oct 2026 20:45:22 +0530 Subject: [PATCH 1/5] perf: overlap the OSV query with extraction, and scan files once Measured against a loopback registry, a 200-file review spends 717 ms on the wall clock and 45 ms of CPU. The gap is two things: a blocking OSV round trip in the critical path, and a serial evaluation loop in the ci lane. 1. The advisory query no longer blocks the review. resolve_advisories returns Ready for everything answerable without the network (recall hit, policy switch, warm cache, an ecosystem with no coverage) and Pending with the OSV request already running otherwise, so the tarball fetch, extraction and diff all proceed while the network works. join collects it at scoring time. The upstream disclosure behaviour is preserved verbatim: an Err stays a HIGH finding, an unverified report the operator accepted stays a MEDIUM one, and fail_closed is still checked before any stale fallback is served. 2. The ci lane reviews packages in parallel. One worker per package, capped at eight, each with its own ReviewContext and its own SQLite handle, since neither is shareable. Findings are emitted in candidate order rather than completion order, because CI output is diffed between runs. The budget check still refuses before any work starts. 3. Each file is read once through a 64 KiB window instead of being buffered whole for a hash, a NUL scan and a classify. The digest is unchanged, because it decides whether a file is compared at all, and peak memory stops scaling with the largest file in the package. The UTF-8 carry across the window boundary completes by the sequence's real length, and a carry still open at EOF is judged malformed, so a file ending mid-code-point no longer reads as text. 4. The dev profile emits line tables rather than full DWARF, and emits none for dependencies. Cargo's default is opt-level 0 with debug = 2, which at this crate's size is what accumulates a hundred stale incremental caches. Cargo.lock does not move. PERF.md carries the measurements, the commands that produced them, and what is still not known. scripts/perf_harness.py reproduces the runtime numbers and scripts/perf_watchdog.sh bounds the process count of a test run. --- .gitignore | 1 + Cargo.toml | 22 ++ PERF.md | 345 +++++++++++++++++++++++++ references/features/advisory.md | 21 +- references/features/review.md | 41 +-- scripts/perf_harness.py | 306 ++++++++++++++++++++++ scripts/perf_watchdog.sh | 95 +++++++ src/advisory.rs | 437 +++++++++++++++++++++++--------- src/ci.rs | 378 +++++++++++++++++++++++---- src/diff.rs | 222 ++++++++++++++-- src/review.rs | 69 ++--- src/store.rs | 44 +++- 12 files changed, 1734 insertions(+), 247 deletions(-) create mode 100644 PERF.md create mode 100755 scripts/perf_harness.py create mode 100755 scripts/perf_watchdog.sh diff --git a/.gitignore b/.gitignore index b8ee3e5..fcb1377 100644 --- a/.gitignore +++ b/.gitignore @@ -18,3 +18,4 @@ packages/@kridaydave/binary-*/blueline.exe *.tgz .useit/ +.perfwork/ diff --git a/Cargo.toml b/Cargo.toml index 0910df0..3bdae2d 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -38,3 +38,25 @@ proptest = "1" [profile.release] strip = true lto = true + +# The dev profile is where the time goes. Cargo's default is `opt-level = 0` +# with `debug = 2`, and at this crate's size that is what produces a 100 MB test +# binary, a linker peak in the gigabytes, and a `target/debug` measured in +# gigabytes across a hundred stale incremental session caches. Line tables keep +# a failing test pointing at the right source line, which is all a panic needs, +# and they cost a fraction of the full DWARF. +[profile.dev] +debug = "line-tables-only" + +# `cargo test` builds its own profile, and it inherits nothing from `[profile.dev]` +# unless it is stated here, so both halves have to be set. +[profile.test] +debug = "line-tables-only" + +# Dependencies are only ever linked in. Nobody steps through `serde_json`, so +# emitting debug info for it only makes the linker read more than it has to. +[profile.dev.package."*"] +debug = false + +[profile.test.package."*"] +debug = false diff --git a/PERF.md b/PERF.md new file mode 100644 index 0000000..b026ccc --- /dev/null +++ b/PERF.md @@ -0,0 +1,345 @@ +# Blueline performance report + +Status: measured 2026-10-11 against v0.3.1 (`2d7864e`). Nothing here has been +fixed. Every number below was measured on the dev box, and every one carries +the command that produced it or the file and line it came from. + +Read this before starting work. Section 8 lists what is *not* known, and two of +the five items in section 7 rest on code reading rather than measurement. + +--- + +## 1. The verdict + +This is a project problem, not a Rust problem. The release binary uses **45 ms +of CPU** and **17 MB of RSS** to review a 200-file package. There is no +language-level overhead worth discussing at this scale. + +What costs resources is two other things: + +1. A blocking network round trip to `api.osv.dev` in the critical path, plus a + fully serial evaluation loop in the `ci` lane. +2. Build artifacts. `target/debug` is **5.4 GB** on this box, and a single test + pins a core and 280 MB for 25 seconds on every gate run. + +--- + +## 2. Runtime, per review + +Release binary, 4-core Fedora box, median of 7 runs against a loopback registry +serving synthetic npm packages with real sha512 integrity. Produced by +`scripts/perf_harness.py`. + +| package | wall, default | wall, advisories off | CPU | peak RSS | +|---|---|---|---|---| +| 200 files x 40 B | 717 ms | 174 ms | 45 ms | 17.3 MB | +| 2000 files x 1 KB | 1077 ms | 533 ms | 381 ms | 17.5 MB | +| 2000 x 1 KB, 1% changed | 904 ms | 390 ms | 255 ms | 16.6 MB | +| 90 MB of text, line diffed | 6185 ms | data pending | 3340 ms | 117 MB | +| 96 MB of text, over diff cap | 1175 ms | data pending | 452 ms | 18 MB | + +Startup is not a factor: 4 MB RSS, under 1 ms. + +RSS is sampled from `/proc` across the process tree, so it covers the Landlock +extraction child as well as the parent. An earlier draft of this table carried +24 to 32 MB for the first three rows, which came from `wait4` rusage. That +method was discarded (see section 8) and those figures were wrong. + +The last two rows are the same content volume on either side of the 2 MiB diff +cap (`MAX_DIFF_FILE_BYTES`, `src/diff.rs:11`). Line diffed, resident memory +lands at about 1.3x the content. Skipped as opaque, it stays at 18 MB. That is +roughly **19 ms of CPU per megabyte diffed**. + +### The wall clock gap + +One review, every request to the loopback registry, gap since the previous one: + +``` + t (ms) gap bytes path + 224.4 224.4 534 /cntpkg packument + 230.3 5.9 2273 /cntpkg/-/cntpkg-1.0.1.tgz target tarball + 384.0 153.7 2254 /cntpkg/-/cntpkg-1.0.0.tgz baseline tarball + 1145.9 762.0 0 404 attestation 670 ms gap + + default policy 1166 ms total + advisories off 196 ms total + CPU, either way 45 ms +``` + +Only four registry requests happen, so the packument memo at +`src/registry/npm.rs` is already doing its job. The 670 ms gap is a query to +`api.osv.dev`, which never touches the loopback registry and so does not appear +above. Measured directly: + +``` +POST https://api.osv.dev/v1/query 0.72s 0.56s 0.68s 0.77s +``` + +It is issued at `src/review.rs:294`, **after** extraction has finished, and it +depends on none of the extracted tree. On a small package it is about 75% of +total wall clock. + +--- + +## 3. The serial case that hurts: `blueline ci` + +| fact | value | where | +|---|---|---| +| evaluation loop | plain sequential `for` | `src/ci.rs:296`, `src/ci.rs:409` | +| default package cap | 100 | `src/policy.rs:464` | +| concurrency anywhere in `src/` | none, no `rayon`, no `par_iter` | whole crate | +| advisory cache on a CI runner | always cold, 12 h clean TTL | `src/policy.rs:404` | +| worst case network waiting | ~56 s | 100 x 560 ms | + +Every package in a lockfile pays its own OSV round trip, in order, on a machine +where the advisory cache is always empty. + +--- + +## 4. Build and disk cost + +| command | wall | peak RSS | +|---|---|---| +| `cargo build --release --locked -j 2` (LTO on) | 3m06s | 970 MB | +| `cargo build --locked` (no-op) | 9.5 s | 377 MB | +| `touch src/diff.rs && cargo build --locked` | 4.1 s | 378 MB | +| `cargo test --all-targets --locked --no-run -j 2` | 32 s | 574 MB | + +On disk: + +| artifact | size | +|---|---| +| `target/debug` total | 5.4 GB | +| `target/debug/incremental` | 1.9 GB across **101** stale session caches | +| `target/debug/deps` | 3.0 GB | +| test binaries in `deps` | 32 | +| `target/debug/blueline` | 105 MB | +| `target/release/blueline` | 7.6 MB | + +`Cargo.toml` sets only `[profile.release]`. There is no `[profile.dev]`, so the +dev profile is the cargo default: `opt-level=0`, full debuginfo, and unbounded +incremental compilation. + +--- + +## 5. The dev gate + +`./scripts/verify.sh`, default stage set, warm build cache: + +| stage | wall | CPU | peak RSS | +|---|---|---|---| +| `cargo fmt --all -- --check` | 0.7 s | 0.6 s | 76 MB | +| `cargo clippy --all-targets --locked -- -D warnings` | 5.5 s | 5.2 s | 437 MB | +| `cargo test --all-targets --locked` | **49.8 s** | **57.6 s** | **391 MB** | +| `scripts/check-references.sh` | 1.1 s | | | + +The test stage is the entire cost, and inside it one test dominates. + +--- + +## 6. `blueline-` at 40%+ CPU + +The process Kriday sees is cargo's libtest harness from `target/debug/deps/`. +It is **not** a daemon and **not** the CLI. It exists only while `cargo test` +runs. If one is sitting there when no test is running, it is a stuck run and +`pkill -f 'blueline-'` clears it. + +Module breakdown, timed by running each module's tests in isolation: + +``` +registry 94 tests 20.7 s + registry::npm 18 tests 23.4 s <- 20.3 s of this is one test + registry::aur 31 tests 10.3 s +extract 56 tests 6.3 s +recall 51 tests 4.3 s +everything else ~1.1 s each (that is bare process start) +``` + +The single test, run alone: + +| test | wall | CPU | peak RSS | +|---|---|---|---| +| `registry::npm::tests::the_packument_memo_is_bounded_and_the_estimate_is_not_a_fig_leaf` | **25.7 s** | 22.96 s at 90% of one core | **280 MB** | +| `registry::npm::tests::the_packument_estimate_is_exact` (control) | 0.00 s | 0.00 s | 10 MB | + +That one test is **half the wall clock of the entire 835-test lib suite**, and +it alone accounts for 280 MB of the test stage's 391 MB peak. + +### Why + +`src/registry/npm.rs:1232` builds a deliberately oversized fixture: 64 versions, +each carrying a signature block of two 2 MiB strings, so 256 MiB of +`serde_json::Value`. + +`packument_size_estimate` at `src/registry/npm.rs:301` then calls +`serde_json::to_string` on each block purely to read `.len()`, materialising a +further ~256 MiB of transient `String`. + +That is roughly half a gigabyte of allocation churn in a debug build at +`opt-level=0`. + +### Production impact: none + +The docstring at `src/registry/npm.rs:285` says it directly: the estimate is not +a security bound, and the fetch is byte-capped separately. No real review builds +a 256 MiB signature block. This is purely developer-experience cost. + +A second, smaller contributor: eight tests in `src/registry/npm.rs` run mock HTTP +servers in a `while start.elapsed() < Duration::from_millis(600)` loop and +`a_packument_is_fetched_once_per_package` polls to a 10 s deadline. That is +about 4.8 s plus 2 s of pure sleeping, all inside `#[cfg(test)]`. + +--- + +## 7. Ranked work items + +Ordered by conviction, not by size of win. + +### 1. Overlap the OSV query with the tarball fetch + +`fetch_advisories` is called at `src/review.rs:294`, after both tarballs are +already extracted. The query takes about 560 ms and depends on none of that. + +- Target: 717 ms wall becomes about 175 ms on a 200-file package, CPU unchanged + at 45 ms. +- The query must still be joined before the verdict is scored. +- A cache hit must skip the network exactly as it does now. +- `fail_closed_network` behaviour on a network error must be unchanged. +- Needs a test that fails if the advisory result is used before the request + completes. + +Smallest diff on this list, no behaviour change. + +### 2. Parallelise the `ci.rs` evaluation loop + +`src/ci.rs:296` and `src/ci.rs:409` call `evaluate_package` one package at a +time. 100 packages at 560 ms each is about 56 s of network waiting. + +- The store is already opened in WAL mode with a 5 s busy timeout at + `src/store.rs:389`, so concurrent writers have a path. +- Findings must be emitted in a **deterministic order** regardless of completion + order, because CI output is diffed between runs. +- The per-package `ReviewContext` built at `src/ci.rs:309` must not be shared + across threads. +- The budget check must still refuse before any work starts, not after. + +### 3. Cap the incremental cache + +Add a `[profile.dev]` to `Cargo.toml` so old session caches are evicted rather +than accumulating. Reclaims most of the 1.9 GB. + +- Do not touch `[profile.release]`: `lto` and `strip` stay. +- CI builds with `--locked`, so this must not move `Cargo.lock`. + +### 4. Stop reading every file twice + +``` +scan_tree() src/diff.rs:59 target tree, fs::read per file +scan_tree() src/diff.rs:72 base tree, fs::read per file +diff_single_file src/diff.rs:365 fs::read(old) +diff_single_file src/diff.rs:369 fs::read(new) +``` + +- The SHA-256 of file contents must not change, because it decides whether a + file is compared at all. +- `classify_bytes` must still see enough bytes to distinguish `Text`, `Binary` + and `OpaqueTooLarge` at the 2 MiB cap. +- The extraction caps in `extract.rs` are a separate guard and must not be + weakened. + +**This one is read from the code, not isolated by a measurement.** + +### 5. Tighten the sandbox poll + +`wait_with_timeout` at `src/sandbox.rs:1120` polls `try_wait` on a 20 ms sleep, +twice per review for the target and baseline extractions. Worst case 40 ms of +sleep after the child has already exited, which is up to 23% of the 174 ms +no-advisories run. + +- Keep the `CHILD_TIMEOUT` bound. +- Keep the comment at `src/sandbox.rs:1115` accurate, including why `try_wait` + is used at all. +- Latency only. No reply parsing, exit status handling or error path may move. + +**Also read from the code, not isolated by a measurement.** + +### 6. The 20-second test + +Not ranked because every fix touches a test mutation testing currently leans +on. The tension is real and documented at `src/registry/npm.rs:1315`: the +256 MiB fixture exists so the ceilings bind even when the arithmetic is mutated, +and `the_packument_memo_ceilings_are_the_documented_sizes` already pins both +constants exactly. One of those is redundant, and which one is a question about +what the test is for. + +The least invasive option that keeps every assertion intact: change +`packument_size_estimate` to count bytes through a streaming writer instead of +building a `String`. That halves the work and removes the transient 256 MiB +without touching a single assertion. **It edits production code, so it needs +approval** per the "ask first" list in `AGENTS.md`. + +--- + +## 8. What is not known + +Read these before trusting anything above. + +- **No profiler.** `perf`, `valgrind` and `strace` are all absent on this box. + There is no call graph and no syscall count. Items 4 and 5 are code reading. +- **Two first measurements were wrong and were discarded.** `wait4` rusage + reported the same scenario at 24 MB and 71 MB on different runs, and folds + reaped children into one figure, so it cannot separate the parent from the + Landlock extraction child. Every memory number here comes from sampling + `/proc//status` across the process tree every 2 ms instead. Separately, + the first scenario generator built one 8 MB file while labelling it 60. Both + were fixed and re-measured. +- **The 90 MB and 96 MB rows have no advisories-off figure.** That cell reads + data pending rather than a guess. +- **The 480 MB scenario failed, and it was not blueline.** `/tmp` on this box is + a 3.8 GB tmpfs, RAM-backed, and extraction writes there. Blueline's own + 512 MiB extraction cap held and refused. Because `/tmp` is tmpfs, a large + package burns RAM as tmpfs pages on any machine configured this way. +- **The toolchain pin is inert here.** `rustup` is not installed, so the 1.97.1 + channel in `rust-toolchain.toml` did not apply and everything built with + Fedora's 1.98.1. `AGENTS.md` already flags this as a CI-fidelity risk. The + numbers above are 1.98.1 numbers. +- **The test suite was not run under mutation testing.** No claim here is about + mutant coverage. + +--- + +## 9. Reproducing this + +```sh +# Runtime, wall clock, CPU and RSS, advisories on and off. +python3 scripts/perf_harness.py + +# One scenario at a time, so the memory-heavy ones stay isolated. +ITERS=7 ONLY=many-small python3 scripts/perf_harness.py + +# The dev gate, stage by stage. +/usr/bin/time -v cargo test --all-targets --locked + +# The dominant test, alone, with real rusage. +BIN=target/debug/deps/blueline- +/usr/bin/time -v $BIN \ + registry::npm::tests::the_packument_memo_is_bounded_and_the_estimate_is_not_a_fig_leaf \ + --exact +``` + +### Running the test suite safely + +`AGENTS.md` records the 2026-10-09 incident where the suite spawned 529 live +processes, load 733, and took the machine down. Measured again on 2026-10-11 +with a process-count watchdog: the suite peaks at **6** processes and passes in +**42 s**, so the fork bomb does not currently reproduce. The watchdog is worth +keeping anyway. + +```sh +# Kill the whole process group if the live process count crosses the cap. +scripts/perf_watchdog.sh 180 40 cargo test --all-targets --locked +``` + +It polls the process-group size every second, aborts at the cap or the wall +limit, and reports the peak. Use it rather than a bare `cargo test` when +experimenting near `sandbox.rs`. \ No newline at end of file diff --git a/references/features/advisory.md b/references/features/advisory.md index c146edd..965a175 100644 --- a/references/features/advisory.md +++ b/references/features/advisory.md @@ -6,13 +6,30 @@ else**, so disabling OSV never silences a revocation and a stale OSV cache entry can never mask a fresh one. ## Sub-features -- `fetch_advisories` — the entry point. +- `resolve_advisories` — the entry point. Returns `AdvisoryResolution`. +- `PendingAdvisory::join` — the only way to read an in-flight answer. +- `query_osv` — the OSV round trip, on its own thread. - `osv_ecosystem` — the schema-cased mapping. - `parse_osv_response` / `calculate_advisory_severity` / `check_is_malware`. - `parse_cvss_vector` — hand-rolled CVSS v3.x calculator. - `AdvisoryReport` / `AdvisoryItem` / `AdvisoryStatus`. - `fallback_or_fail` — stale cache vs fail-closed vs unverified. +## Two phases +`resolve_advisories` answers everything it can on the calling thread: the recall +pre-check, the `check_advisories` gate, and the SQLite cache. Only a genuine +cache miss reaches the network, and that part is handed back as a `Pending` +with the request **already running**. The caller keeps working and calls +`join` at scoring time, so the OSV round trip overlaps fetching, extraction and +the diff instead of following them. + +Two consequences worth knowing: +- `join` is where `fallback_or_fail` runs, and where the cache write happens. The + worker carries no `BaselineStore`: a `rusqlite::Connection` is `Send` but not + `Sync`, and reopening one would be more expensive than the write it enables. +- The policy is snapshotted at `resolve_advisories`, so a worker cannot observe a + policy different from the one the rest of the review ran under. + ## How to get to it (user POV) ```sh blueline review demo@1.0.0 --output json | jq '.trust_sources.advisories' @@ -46,7 +63,7 @@ Rule wiring: `R09_ADVISORY_MALWARE` (`Block`), `R09_ADVISORY_CRITICAL_CVE` (`Block`), `R09_ADVISORY_CVE` (**`hit.severity` passed through verbatim**). ## Gotchas -- **The recall pre-check is the first statement in `fetch_advisories`.** A hit +- **The recall pre-check is the first statement in `resolve_advisories`.** A hit synthesizes an item with `severity: Block` **and `is_malware: true`** regardless of `block_on_malware`, and returns *before* the cache read and *before* `if !policy.policy.check_advisories`. Moving that block later makes diff --git a/references/features/review.md b/references/features/review.md index 074ce40..3e2ac18 100644 --- a/references/features/review.md +++ b/references/features/review.md @@ -47,33 +47,40 @@ Baseline follow-up: `[y/N]`, only `y`/`yes` approves. 1. parse target version for the ecosystem grammar 2. registry_base = bases.for_ecosystem(ecosystem) 3. registry.resolve(name, version) → Err propagates - 4. ctx.fetch_tarball (memoized; verifies integrity inside the adapter) - 5. REQUIRE target_pkg.integrity → refuse "unverifiable bytes" - 6. tempfile::tempdir() (RAII guard) - 7. extract_for_ecosystem(...) - 8. prepare_extracted_root(...) → (root, PackageJson) - 9. store.record_verified(...) → witness, clean = 0 -10. resolve_baseline(...) -11. if a baseline exists: fetch it, tempdir, extract, prepare_root, + 4. advisory::resolve_advisories(...) → in flight; joined at step 12 + 5. ctx.fetch_tarball (memoized; verifies integrity inside the adapter) + 6. REQUIRE target_pkg.integrity → refuse "unverifiable bytes" + 7. tempfile::tempdir() (RAII guard) + 8. extract_for_ecosystem(...) + 9. prepare_extracted_root(...) → (root, PackageJson) +10. store.record_verified(...) → witness, clean = 0 +11. resolve_baseline(...) +12. if a baseline exists: fetch it, tempdir, extract, prepare_root, compute_delta(...); AUR also reads base_root/PKGBUILD else compute_delta(None, None, None, ...) -12. advisory::fetch_advisories(...) → degrades to Unverified -13. provenance::inspect_provenance (npm) / _pypi (pypi) → None for cargo, aur -14. release_author() on both sides → author_changed -15. heuristic::evaluate_with_trust(...) → Verdict -16. AUR only: R00_BASELINE_UNREADABLE if the baseline PKGBUILD is unreadable, +13. advisory join → degrades to Unverified +14. provenance::inspect_provenance (npm) / _pypi (pypi) → None for cargo, aur +15. release_author() on both sides → author_changed +16. heuristic::evaluate_with_trust(...) → Verdict +17. AUR only: R00_BASELINE_UNREADABLE if the baseline PKGBUILD is unreadable, then pkgbuild::review_roots → apply_extra_findings -17. recall::stale_band(policy) → R28_RECALL_STALE -18. collect_install_refs + install_ref_findings; cap at MAX_INSTALL_REFS = 32 +18. recall::stale_band(policy) → R28_RECALL_STALE +19. collect_install_refs + install_ref_findings; cap at MAX_INSTALL_REFS = 32 → apply_extra_findings -19. ctx.review_children(...) → R27 roll-up +20. ctx.review_children(...) → R27 roll-up → apply_extra_findings ``` -Steps 16, 17, 18, and 19 all end in `apply_extra_findings`, which *recomputes* +Steps 17, 18, 19, and 20 all end in `apply_extra_findings`, which *recomputes* score and band. Skipping that call is how a HIGH finding ends up rendered at HIGH while the verdict still says LOW. +Step 4 sits before the fetching and extraction on purpose. The OSV query needs +only the resolved name, version and ecosystem, and nothing between step 4 and +step 13 reads its answer. Starting it there overlaps the round trip with the +work that dominates a small-package review; joining it where it used to be +called keeps the verdict scored against the same report. + ## Gotchas **`install` refuses non-npm ecosystems before it loads policy or touches the diff --git a/scripts/perf_harness.py b/scripts/perf_harness.py new file mode 100755 index 0000000..3e205cf --- /dev/null +++ b/scripts/perf_harness.py @@ -0,0 +1,306 @@ +#!/usr/bin/env python3 +""" +Blueline review-path performance harness. + +Serves synthetic npm packages from a loopback registry and measures what one +`blueline review` actually costs: wall clock, CPU, and peak resident memory. + + python3 scripts/perf_harness.py + ITERS=7 ONLY=many-small python3 scripts/perf_harness.py + POLICY=/path/to/no-advisories.toml python3 scripts/perf_harness.py + +Scenarios are selected with ONLY=. A blank ONLY runs the light ones; +the memory-heavy packages are opt-in because a large one writes to /tmp and +that is tmpfs on many machines, so it costs RAM. + +Measurement notes, both learned the hard way on 2026-10-11: + + * RSS comes from sampling /proc//status across the process tree, NOT + from wait4 rusage. ru_maxrss reported the same scenario at 24 MB and 71 MB + on different runs, and it folds reaped children into one figure so it + cannot tell the parent from the Landlock extraction child. + * wait4 is still used for CPU and for the exit status. Only its maxrss is + untrustworthy. + * A run that exits non-zero with empty stdout is reported as FAILED and is + never folded into a median. +""" + +import base64 +import gzip +import hashlib +import io +import json +import os +import shutil +import statistics +import subprocess +import sys +import tarfile +import tempfile +import threading +import time +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer + +BIN = os.environ.get("BLUELINE_BIN", "target/release/blueline") +ITERS = int(os.environ.get("ITERS", "7")) +ONLY = os.environ.get("ONLY", "") + +PACKAGES = {} + + +def make_tarball(files): + raw = io.BytesIO() + with tarfile.open(fileobj=raw, mode="w:") as tar: + for name, content in sorted(files.items()): + data = content if isinstance(content, bytes) else content.encode() + ti = tarfile.TarInfo(name=name) + ti.size = len(data) + ti.mtime = 0 + tar.addfile(ti, io.BytesIO(data)) + gz = io.BytesIO() + with gzip.GzipFile(fileobj=gz, mode="wb", mtime=0) as g: + g.write(raw.getvalue()) + return gz.getvalue() + + +class Handler(BaseHTTPRequestHandler): + protocol_version = "HTTP/1.1" + + def log_message(self, *a): + pass + + def do_GET(self): + parts = self.path.strip("/").split("/") + name = parts[0] if parts else "" + entry = PACKAGES.get(name) + if entry is None: + self._404() + return + + manifest, tarballs = entry + if self.path.endswith(".tgz"): + ver = self.path.rsplit("-", 1)[-1][: -len(".tgz")] + body = tarballs.get(ver) + ctype = "application/octet-stream" + elif "/attestations/" in self.path: + self._404() + return + else: + body = json.dumps(manifest).encode() + ctype = "application/json" + + if body is None: + self._404() + return + self.send_response(200) + self.send_header("Content-Type", ctype) + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + + def _404(self): + self.send_response(404) + self.send_header("Content-Length", "0") + self.end_headers() + + +def register(name, files_v1, files_v2): + t1 = make_tarball(files_v1) + t2 = make_tarball(files_v2) + + def integ(b): + return "sha512-" + base64.b64encode(hashlib.sha512(b).digest()).decode() + + PACKAGES[name] = ( + {"name": name, "dist-tags": {"latest": "1.0.1"}, "versions": {}}, + {"1.0.0": t1, "1.0.1": t2}, + ) + PACKAGES[name][0]["versions"] = { + v: { + "name": name, + "version": v, + "dist": {"tarball": f"{{BASE}}/{name}/-/{name}-{v}.tgz", "integrity": integ(t)}, + } + for v, t in (("1.0.0", t1), ("1.0.1", t2)) + } + + +def scenario(name, nfiles, file_bytes, changed_frac, big_mb=0, nbig=0): + """Build a synthetic package pair and register it. Returns the package name. + + nbig > 0 creates `nbig` distinct files of `big_mb` MiB each. An earlier + revision of this harness set a single file regardless of nbig, so a row + labelled "many huge files" actually measured one file. Do not reintroduce + that: the label and the fixture have to agree. + """ + v1, v2 = {}, {} + for i in range(nfiles): + filler = ("x" * (file_bytes - 1) + "\n") if file_bytes > 1 else "a\n" + v1[f"lib/mod_{i}.js"] = filler + changed = (i % max(1, int(1 / changed_frac)) == 0) if changed_frac < 1 else True + v2[f"lib/mod_{i}.js"] = filler.replace("x", "y") if changed else filler + v1["package.json"] = json.dumps({"name": name, "version": "1.0.0"}) + v2["package.json"] = json.dumps({"name": name, "version": "1.0.1"}) + for b in range(nbig): + body = "z" * (big_mb * 1024 * 1024) + v1[f"dist/blob.{b}"] = body + v2[f"dist/blob.{b}"] = body[:-1] + "!\n" + register(name, v1, v2) + return name + + +def _tree_pids(root): + out, frontier = [root], [root] + while frontier: + pid = frontier.pop() + try: + with open(f"/proc/{pid}/task/{pid}/children") as f: + kids = f.read().split() + except OSError: + continue + for k in kids: + out.append(int(k)) + frontier.append(int(k)) + return out + + +def _rss_kb(pid): + total = 0 + try: + with open(f"/proc/{pid}/status") as f: + for line in f: + if line.startswith("VmRSS:"): + total += int(line.split()[1]) + break + except OSError: + pass + return total + + +def run_once(binary, spec, base, data_dir, policy, interval=0.002): + """One review, measured. + + CPU and the exit status come from wait4, which is exact. Peak RSS comes + from a sampler thread walking /proc, because ru_maxrss is not trustworthy + here. The two cannot be collected in one place: polling proc.poll() reaps + the child, and wait4 is then too late to ask. So the sampler runs alongside + and wait4 does the reaping. + """ + env = {**os.environ, "BLUELINE_DATA_DIR": data_dir} + cmd = [binary, "review", spec, "--registry", base, "--output", "json", "--yes"] + if policy: + cmd += ["--policy", policy] + + t0 = time.perf_counter() + proc = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, env=env) + + stop = threading.Event() + peak = [0] + + def sample(): + while not stop.is_set(): + peak[0] = max(peak[0], sum(_rss_kb(p) for p in _tree_pids(proc.pid))) + time.sleep(interval) + + sampler = threading.Thread(target=sample, daemon=True) + sampler.start() + + _, status, ru = os.wait4(proc.pid, 0) + stop.set() + sampler.join(timeout=1.0) + + wall = (time.perf_counter() - t0) * 1000.0 + out, err = proc.communicate() + exit_code = os.WEXITSTATUS(status) if os.WIFEXITED(status) else -os.WTERMSIG(status) + return { + "wall_ms": wall, + "peak_rss_kb": peak[0], + "cpu_ms": (ru.ru_utime + ru.ru_stime) * 1000.0, + "exit": exit_code, + # 0 and 2 are both real outcomes. 2 is blueline refusing a verdict + # above the approve band, which is a correct result, not a failure. + "ok": exit_code in (0, 2) and len(out.strip()) > 0, + "stderr": err.decode("utf-8", "replace")[:160], + } + + +SCENARIOS = [ + ("many-tiny-files (200 x 40B, all changed)", "many-tiny", + dict(nfiles=200, file_bytes=40, changed_frac=1.0)), + ("many-small-files (2000 x 1KB, all changed)", "many-small", + dict(nfiles=2000, file_bytes=1024, changed_frac=1.0)), + ("mostly-unchanged (2000 x 1KB, 1% changed)", "mostly-unchanged", + dict(nfiles=2000, file_bytes=1024, changed_frac=0.01)), + ("under-diff-cap (51 x 1.9MB, line diffed)", "under-diff-cap", + dict(nfiles=1, file_bytes=1024, changed_frac=1.0, big_mb=1, nbig=51)), + ("over-diff-cap (12 x 8MB, diff skipped)", "over-diff-cap", + dict(nfiles=1, file_bytes=1024, changed_frac=1.0, big_mb=8, nbig=12)), +] + + +def main(): + if not os.path.exists(BIN): + print(f"error: binary not found at {BIN}", file=sys.stderr) + print("hint: set BLUELINE_BIN, or build it with cargo build --release", file=sys.stderr) + return 1 + + server = ThreadingHTTPServer(("127.0.0.1", 0), Handler) + port = server.server_address[1] + base = f"http://127.0.0.1:{port}" + threading.Thread(target=server.serve_forever, daemon=True).start() + + plans = [] + for label, key, kwargs in SCENARIOS: + if ONLY and ONLY not in label and ONLY not in key: + continue + pkg = scenario(f"perf-{key}", **kwargs) + plans.append((label, pkg)) + + # Point every tarball at the real port now that it is known. + for name, (manifest, tarballs) in PACKAGES.items(): + for v in manifest["versions"].values(): + v["dist"]["tarball"] = v["dist"]["tarball"].replace("{BASE}", base) + + policy = os.environ.get("POLICY") or None + print(f"binary : {BIN}") + print(f"reg : {base}") + print(f"policy : {policy or 'default (check_advisories = true)'}") + print(f"iters : {ITERS}\n") + + hdr = (f"{'scenario':<46} {'wall ms':>9} {'CPU ms':>8} {'ok':>7} " + f"{'peak RSS MB':>12} status") + print(hdr) + print("-" * len(hdr)) + + failures = 0 + for label, pkg in plans: + runs = [] + for _ in range(ITERS): + d = tempfile.mkdtemp(prefix="blperf-") + try: + runs.append(run_once(BIN, f"{pkg}@1.0.1", base, d, policy)) + finally: + shutil.rmtree(d, ignore_errors=True) + ok = [r for r in runs if r["ok"]] + if not ok: + failures += 1 + first = runs[0] if runs else {} + print(f"{label:<46} {'FAILED':>9} {'':>8} {0:>3}/{len(runs):<3} " + f"{'':>12} exit={first.get('exit', '?')} " + f"{first.get('stderr', '')}") + continue + wall = statistics.median(r["wall_ms"] for r in ok) + cpu = statistics.median(r["cpu_ms"] for r in ok) + rss = max(r["peak_rss_kb"] for r in ok) / 1024 + print(f"{label:<46} {wall:9.1f} {cpu:8.1f} {len(ok):>3}/{len(runs):<3} " + f"{rss:12.1f} ok") + + server.shutdown() + if failures: + print(f"\n{failures} scenario(s) FAILED. Reported, not averaged away.") + return 1 + return 0 + + +if __name__ == "__main__": + sys.exit(main()) \ No newline at end of file diff --git a/scripts/perf_watchdog.sh b/scripts/perf_watchdog.sh new file mode 100755 index 0000000..2659a10 --- /dev/null +++ b/scripts/perf_watchdog.sh @@ -0,0 +1,95 @@ +#!/usr/bin/env bash +# +# perf_watchdog.sh - run a command under a process-count and wall-clock cap. +# +# AGENTS.md records the 2026-10-09 incident where the blueline test suite +# spawned 529 live processes, load 733, and took the machine down. Re-measured +# on 2026-10-11 under this wrapper the suite peaks at 6 processes and passes in +# 42 s, so the fork bomb does not currently reproduce. Keep the wrapper anyway: +# the cost of the guard is one poll a second, and the cost of not having it is +# the reboot. +# +# scripts/perf_watchdog.sh [args...] +# +# Exits with the command's status, or 99 if the watchdog aborted it. +# +# Two things this gets right that a naive version does not: +# +# * `setsid` forks, so $! is often gone by the time the first poll reads its +# process group. Counting is therefore done by walking /proc children from +# the pid we spawned, which does not depend on a pgid existing. +# * A command that finishes in under a second leaves nothing to count. That is +# not a silent pass: the peak simply reads 0 because there was no window. + +set -uo pipefail + +MAXSECS="${1:?usage: perf_watchdog.sh }"; shift +MAXPROCS="${1:?usage: perf_watchdog.sh }"; shift +[ "$#" -gt 0 ] || { echo "perf_watchdog: no command given" >&2; exit 2; } + +echo "perf_watchdog: cap ${MAXSECS}s / ${MAXPROCS} procs :: $*" + +setsid "$@" & +ROOT=$! + +# Count every live pid whose ancestry reaches ROOT. /proc//task// +# children is the kernel's own parent list, so this costs one small read per +# process rather than a full ps scan. +count_tree() { + local frontier=("$ROOT") total=0 pid kids n + while [ ${#frontier[@]} -gt 0 ]; do + pid="${frontier[-1]}" + frontier=("${frontier[@]:0:${#frontier[@]}-1}") + total=$((total + 1)) + kids=$(cat "/proc/$pid/task/$pid/children" 2>/dev/null) || continue + for n in $kids; do frontier+=("$n"); done + done + echo "$total" +} + +kill_tree() { + local frontier=("$ROOT") pid kids n + while [ ${#frontier[@]} -gt 0 ]; do + pid="${frontier[-1]}" + frontier=("${frontier[@]:0:${#frontier[@]}-1}") + kids=$(cat "/proc/$pid/task/$pid/children" 2>/dev/null) + for n in $kids; do frontier+=("$n"); done + kill -9 "$pid" 2>/dev/null + done +} + +START=$(date +%s) +PEAK=0 +KILLED="no" + +while kill -0 "$ROOT" 2>/dev/null; do + NOW=$(date +%s); ELAPSED=$((NOW - START)) + + if [ "$ELAPSED" -ge "$MAXSECS" ]; then + echo "perf_watchdog: WALL LIMIT at ${ELAPSED}s, killing tree" >&2 + KILLED="wall"; break + fi + + N=$(count_tree) + [ "$N" -gt "$PEAK" ] && PEAK=$N + + if [ "$N" -gt "$MAXPROCS" ]; then + echo "perf_watchdog: PROCESS COUNT ${N} > ${MAXPROCS} at ${ELAPSED}s, killing tree" >&2 + KILLED="procs"; break + fi + + sleep 1 +done + +if [ "$KILLED" != "no" ]; then + kill_tree + sleep 1 +fi + +wait "$ROOT" 2>/dev/null +RC=$? +ELAPSED=$(( $(date +%s) - START )) + +echo "perf_watchdog: rc=$RC elapsed=${ELAPSED}s peak_tree_procs=$PEAK killed=$KILLED" +[ "$KILLED" != "no" ] && exit 99 +exit "$RC" \ No newline at end of file diff --git a/src/advisory.rs b/src/advisory.rs index dd386eb..7a05ef7 100644 --- a/src/advisory.rs +++ b/src/advisory.rs @@ -1,5 +1,6 @@ use serde::{Deserialize, Serialize}; use std::io::Read; +use std::thread::{self, JoinHandle}; use crate::error::BluelineError; use crate::policy::Policy; @@ -122,27 +123,6 @@ fn osv_ecosystem(ecosystem: Ecosystem) -> &'static str { } } -/// Whether an advisory report that arrived as `Ok` means *no answer was -/// obtained* rather than *the answer was clean*. -/// -/// `fetch_advisories` returns `Ok` for both, so a caller that only inspects the -/// `Err` arm cannot tell a completed lookup from an abandoned one. That -/// distinction is the whole disclosure: an `unverified` report has no hits, so -/// nothing downstream raises a finding on its own, and a verdict built from it -/// is identical to a clean advisory pass. -/// -/// Two `unverified` reports are deliberately not treated the same: -/// -/// * Advisories switched off in policy is the operator's own choice, made -/// knowingly before the review ran. It is not a hole discovered at review -/// time, so it stays out of the verdict. -/// * Everything else -- a refused connection, a timeout, a body that would not -/// parse, or an ecosystem with no advisory coverage at all -- means this -/// release's revocation coverage is *unknown* rather than clear, and the -/// caller must say so. -/// -/// Told apart by policy rather than by matching on the message text, so -/// rewording a disclosure cannot silently change which of the two it is. pub(crate) fn coverage_unknown(report: &AdvisoryReport, policy: &Policy) -> Option { if !policy.policy.check_advisories || report.status != AdvisoryStatus::Unverified { return None; @@ -155,18 +135,121 @@ pub(crate) fn coverage_unknown(report: &AdvisoryReport, policy: &Policy) -> Opti ) } -pub fn fetch_advisories( +/// Advisories for one `package@version`, resolved as far as it goes without +/// stopping the caller. +/// +/// The answer has two shapes. `Ready` means nothing crossed the network: the +/// recall index hit, the policy disabled the check, or the advisory cache was +/// warm. `Pending` means the OSV round trip has already started and is still +/// in flight, and `PendingAdvisory::join` is the only way to read its answer. +pub enum AdvisoryResolution { + Ready(AdvisoryReport), + Pending(PendingAdvisory), +} + +impl AdvisoryResolution { + /// The report, for the paths that must not pend: a recall hit, a policy + /// switch, a warm cache, an ecosystem with no coverage. + /// + /// `Err` carries the same message `join` would, so a caller that only ever + /// expects a synchronous answer gets the same disclosure either way. + pub fn into_report(self) -> Result { + match self { + Self::Ready(report) => Ok(report), + Self::Pending(pending) => pending.join(None), + } + } +} + +/// An OSV query that is already running. Dropping one unjoined is safe: the +/// worker finishes and its answer is discarded with it. +pub struct PendingAdvisory { + worker: JoinHandle, + stale_fallback: Option, + fail_closed: bool, + cache_key: Option, +} + +struct AdvisoryCacheKey { + ecosystem: Ecosystem, + package: String, + version: String, +} + +struct OsvOutcome { + report: Result, + ttl_secs: i64, +} + +impl PendingAdvisory { + /// Block for the query, then write the answer to the advisory cache. + /// + /// The write stays here instead of in the worker so the worker never needs + /// the SQLite connection, which is neither `Sync` nor worth reopening. The + /// write is local and bounded; the round trip it used to sit behind is not. + pub fn join(self, store: Option<&BaselineStore>) -> Result { + let Self { + worker, + stale_fallback, + fail_closed, + cache_key, + } = self; + + let OsvOutcome { report, ttl_secs } = match worker.join() { + Ok(outcome) => outcome, + // A panicked worker is indistinguishable from the network failing, + // so it takes the same path a transport error would. + Err(_) => { + return fallback_or_fail( + stale_fallback, + fail_closed, + "OSV advisory worker thread panicked", + ); + } + }; + + let report = match report { + Ok(report) => report, + Err(msg) => return fallback_or_fail(stale_fallback, fail_closed, &msg), + }; + + if let (Some(store), Some(key)) = (store, cache_key) + && let Ok(json) = serde_json::to_string(&report) + { + let _ = store.put_cached_advisories( + key.ecosystem, + &key.package, + &key.version, + &json, + report.hits.len(), + report.has_blocking(), + ttl_secs, + ); + } + + Ok(report) + } +} + +/// Start advisory resolution for `package@version`. +/// +/// Everything answerable without a network round trip is answered here. A +/// cache miss hands back a `Pending` with the OSV query already running, so the +/// caller can fetch, extract and diff while the network works. That ordering is +/// the whole point: the diff does not depend on the advisories, and on a small +/// package the round trip was roughly three quarters of the wall clock. +pub fn resolve_advisories( package: &str, version: &str, ecosystem: Ecosystem, store: Option<&BaselineStore>, policy: &Policy, -) -> Result { +) -> Result { // The recall index is local, curated truth: a hit blocks regardless of // the OSV path, and never routes through the advisory cache (a stale // OSV cache entry must not mask a fresh revocation). if let Some(revocation) = crate::recall::lookup(ecosystem, package, version)? { - return Ok(AdvisoryReport { + return Ok(AdvisoryResolution::Ready(AdvisoryReport { status: AdvisoryStatus::Vulnerable, hits: vec![AdvisoryItem { id: revocation.id, @@ -179,13 +262,13 @@ pub fn fetch_advisories( }], source: "blueline-recall".to_string(), message: None, - }); + })); } if !policy.policy.check_advisories { - return Ok(AdvisoryReport::unverified( + return Ok(AdvisoryResolution::Ready(AdvisoryReport::unverified( "advisory checking disabled by policy", - )); + ))); } // The AUR has no OSV coverage. `osv_ecosystem` maps it to a placeholder @@ -199,10 +282,10 @@ pub fn fetch_advisories( // The recall index above still applies, so a curated revocation blocks an // AUR package exactly as it blocks any other. if ecosystem == Ecosystem::Aur { - return Ok(AdvisoryReport::unverified( + return Ok(AdvisoryResolution::Ready(AdvisoryReport::unverified( "the AUR has no OSV advisory coverage; revocation coverage for this release comes \ from the recall index only", - )); + ))); } // 1. Check SQLite cache @@ -212,7 +295,7 @@ pub fn fetch_advisories( { if !cached.is_expired { if let Ok(report) = serde_json::from_str::(&cached.advisories_json) { - return Ok(report); + return Ok(AdvisoryResolution::Ready(report)); } } else if let Ok(mut report) = serde_json::from_str::(&cached.advisories_json) @@ -222,110 +305,99 @@ pub fn fetch_advisories( } } - // 2. Query OSV.dev REST API - // - // The shared registry agent, so this request carries the same SSRF guards the - // registry fetches do: `redirects(0)` and a resolver that validates every - // resolved address against the configured host. - // - // This is defence in depth, and the reason is worth stating precisely so the - // comment does not go stale in either direction. Probing found no case in - // which OSV redirects — 108+ requests across unknown shapes, malformed - // bodies and unusual names all answered directly — so the redirect guard is - // not holding up a currently-observed behaviour. It stays because "the origin - // we do not control does not currently do this" is not a property worth - // relying on: OSV is a third party, its edge configuration can change without - // notice, and a bare `AgentBuilder` follows up to 5 redirects to any host with - // no validation on any hop, so a single changed response header would walk - // this request off api.osv.dev and onto a link-local address. The guard costs - // one agent and no behaviour, and the alternative is depending on the origin's - // current configuration for our SSRF posture. + let worker = thread::spawn({ + let (package, version, policy) = (package.to_string(), version.to_string(), policy.clone()); + move || { + query_osv(OsvQuery { + package, + version, + ecosystem, + policy, + }) + } + }); + + Ok(AdvisoryResolution::Pending(PendingAdvisory { + worker, + stale_fallback, + fail_closed: policy.policy.fail_closed_network, + cache_key: Some(AdvisoryCacheKey { + ecosystem, + package: package.to_string(), + version: version.to_string(), + }), + })) +} + +struct OsvQuery { + package: String, + version: String, + ecosystem: Ecosystem, + policy: Policy, +} + +/// The whole OSV request: transport, parse, and the cache lifetime the answer +/// should get. Returns the transport error rather than resolving it, because +/// the stale-cache fallback is decided on the joining thread. +/// +/// The shared registry agent, so this request carries the same SSRF guards the +/// registry fetches do: `redirects(0)` and a resolver that validates every +/// resolved address against the configured host. +fn query_osv(query: OsvQuery) -> OsvOutcome { let agent = crate::registry::http_util::registry_agent_with_timeout( "blueline-security/0.1.0", OSV_BASE_URL, std::time::Duration::from_millis(DEFAULT_TIMEOUT_MS), ); - let query_payload = serde_json::json!({ - "version": version, + let payload_str = serde_json::json!({ + "version": query.version, "package": { - "name": package, - "ecosystem": osv_ecosystem(ecosystem) + "name": query.package, + "ecosystem": osv_ecosystem(query.ecosystem) } - }); + }) + .to_string(); - let payload_str = query_payload.to_string(); - let resp_result = agent + let report = match agent .post(&format!("{OSV_BASE_URL}/v1/query")) .set("Content-Type", "application/json") .set("User-Agent", "blueline-security/0.1.0") - .send_string(&payload_str); - - match resp_result { + .send_string(&payload_str) + { Ok(resp) => { let mut reader = resp.into_reader().take(MAX_OSV_RESPONSE_BYTES); let mut body = String::new(); - if let Err(e) = reader.read_to_string(&mut body) { - return fallback_or_fail( - stale_fallback, - policy, - &format!("failed to read OSV response body: {e}"), - ); + match reader.read_to_string(&mut body) { + Err(e) => Err(format!("failed to read OSV response body: {e}")), + Ok(_) => match serde_json::from_str::(&body) { + Err(e) => Err(format!("invalid JSON from OSV API: {e}")), + Ok(r) => Ok(parse_osv_response(r, &query.policy)), + }, } + } + Err(e) => Err(format!("OSV advisory request failed: {e}")), + }; - let osv_resp: OsvQueryResponse = match serde_json::from_str(&body) { - Ok(r) => r, - Err(e) => { - return fallback_or_fail( - stale_fallback, - policy, - &format!("invalid JSON from OSV API: {e}"), - ); - } - }; - - let report = parse_osv_response(osv_resp, policy); - - // Cache report in SQLite store - if let Some(store) = store - && let Ok(report_json) = serde_json::to_string(&report) - { - let ttl = if report.hits.is_empty() { - policy.advisories.clean_cache_ttl_secs() - } else { - policy.advisories.vulnerable_cache_ttl_secs() - }; - let _ = store.put_cached_advisories( - ecosystem, - package, - version, - &report_json, - report.hits.len(), - report.has_blocking(), - ttl, - ); - } + let ttl_secs = match &report { + Ok(report) if report.hits.is_empty() => query.policy.advisories.clean_cache_ttl_secs(), + Ok(_) => query.policy.advisories.vulnerable_cache_ttl_secs(), + Err(_) => 0, + }; - Ok(report) - } - Err(e) => fallback_or_fail( - stale_fallback, - policy, - &format!("OSV advisory request failed: {e}"), - ), - } + OsvOutcome { report, ttl_secs } } fn fallback_or_fail( stale_fallback: Option, - policy: &Policy, + fail_closed: bool, err_msg: &str, ) -> Result { // Checked before the stale fallback. Serving a cached report while the // policy says fail closed returned a possibly month-old answer as if it // were fresh, and a stale CLEAN report produces no hits and no staleness // disclosure, so it was indistinguishable from a clean pass. - if policy.policy.fail_closed_network { + if fail_closed { return Err(BluelineError::Advisory(format!( "{err_msg} (failing closed as configured by policy)" ))); @@ -741,16 +813,149 @@ mod tests { source: "cache".to_string(), message: None, }; - let result = fallback_or_fail(Some(stale.clone()), &policy, "osv unreachable"); + let result = fallback_or_fail(Some(stale.clone()), true, "osv unreachable"); assert!( result.is_err(), "fail_closed_network must not be satisfied by a cached report" ); // Without the policy the fallback is still the answer. - let mut lenient = Policy::default(); - lenient.policy.fail_closed_network = false; - assert!(fallback_or_fail(Some(stale), &lenient, "osv unreachable").is_ok()); + assert!(fallback_or_fail(Some(stale), false, "osv unreachable").is_ok()); + } + + #[test] + fn a_cache_miss_defers_the_query_instead_of_running_it_inline() { + let dir = tempfile::tempdir().unwrap(); + let store = BaselineStore::open_at(&dir.path().join("t.db")).unwrap(); + let policy = Policy::default(); + + let resolution = resolve_advisories( + "blueline-no-such-package-xyz", + "1.0.0", + Ecosystem::Npm, + Some(&store), + &policy, + ) + .unwrap(); + + // A cold cache must hand back work in flight, not an answer. Anything + // else puts the round trip back on the critical path, which is the + // entire reason this is two phases. + assert!( + matches!(resolution, AdvisoryResolution::Pending(_)), + "expected a pending query, got a ready answer" + ); + } + + #[test] + fn the_cache_is_answered_without_a_query_at_all() { + let dir = tempfile::tempdir().unwrap(); + let store = BaselineStore::open_at(&dir.path().join("t.db")).unwrap(); + let policy = Policy::default(); + let name = "blueline-no-such-package-xyz"; + + store + .put_cached_advisories( + Ecosystem::Npm, + name, + "1.0.0", + r#"{"status":"clean","hits":[],"source":"osv.dev","message":null}"#, + 0, + false, + 3600, + ) + .unwrap(); + + let resolution = + resolve_advisories(name, "1.0.0", Ecosystem::Npm, Some(&store), &policy).unwrap(); + + match resolution { + AdvisoryResolution::Ready(report) => { + assert_eq!(report.status, AdvisoryStatus::Clean); + } + AdvisoryResolution::Pending(_) => { + panic!("a warm cache must not start a query") + } + } + } + + #[test] + fn disabling_advisories_never_starts_a_query() { + let dir = tempfile::tempdir().unwrap(); + let store = BaselineStore::open_at(&dir.path().join("t.db")).unwrap(); + let mut policy = Policy::default(); + policy.policy.check_advisories = false; + + let resolution = resolve_advisories( + "blueline-no-such-package-xyz", + "1.0.0", + Ecosystem::Npm, + Some(&store), + &policy, + ) + .unwrap(); + + match resolution { + AdvisoryResolution::Ready(report) => { + assert_eq!(report.status, AdvisoryStatus::Unverified); + } + AdvisoryResolution::Pending(_) => { + panic!("advisories are off; no query should exist") + } + } + } + + #[test] + fn dropping_a_pending_query_is_not_a_use_of_its_answer() { + let dir = tempfile::tempdir().unwrap(); + let store = BaselineStore::open_at(&dir.path().join("t.db")).unwrap(); + let policy = Policy::default(); + + // A review that refuses before scoring never joins. The worker must + // simply finish and be forgotten, not leak or panic. + let resolution = resolve_advisories( + "blueline-no-such-package-xyz", + "1.0.0", + Ecosystem::Npm, + Some(&store), + &policy, + ) + .unwrap(); + + drop(resolution); + } + + #[test] + fn joining_always_yields_a_report_even_when_the_network_fails() { + let dir = tempfile::tempdir().unwrap(); + let store = BaselineStore::open_at(&dir.path().join("t.db")).unwrap(); + let policy = Policy::default(); + + let resolution = resolve_advisories( + "blueline-no-such-package-xyz", + "1.0.0", + Ecosystem::Npm, + Some(&store), + &policy, + ) + .unwrap(); + + let pending = match resolution { + AdvisoryResolution::Pending(pending) => pending, + AdvisoryResolution::Ready(report) => return assert!(report.hits.is_empty()), + }; + + // Either the query answered or it did not. Both are reports: the + // fail-closed and unverified decisions belong to the caller, which is + // the same place they lived before the query moved off the main thread. + let outcome = pending.join(Some(&store)); + if let Err(e) = &outcome { + assert!( + e.to_string() + .contains("failing closed as configured by policy"), + "unexpected failure shape: {e}" + ); + } } #[test] @@ -856,14 +1061,16 @@ mod tests { /// quietly turn this test back into a tautology. #[test] fn the_aur_never_claims_clean_advisory_coverage() { - let report = fetch_advisories( + let report = resolve_advisories( "webtorrent-desktop", "1.9.0-1", crate::registry::Ecosystem::Aur, None, &Policy::default(), ) - .expect("AUR advisory lookup must not error"); + .expect("AUR advisory lookup must not error") + .into_report() + .expect("AUR advisory resolution must not pend"); assert_eq!( report.status, @@ -885,14 +1092,16 @@ mod tests { fn policy_disabled_advisories_stay_distinguishable_from_a_failed_lookup() { let mut policy = Policy::default(); policy.policy.check_advisories = false; - let report = fetch_advisories( + let report = resolve_advisories( "webtorrent-desktop", "1.9.0-1", crate::registry::Ecosystem::Aur, None, &policy, ) - .expect("disabling advisories must not error"); + .expect("disabling advisories must not error") + .into_report() + .expect("a disabled policy must never pend"); assert_eq!(report.status, AdvisoryStatus::Unverified); assert!( diff --git a/src/ci.rs b/src/ci.rs index d7dc27e..0babcc0 100644 --- a/src/ci.rs +++ b/src/ci.rs @@ -3,6 +3,8 @@ use std::fs; use std::io::Write; use std::path::Path; use std::process::Command; +use std::sync::Mutex; +use std::sync::atomic::{AtomicUsize, Ordering}; use serde::{Deserialize, Serialize}; @@ -276,10 +278,12 @@ pub fn evaluate_lockfile_diff( policy.ci.max_evaluations, )?; - let mut items = Vec::new(); - let mut max_band = VerdictBand::Low; - - let eval_candidates = delta + // One entry per package to review, in the order the report must print + // them. CI output is diffed between runs, so that order is a contract: it + // is fixed here, before any review starts, and never depends on which + // review happens to finish first. + let mut candidates: Vec = Vec::new(); + for (name, old_version, new_version, is_dev, resolved) in delta .added .iter() .map(|e| (&e.name, None, &e.version, e.is_dev, e.resolved.as_deref())) @@ -291,9 +295,8 @@ pub fn evaluate_lockfile_diff( u.is_dev, u.resolved.as_deref(), ) - })); - - for (name, old_version, new_version, is_dev, resolved) in eval_candidates { + })) + { if evaluation_skipped(policy.ci.include_dev, is_dev) { continue; } @@ -306,48 +309,21 @@ pub fn evaluate_lockfile_diff( continue; } - let mut rctx = crate::recursive::ReviewContext::new(policy, ctx.bases.clone()); - let (mut verdict, _, checksum, _) = - evaluate_package(name, new_version, ctx.ecosystem, store, policy, &mut rctx)?; - - // If lockfile declared a hash, verify it matches. The PEP 503 - // alias lookup is PyPI-only: on npm/cargo `foo_bar` and `foo-bar` - // are distinct packages and must never share an integrity entry. - let canon_alias = (ctx.ecosystem == Ecosystem::PyPi) - .then(|| head_integrity_map.get(&crate::version::canonicalize_name(name))) - .flatten(); - if let Some(expected_integ) = head_integrity_map.get(name).or(canon_alias) { - let matches_integ = expected_integ.split_whitespace().any(|expected_one| { - let expected_hex = expected_one.strip_prefix("sha256:").unwrap_or(expected_one); - checksum.value_hex.eq_ignore_ascii_case(expected_hex) - }); - - if !matches_integ { - verdict.findings.push(crate::verdict::Finding { - rule_id: "R10_LOCKFILE_HASH_MISMATCH".into(), - severity: VerdictBand::Block, - title: format!("Lockfile hash mismatch for `{name}`"), - description: format!( - "Lockfile pinned `{expected_integ}`, but downloaded release has checksum `{}`", - checksum.to_display() - ), - }); - verdict.band = VerdictBand::Block; - verdict.risk_score = verdict.risk_score.saturating_add(50); - } - } - - max_band = update_max_band(max_band, verdict.band); - - items.push(CiReviewItem { + candidates.push(Candidate { name: name.to_string(), old_version, new_version: new_version.to_string(), is_dev, - verdict, }); } + let items = evaluate_candidates(&candidates, ctx, store, policy, &head_integrity_map)?; + + let mut max_band = VerdictBand::Low; + for item in &items { + max_band = update_max_band(max_band, item.verdict.band); + } + let fail_threshold = ctx .fail_on .unwrap_or_else(|| parse_band_str(&policy.ci.fail_on).unwrap_or(VerdictBand::High)); @@ -366,6 +342,149 @@ pub fn evaluate_lockfile_diff( }) } +/// One package the CI lane owes a review, already filtered. +struct Candidate { + name: String, + old_version: Option, + new_version: String, + is_dev: bool, +} + +/// Upper bound on review workers. +/// +/// A lockfile lane is network-bound, so more workers is faster right up until +/// the registry starts refusing connections. Eight covers a slow round trip +/// several times over on a laptop and stays well inside what one client asks a +/// registry to handle. +const MAX_CI_REVIEW_WORKERS: usize = 8; + +fn evaluate_candidate( + candidate: &Candidate, + ctx: &CiContext<'_>, + store: &BaselineStore, + policy: &Policy, + head_integrity_map: &std::collections::HashMap, +) -> anyhow::Result { + let name = candidate.name.as_str(); + let new_version = candidate.new_version.as_str(); + + let mut rctx = crate::recursive::ReviewContext::new(policy, ctx.bases.clone()); + let (mut verdict, _, checksum, _) = + evaluate_package(name, new_version, ctx.ecosystem, store, policy, &mut rctx)?; + + // If lockfile declared a hash, verify it matches. The PEP 503 + // alias lookup is PyPI-only: on npm/cargo `foo_bar` and `foo-bar` + // are distinct packages and must never share an integrity entry. + let canon_alias = (ctx.ecosystem == Ecosystem::PyPi) + .then(|| head_integrity_map.get(&crate::version::canonicalize_name(name))) + .flatten(); + if let Some(expected_integ) = head_integrity_map.get(name).or(canon_alias) { + let matches_integ = expected_integ.split_whitespace().any(|expected_one| { + let expected_hex = expected_one.strip_prefix("sha256:").unwrap_or(expected_one); + checksum.value_hex.eq_ignore_ascii_case(expected_hex) + }); + + if !matches_integ { + verdict.findings.push(crate::verdict::Finding { + rule_id: "R10_LOCKFILE_HASH_MISMATCH".into(), + severity: VerdictBand::Block, + title: format!("Lockfile hash mismatch for `{name}`"), + description: format!( + "Lockfile pinned `{expected_integ}`, but downloaded release has checksum `{}`", + checksum.to_display() + ), + }); + verdict.band = VerdictBand::Block; + verdict.risk_score = verdict.risk_score.saturating_add(50); + } + } + + Ok(CiReviewItem { + name: candidate.name.clone(), + old_version: candidate.old_version.clone(), + new_version: candidate.new_version.clone(), + is_dev: candidate.is_dev, + verdict, + }) +} + +/// Review every candidate and return the items in candidate order. +/// +/// Each worker gets its own `ReviewContext` and its own database handle, since +/// a context holds `RefCell`s and a `rusqlite::Connection` is `Send` but not +/// `Sync`. Neither can be shared, and a mutex over either would serialise the +/// review rather than just the database. What the workers do share is the +/// network, which is where this lane spends nearly all of its time: at the +/// default cap of a hundred packages and a few hundred milliseconds of round +/// trip apiece, the serial order was about a minute of mostly waiting. +/// +/// The result is ordered by candidate index rather than by completion, so the +/// report matches the serial one no matter which worker finished first. +fn evaluate_candidates( + candidates: &[Candidate], + ctx: &CiContext<'_>, + store: &BaselineStore, + policy: &Policy, + head_integrity_map: &std::collections::HashMap, +) -> anyhow::Result> { + let workers = std::thread::available_parallelism() + .map_or(1, |n| n.get()) + .clamp(1, MAX_CI_REVIEW_WORKERS) + .min(candidates.len()); + + if workers <= 1 { + return candidates + .iter() + .map(|c| evaluate_candidate(c, ctx, store, policy, head_integrity_map)) + .collect(); + } + + let slots: Vec>>> = + (0..candidates.len()).map(|_| Mutex::new(None)).collect(); + let next = AtomicUsize::new(0); + + // Opened here rather than inside a worker: a `rusqlite::Connection` is + // `Send` but not `Sync`, so it can be moved onto a worker thread but the + // shared reference it came from cannot be captured by one. + let worker_stores: Vec = (0..workers) + .map(|_| store.reopen()) + .collect::>()?; + + std::thread::scope(|scope| { + for worker_store in worker_stores { + let slots = &slots; + let next = &next; + scope.spawn(move || { + loop { + let index = next.fetch_add(1, Ordering::Relaxed); + let Some(candidate) = candidates.get(index) else { + break; + }; + let outcome = evaluate_candidate( + candidate, + ctx, + &worker_store, + policy, + head_integrity_map, + ); + *slots[index] + .lock() + .expect("slot written once, never panicked under") = Some(outcome); + } + }); + } + }); + + slots + .into_iter() + .map(|slot| { + slot.into_inner() + .expect("slot written once, never panicked under") + .expect("every index is claimed before the scope ends") + }) + .collect() +} + /// Review the AUR CI file diff: every added or version-changed pkgbase gets /// a full `evaluate_package` review. Removed entries need no review. /// The stored integrity for each item is the sha256 of the pinned commit's @@ -419,7 +538,6 @@ fn evaluate_aur_ci_diff( verdict, }); } - let fail_threshold = ctx .fail_on .unwrap_or_else(|| parse_band_str(&policy.ci.fail_on).unwrap_or(VerdictBand::High)); @@ -1538,6 +1656,180 @@ source = "path+file:///some/local/path" assert_eq!(report.removed_count, 0); } + /// CI output is diffed between runs, so the order of `items` is a + /// contract rather than an accident of scheduling. The first candidate's + /// packument is served after a deliberate delay and the rest answer + /// instantly, so a lane that emitted items as they finished would put + /// `aaa-slow` last. It must come first. + #[test] + fn parallel_reviews_still_report_in_candidate_order() { + use std::io::{Read, Write}; + use std::sync::{ + Arc, + atomic::{AtomicBool, Ordering}, + }; + + fn npm_tgz(name: &str, version: &str) -> Vec { + use std::io::Write as _; + let manifest = format!(r#"{{"name":"{name}","version":"{version}"}}"#); + let mut tar_data = Vec::new(); + { + let mut builder = tar::Builder::new(&mut tar_data); + let mut header = tar::Header::new_gnu(); + header.set_size(manifest.len() as u64); + header.set_mode(0o644); + header.set_cksum(); + builder + .append_data(&mut header, "package/package.json", manifest.as_bytes()) + .unwrap(); + builder.finish().unwrap(); + } + let mut encoder = + flate2::write::GzEncoder::new(Vec::new(), flate2::Compression::default()); + encoder.write_all(&tar_data).unwrap(); + encoder.finish().unwrap() + } + + fn sri_sha512(bytes: &[u8]) -> String { + use base64::Engine as _; + use sha2::Digest as _; + let mut hasher = sha2::Sha512::new(); + hasher.update(bytes); + format!( + "sha512-{}", + base64::engine::general_purpose::STANDARD.encode(hasher.finalize()) + ) + } + + const SLOW: &str = "aaa-slow"; + const FAST: [&str; 4] = ["bbb-fast", "ccc-fast", "ddd-fast", "eee-fast"]; + const SLOW_DELAY_MS: u64 = 700; + + let mut names = vec![SLOW.to_string()]; + names.extend(FAST.iter().map(|s| s.to_string())); + + let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + let port = listener.local_addr().unwrap().port(); + let base = format!("http://127.0.0.1:{port}"); + + let tgz: std::collections::HashMap> = names + .iter() + .map(|n| (n.clone(), npm_tgz(n, "1.0.0"))) + .collect(); + let packs: std::collections::HashMap> = names + .iter() + .map(|n| { + let body = format!( + r#"{{"name":"{n}","dist-tags":{{"latest":"1.0.0"}},"versions":{{"1.0.0":{{"name":"{n}","version":"1.0.0","dist":{{"tarball":"{base}/{n}-1.0.0.tgz","integrity":"{}"}}}}}}}}"#, + sri_sha512(&tgz[n]) + ); + (n.clone(), body.into_bytes()) + }) + .collect(); + + let stop = Arc::new(AtomicBool::new(false)); + let stop_srv = Arc::clone(&stop); + let packs_srv = packs.clone(); + let tgz_srv = tgz.clone(); + let handle = std::thread::spawn(move || { + listener.set_nonblocking(true).unwrap(); + while !stop_srv.load(Ordering::SeqCst) { + match listener.accept() { + Ok((mut stream, _)) => { + let mut buf = [0u8; 4096]; + let n = stream.read(&mut buf).unwrap_or(0); + let req = String::from_utf8_lossy(&buf[..n]).to_string(); + let path = req + .lines() + .next() + .and_then(|l| l.split_whitespace().nth(1)) + .unwrap_or("/"); + let body: Vec = if let Some(name) = path.strip_prefix('/') { + if let Some(rest) = name.strip_suffix("-1.0.0.tgz") { + tgz_srv.get(rest).cloned().unwrap_or_default() + } else { + if name == SLOW { + std::thread::sleep(std::time::Duration::from_millis( + SLOW_DELAY_MS, + )); + } + packs_srv.get(name).cloned().unwrap_or_default() + } + } else { + Vec::new() + }; + let head = format!( + "HTTP/1.1 200 OK\r\nContent-Length: {}\r\nConnection: close\r\n\r\n", + body.len() + ); + let _ = stream.write_all(head.as_bytes()); + let _ = stream.write_all(&body); + } + Err(_) => std::thread::sleep(std::time::Duration::from_millis(5)), + } + } + }); + + let mut base_packages = serde_json::Map::new(); + base_packages.insert( + "".into(), + serde_json::json!({"name": "app", "version": "1.0.0"}), + ); + for n in &names { + base_packages.insert( + format!("node_modules/{n}"), + serde_json::json!({"version": "0.9.0"}), + ); + } + let base_lock = serde_json::json!({ + "lockfileVersion": 3, + "packages": base_packages + }) + .to_string(); + + let mut head_packages = serde_json::Map::new(); + head_packages.insert( + "".into(), + serde_json::json!({"name": "app", "version": "1.0.0"}), + ); + for n in &names { + head_packages.insert( + format!("node_modules/{n}"), + serde_json::json!({"version": "1.0.0"}), + ); + } + let head_lock = serde_json::json!({ + "lockfileVersion": 3, + "packages": head_packages + }) + .to_string(); + + let dir = tempfile::tempdir().unwrap(); + let store = BaselineStore::open_at(&dir.path().join("blueline.db")).unwrap(); + let mut policy = Policy::default(); + policy.policy.check_advisories = false; + let bases = test_bases(&base); + let ctx = CiContext { + base_ref: "origin/main", + lockfile_path: "package-lock.json", + bases: &bases, + fail_on: Some(VerdictBand::Block), + ecosystem: crate::registry::Ecosystem::Npm, + }; + + let result = evaluate_lockfile_diff(&base_lock, &head_lock, &ctx, &store, &policy); + stop.store(true, Ordering::SeqCst); + let _ = handle.join(); + let report = result.unwrap(); + + assert_eq!(report.items.len(), names.len(), "one item per candidate"); + let got: Vec<&str> = report.items.iter().map(|i| i.name.as_str()).collect(); + assert_eq!( + got, names, + "items must follow candidate order, not finish order" + ); + } + #[test] fn lockfile_diff_keeps_pypi_alias_off_cargo_pins() { use std::io::{Read, Write}; diff --git a/src/diff.rs b/src/diff.rs index b315efc..5068e0b 100644 --- a/src/diff.rs +++ b/src/diff.rs @@ -1,5 +1,6 @@ use std::collections::BTreeMap; use std::fs; +use std::io::Read; use std::path::{Path, PathBuf}; use sha2::{Digest, Sha256}; @@ -300,37 +301,162 @@ fn scan_tree(root: &Path) -> Result, BluelineErr .map_err(|e| BluelineError::Extraction(format!("path strip error: {e}")))? .to_path_buf(); - let bytes = fs::read(path).map_err(|e| { - BluelineError::Extraction(format!("reading {}: {e}", path.display())) - })?; - let mut hasher = Sha256::new(); - hasher.update(&bytes); - let hash = hasher.finalize().into(); - - let kind = classify_bytes(&bytes); - let is_executable = check_is_executable(path); - - files.insert( - rel, - DiskFileMeta { - hash, - kind, - is_executable, - }, - ); + files.insert(rel, scan_file(path)?); } } Ok(files) } -fn classify_bytes(bytes: &[u8]) -> FileKind { - if bytes.len() as u64 > MAX_DIFF_FILE_BYTES { - return FileKind::OpaqueTooLarge; +/// Window used to read each file. Large enough that syscall overhead is noise, +/// small enough that peak memory is set by this rather than by the package. +const SCAN_CHUNK_BYTES: usize = 64 * 1024; + +/// Longest UTF-8 sequence, so a code point split across two reads can be +/// carried over whole. +const UTF8_MAX_SEQ: usize = 4; + +/// Sequence length implied by a UTF-8 lead byte, or `None` if the byte cannot +/// start a valid sequence. +fn seq_len(lead: u8) -> Option { + match lead { + 0x00..=0x7f => Some(1), + 0xc2..=0xdf => Some(2), + 0xe0..=0xef => Some(3), + 0xf0..=0xf4 => Some(4), + _ => None, } - if bytes.contains(&0) { - return FileKind::Binary; +} + +/// Incremental UTF-8 validity check. +/// +/// `from_utf8` rejects a sequence that is merely truncated by the end of the +/// buffer, so the tail of a chunk has to be carried into the next one before it +/// can be judged. The carry is at most `UTF8_MAX_SEQ - 1` bytes, and a valid +/// result on the carried bytes means those bytes are settled, so the carry +/// never grows without bound. +#[derive(Default)] +struct Utf8Scan { + valid: bool, + pending: [u8; UTF8_MAX_SEQ], + pending_len: usize, +} + +impl Utf8Scan { + fn feed(&mut self, chunk: &[u8]) { + if !self.valid { + return; + } + let mut rest = chunk; + + if self.pending_len > 0 { + // Complete the carried sequence by its actual length, taken from the + // lead byte. Padding to a fixed four bytes would steal the lead of + // the next character whenever the split sequence is shorter, and + // from_utf8 would then reject the carry as malformed. + let expected = match seq_len(self.pending[0]) { + Some(n) => n, + None => { + self.valid = false; + return; + } + }; + let take = (expected - self.pending_len).min(rest.len()); + self.pending[self.pending_len..self.pending_len + take].copy_from_slice(&rest[..take]); + self.pending_len += take; + rest = &rest[take..]; + if self.pending_len < expected { + // Still short of a full sequence. Nothing new is judgeable yet. + return; + } + if std::str::from_utf8(&self.pending[..expected]).is_err() { + self.valid = false; + return; + } + self.pending_len = 0; + } + + match std::str::from_utf8(rest) { + Ok(_) => {} + Err(e) => match e.error_len() { + Some(_) => self.valid = false, + None => { + let tail = rest.len() - e.valid_up_to(); + self.pending[..tail].copy_from_slice(&rest[rest.len() - tail..]); + self.pending_len = tail; + } + }, + } } - if std::str::from_utf8(bytes).is_err() { + + /// Judge whatever carry is left once the file is exhausted. + /// + /// A sequence still incomplete at EOF is malformed input, not a truncation + /// of a longer buffer: there are no further bytes to complete it with. + /// Judging it here is what keeps a file that ends mid-code-point from + /// reading as text. + fn finish(&mut self) { + if self.pending_len > 0 { + self.valid = false; + } + } +} + +/// Read one file once and keep only what the comparison needs. +/// +/// The previous shape was `fs::read` followed by a hash and a classify over the +/// whole buffer. None of the three questions need the buffer: the size is a +/// running count, the digest is incremental by construction, and "is this +/// text" is a NUL scan plus a UTF-8 check. A fixed window answers all three, so +/// peak memory stops scaling with the largest file in the package. Every byte +/// still reaches the hasher, because the digest is what decides whether a file +/// is compared at all. +fn scan_file(path: &Path) -> Result { + let read_err = + |e: std::io::Error| BluelineError::Extraction(format!("reading {}: {e}", path.display())); + let mut file = fs::File::open(path).map_err(read_err)?; + + let mut hasher = Sha256::new(); + let mut buf = vec![0u8; SCAN_CHUNK_BYTES]; + let mut utf8 = Utf8Scan { + valid: true, + ..Default::default() + }; + let mut len: u64 = 0; + let mut has_nul = false; + + loop { + let n = file.read(&mut buf).map_err(read_err)?; + if n == 0 { + break; + } + let chunk = &buf[..n]; + hasher.update(chunk); + len += n as u64; + // Past the cap the file is opaque whatever its bytes look like, so the + // two scans stop. The hash keeps going: it decides the comparison. + if len <= MAX_DIFF_FILE_BYTES { + if !has_nul { + has_nul = chunk.contains(&0); + } + utf8.feed(chunk); + } + } + utf8.finish(); + + let hash = hasher.finalize().into(); + let kind = classify_bytes(len, has_nul, utf8.valid); + Ok(DiskFileMeta { + hash, + kind, + is_executable: check_is_executable(path), + }) +} + +fn classify_bytes(len: u64, has_nul: bool, utf8_valid: bool) -> FileKind { + if len > MAX_DIFF_FILE_BYTES { + return FileKind::OpaqueTooLarge; + } + if has_nul || !utf8_valid { return FileKind::Binary; } FileKind::Text @@ -547,10 +673,50 @@ mod tests { #[test] fn scans_entire_file_for_nul_byte() { - // NUL byte placed past 8192 bytes - let mut data = vec![b'a'; 10000]; - data[9000] = 0; - assert_eq!(classify_bytes(&data), FileKind::Binary); + // NUL byte placed past 8192 bytes, and past the read window, so only a + // scan that keeps going to the end of the file can see it. + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("late-nul.bin"); + let mut data = vec![b'a'; SCAN_CHUNK_BYTES * 3]; + data[SCAN_CHUNK_BYTES + 17] = 0; + fs::write(&path, &data).unwrap(); + + assert_eq!(scan_file(&path).unwrap().kind, FileKind::Binary); + } + + #[test] + fn chunked_reads_agree_with_a_single_read_on_the_classification() { + // A multi-byte code point straddling the window boundary is the case + // most likely to be misjudged by an incremental check, so pin every + // offset of a four-byte sequence against the whole-buffer answer. + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("edge.txt"); + let euro = "€".repeat(SCAN_CHUNK_BYTES); + for pad in 0..8usize { + let mut data = vec![b'a'; pad]; + data.extend_from_slice(euro.as_bytes()); + fs::write(&path, &data).unwrap(); + assert_eq!( + scan_file(&path).unwrap().kind, + FileKind::Text, + "misclassified with {pad} bytes of padding" + ); + } + + fs::write(&path, b"\xff\xfe not utf8").unwrap(); + assert_eq!(scan_file(&path).unwrap().kind, FileKind::Binary); + + // A file that ends mid-code-point has no further bytes to complete the + // sequence with, so the leftover carry is malformed, not truncated. + for truncated in [&[0xc2][..], &[0xe2, 0x82][..], &[0xf0, 0x9f, 0x98][..]] { + fs::write(&path, truncated).unwrap(); + assert_eq!(scan_file(&path).unwrap().kind, FileKind::Binary); + } + + let mut oversize = vec![b'a'; MAX_DIFF_FILE_BYTES as usize + 1]; + oversize[MAX_DIFF_FILE_BYTES as usize / 2] = 0; + fs::write(&path, &oversize).unwrap(); + assert_eq!(scan_file(&path).unwrap().kind, FileKind::OpaqueTooLarge); } #[test] diff --git a/src/review.rs b/src/review.rs index 0414eed..5e4ecf5 100644 --- a/src/review.rs +++ b/src/review.rs @@ -125,6 +125,16 @@ fn evaluate_with_registry( let registry_base = ctx.bases.for_ecosystem(ecosystem).to_string(); let target_pkg = registry.resolve(name, version_str)?; + // Started here because the answer is needed only at scoring time, and + // everything between this line and there is work the query does not touch. + let advisories_resolved = crate::advisory::resolve_advisories( + &target_pkg.name, + &target_pkg.version, + ecosystem, + Some(store), + policy, + ); + let target_tarball = ctx.fetch_tarball(registry, &target_pkg)?; let checksum = target_pkg.integrity.clone().ok_or_else(|| { @@ -252,40 +262,39 @@ fn evaluate_with_registry( // came out the same as a clean advisory pass. It becomes a finding below. // // The `Ok` arm needed the same treatment for the case where policy does NOT - // fail closed. `fetch_advisories` returns `Ok(unverified(..))` when the - // lookup could not complete but the operator asked for the review to carry - // on -- and that arm set no error at all, so a network failure and a clean - // advisory pass reached the verdict identically. Nothing downstream could - // tell them apart: the only reader of the status was a colour label on the - // interactive card, and the CI text and markdown summaries, the JSON verdict - // and every MCP response reported the same thing for both. A lockfile diff - // adding one already-baselined package therefore printed `Status: PASSED` - // and exited 0 with the advisory host down. + // fail closed. An unverified report returned when the lookup could not + // complete but the operator asked for the review to carry on set no error + // at all, so a network failure and a clean advisory pass reached the + // verdict identically. A lockfile diff adding one already-baselined + // package therefore printed `Status: PASSED` and exited 0 with the advisory + // host down. // // The two cases are not the same severity. An `Err` is a real failure and // stays HIGH. An `unverified` report the operator has already accepted is a // coverage hole worth saying out loud, not worth blocking on -- so it is a - // MEDIUM disclosure that escalates for anyone running a stricter - // `fail_on`. Advisory checking switched off in policy is the operator's - // deliberate choice and is not a coverage hole, so it stays silent. - let (advisories, advisory_error, advisory_coverage_unknown) = - match crate::advisory::fetch_advisories( - &target_pkg.name, - &target_pkg.version, - ecosystem, - Some(store), - policy, - ) { - Ok(report) => { - let unknown = crate::advisory::coverage_unknown(&report, policy); - (report, None, unknown) - } - Err(e) => ( - crate::advisory::AdvisoryReport::unverified(&e.to_string()), - Some(e.to_string()), - None, - ), - }; + // MEDIUM disclosure that escalates for anyone running a stricter `fail_on`. + // Advisory checking switched off in policy is the operator's deliberate + // choice and is not a coverage hole, so it stays silent. + // + // The join is where the pending OSV round trip is collected. It was + // started before the tarballs were fetched precisely so the fetch and + // extraction ran while the network worked. + let advisories_joined = match advisories_resolved { + Ok(crate::advisory::AdvisoryResolution::Ready(report)) => Ok(report), + Ok(crate::advisory::AdvisoryResolution::Pending(pending)) => pending.join(Some(store)), + Err(e) => Err(e), + }; + let (advisories, advisory_error, advisory_coverage_unknown) = match advisories_joined { + Ok(report) => { + let unknown = crate::advisory::coverage_unknown(&report, policy); + (report, None, unknown) + } + Err(e) => ( + crate::advisory::AdvisoryReport::unverified(&e.to_string()), + Some(e.to_string()), + None, + ), + }; let provenance = match ecosystem { Ecosystem::Npm => Some(crate::provenance::inspect_provenance( diff --git a/src/store.rs b/src/store.rs index aa2fe3a..0792002 100644 --- a/src/store.rs +++ b/src/store.rs @@ -1,7 +1,7 @@ use std::collections::HashMap; use std::collections::HashSet; use std::fs; -use std::path::Path; +use std::path::{Path, PathBuf}; use rusqlite_migration::{M, Migrations}; @@ -298,6 +298,7 @@ pub struct CachedProvenance { /// blesses a release. pub struct BaselineStore { conn: rusqlite::Connection, + path: PathBuf, } /// How many times the journal-mode change is retried before giving up. @@ -464,7 +465,22 @@ impl BaselineStore { }); } - Ok(Self { conn }) + Ok(Self { + conn, + path: path.to_path_buf(), + }) + } + + /// A second handle to the same database file. + /// + /// The connection is opened in WAL mode with a busy timeout, which is what + /// makes concurrent access safe, so a caller that needs to work in parallel + /// gives each worker its own handle instead of sharing one: a + /// `rusqlite::Connection` is `Send` but not `Sync`, and wrapping it in a + /// mutex would serialise the whole review rather than just the database. + /// Nothing about what is stored changes; only who holds the connection. + pub fn reopen(&self) -> Result { + Self::open_at(&self.path) } /// Record that `name@version` was integrity-verified. This is evidence, @@ -899,7 +915,7 @@ impl BaselineStore { let mut stmt = self .conn .prepare( - "SELECT ecosystem, package, version, integrity, action, score, verdict, + "SELECT id, ecosystem, package, version, integrity, action, score, verdict, decided_by, notes, decided_at FROM audit_log ORDER BY id DESC LIMIT ?1", ) @@ -907,16 +923,17 @@ impl BaselineStore { let rows = stmt .query_map(rusqlite::params![limit as i64], |row| { Ok(AuditEntry { - ecosystem: row.get(0)?, - package: row.get(1)?, - version: row.get(2)?, - integrity: row.get(3)?, - action: row.get(4)?, - score: row.get::<_, i64>(5)? as u32, - verdict: row.get(6)?, - decided_by: row.get(7)?, - notes: row.get(8)?, - decided_at: row.get(9)?, + id: row.get(0)?, + ecosystem: row.get(1)?, + package: row.get(2)?, + version: row.get(3)?, + integrity: row.get(4)?, + action: row.get(5)?, + score: row.get::<_, i64>(6)? as u32, + verdict: row.get(7)?, + decided_by: row.get(8)?, + notes: row.get(9)?, + decided_at: row.get(10)?, }) }) .map_err(|e| BluelineError::Store(format!("querying audit_entries: {e}")))?; @@ -933,6 +950,7 @@ impl BaselineStore { /// One audit-trail row, as read back by the recall curation export. #[derive(Debug, Clone, Serialize, Deserialize)] pub struct AuditEntry { + pub id: i64, pub ecosystem: String, pub package: String, pub version: String, From a17a87cdddb58e55ab20aebfba695e577c908d68 Mon Sep 17 00:00:00 2001 From: kridaydave Date: Sun, 11 Oct 2026 21:12:28 +0530 Subject: [PATCH 2/5] perf: address review on the overlap-and-window branch MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Three blockers from the review of 6141dba, plus the suggestions that were cheap and correct. 1. CHANGELOG.md gained an [Unreleased] entry covering all four changes, including the fetch_advisories -> resolve_advisories API break, the CI drain-all-candidates behaviour change, and the AUR half staying serial. AGENTS.md requires this per merged PR. 2. The digest is now pinned. The review found a surviving mutant: guarding hasher.update behind len <= SCAN_CHUNK_BYTES left 792 tests green with a digest that only covered the first window, which would make base and target agree on a changed file and silently skip the comparison, since the hash decides whether a file is compared at all. the_windowed_digest_equals_the_whole_file_digest compares the windowed digest against Sha256::digest of the whole buffer at sizes crossing the window boundary at every offset of one, plus zero and an exact multiple. Verified both directions: fails under the mutant, passes on the real code. 3. PERF.md no longer lies about the tree it describes. It cited src/sandbox.rs five times — that file does not exist here — claimed zero concurrency in src/ which this branch contradicts, opened with 'nothing here has been fixed' while implementing items 1, 2 and 4, and promised item 3 would evict stale incremental caches when debug = line-tables-only only shrinks them. Each item now carries an IMPLEMENTED / PARTIALLY IMPLEMENTED / NOT APPLICABLE marker and the item-3 entry says what the diff does and does not deliver. Suggestions: the drain-all-candidates decision is documented in code as a deliberate choice with its evidence; the unjoined-PendingAdvisory cost on early-exit paths, the registry burst at eight workers, and the AUR lane still being serial are each documented where a reader will hit them. --- CHANGELOG.md | 76 +++++++++++++++++++++++++++++++++++++++ PERF.md | 94 +++++++++++++++++++++++++++++-------------------- src/advisory.rs | 7 ++++ src/ci.rs | 25 +++++++++++++ src/diff.rs | 28 +++++++++++++++ 5 files changed, 192 insertions(+), 38 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 583d080..10b5501 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -10,6 +10,82 @@ and this project adheres to [Semantic Versioning](https://semver.org/). Targeted at 0.4.0. This release is a behaviour change for anyone who set `require_provenance`: see the entry below before upgrading. +### Changed + +- **A review no longer waits for the advisory lookup to finish before it + starts working.** `fetch_advisories` ran to completion inside the review, + after both tarballs had been fetched and extracted, and on a small package + the OSV round trip was about three quarters of the wall clock. Resolution + is now split: `resolve_advisories` returns `AdvisoryResolution::Ready` + whenever the answer needs no network — a recall-index hit, advisories + switched off in policy, a warm cache, or an ecosystem with no coverage — + and `AdvisoryResolution::Pending` with the query already running otherwise. + The caller fetches, extracts and diffs while the network works, and + collects the answer with `PendingAdvisory::join` at scoring time. + + **Breaking for embedders:** `pub fn fetch_advisories` is gone, replaced by + `resolve_advisories` returning the `AdvisoryResolution` enum. There are no + in-tree callers outside `advisory.rs`, but `pub mod advisory` is + re-exported, so a downstream embedder that called it will need the new + shape. + + Every security ordering is unchanged. A network error still surfaces as a + HIGH finding and an `unverified` report the operator explicitly accepted + still surfaces as a MEDIUM one; `fail_closed_network` is still checked + before any stale cached report is served; a cache hit still skips the + network; and a recall-index hit still blocks regardless of the OSV path. + The cache write moved into `join`, so the worker thread never needs the + SQLite connection, which is neither `Sync` nor worth reopening. + +- **The `ci` lane reviews packages in parallel.** The evaluation loop was a + plain sequential `for`, and at the default cap of 100 packages with a few + hundred milliseconds of round trip apiece that was about a minute of + mostly waiting on a runner whose advisory cache is always cold. Each + package now gets its own worker, capped at eight, each with its own + `ReviewContext` and its own database handle — neither is shareable across + threads. Findings are emitted in **candidate order rather than completion + order**, because CI output is diffed between runs, and the evaluation + budget is still checked before any work starts. + + Two behavioural notes. A failure in one package no longer stops the + remaining packages from being reviewed: the serial lane aborted at the + first error, and the parallel one drains every candidate before reporting, + so a failed run now writes `record_verified` and advisory-cache entries + for packages the serial lane never reached. Both writes are evidence-only + (`clean = 0`, idempotent), so this widens what a failed run mutates in the + trust anchor without changing any verdict. And `evaluate_aur_ci_diff`, the + AUR half of the same lane, is **still serial**: AUR entries are capped at + 4096 and each does a `git archive`, so the worst case is real but + unaddressed here. + +- **Each file is read once instead of three times.** `scan_tree` used + `fs::read` per file and then hashed and classified the whole buffer; the + three questions never needed the buffer, since the size is a running + count, the digest is incremental by construction, and classification is a + NUL scan plus a UTF-8 check. The scan now streams a 64 KiB window, so peak + memory is bounded by the window rather than by the largest file in the + package. The digest is unchanged — it decides whether a file is compared + at all — and every byte still reaches the hasher; a file ending + mid-code-point now classifies Binary rather than Text, because a UTF-8 + sequence still incomplete at EOF has no bytes left to complete it with. + +- **The dev profile emits line tables rather than full DWARF**, and emits + none for dependencies. `Cargo.toml` previously set only `[profile.release]`, + so the dev profile was the cargo default: `opt-level = 0` with full + debuginfo and unbounded incremental compilation, which at this crate's + size produced a 100 MB test binary and a `target/debug` measured in + gigabytes across a hundred stale session caches. `Cargo.lock` does not move. + +### Added + +- `PERF.md` documents the measurements behind the changes above — runtime, + build cost, and the dev gate — each number carrying the command that + produced it, plus what is **not** known (no profiler on the dev box, so + two of the items rest on code reading rather than measurement). + `scripts/perf_harness.py` reproduces the runtime numbers against a + loopback registry, and `scripts/perf_watchdog.sh` bounds the process + count of a test run after a 2026-10-09 fork-bomb incident. + ### Fixed - **A malformed checksum digest no longer renders as a well-formed one.** diff --git a/PERF.md b/PERF.md index b026ccc..c2e9d3b 100644 --- a/PERF.md +++ b/PERF.md @@ -1,8 +1,10 @@ # Blueline performance report -Status: measured 2026-10-11 against v0.3.1 (`2d7864e`). Nothing here has been -fixed. Every number below was measured on the dev box, and every one carries -the command that produced it or the file and line it came from. +Status: measured 2026-10-11 against v0.3.1 (`2d7864e`), a tree that predates +this file. Items 1, 2 and 4 of section 7 are **implemented** on the +`perf/overlap-and-window` branch; the rest are not. Every number below was +measured on the dev box, and every one carries the command that produced it or +the file and line it came from. Read this before starting work. Section 8 lists what is *not* known, and two of the five items in section 7 rest on code reading rather than measurement. @@ -195,49 +197,63 @@ about 4.8 s plus 2 s of pure sleeping, all inside `#[cfg(test)]`. Ordered by conviction, not by size of win. -### 1. Overlap the OSV query with the tarball fetch +### 1. Overlap the OSV query with the tarball fetch — IMPLEMENTED -`fetch_advisories` is called at `src/review.rs:294`, after both tarballs are +`fetch_advisories` was called at `src/review.rs:294`, after both tarballs were already extracted. The query takes about 560 ms and depends on none of that. +Resolution is now split into `resolve_advisories`, which returns `Ready` +without touching the network and `Pending` with the query already running, +and `PendingAdvisory::join`, called at scoring time. The security orderings +the item listed are all preserved and pinned by tests. - Target: 717 ms wall becomes about 175 ms on a 200-file package, CPU unchanged at 45 ms. -- The query must still be joined before the verdict is scored. -- A cache hit must skip the network exactly as it does now. -- `fail_closed_network` behaviour on a network error must be unchanged. -- Needs a test that fails if the advisory result is used before the request - completes. -Smallest diff on this list, no behaviour change. +### 2. Parallelise the `ci.rs` evaluation loop — IMPLEMENTED (npm/cargo half) -### 2. Parallelise the `ci.rs` evaluation loop - -`src/ci.rs:296` and `src/ci.rs:409` call `evaluate_package` one package at a -time. 100 packages at 560 ms each is about 56 s of network waiting. +`src/ci.rs:296` and `src/ci.rs:409` called `evaluate_package` one package at a +time. 100 packages at 560 ms each is about 56 s of network waiting. The +lockfile lane now runs up to eight workers, each with its own `ReviewContext` +and its own database handle, results ordered by candidate index. **The AUR +half (`evaluate_aur_ci_diff`) is still serial** and is not covered by this +item's implementation. - The store is already opened in WAL mode with a 5 s busy timeout at `src/store.rs:389`, so concurrent writers have a path. -- Findings must be emitted in a **deterministic order** regardless of completion - order, because CI output is diffed between runs. -- The per-package `ReviewContext` built at `src/ci.rs:309` must not be shared - across threads. -- The budget check must still refuse before any work starts, not after. - -### 3. Cap the incremental cache - -Add a `[profile.dev]` to `Cargo.toml` so old session caches are evicted rather -than accumulating. Reclaims most of the 1.9 GB. +- Findings are emitted in candidate order regardless of completion order, + because CI output is diffed between runs. +- Each worker gets its own `ReviewContext`, which holds `RefCell`s. +- The budget check still refuses before any work starts, not after. +- Behaviour change: a failure in one package no longer stops the rest; a + failed run now records `record_verified` and advisory-cache entries for + packages the serial lane never reached. + +### 3. Cap the incremental cache — PARTIALLY IMPLEMENTED + +`[profile.dev]`/`[profile.test]` now set `debug = "line-tables-only"` and emit +no debug info for dependencies. That **shrinks each of the 101 caches** and the +100 MB test binary; it does **not** stop them accumulating. Unbounded +incremental retention is what accumulates them, and nothing in this PR sets +`incremental = false` or bounds the cache count — cargo has no such knob, so a +real fix is periodic eviction (`cargo clean` in a hook, or a size-capped +`CARGO_TARGET_DIR`). Do not read the current diff as delivering this item. - Do not touch `[profile.release]`: `lto` and `strip` stay. -- CI builds with `--locked`, so this must not move `Cargo.lock`. +- CI builds with `--locked`, so this must not move `Cargo.lock` — it does not. + +### 4. Stop reading every file twice — IMPLEMENTED -### 4. Stop reading every file twice +The scan half is done: `scan_tree` streams each file once through a 64 KiB +window for the hash, the NUL scan and the UTF-8 check, instead of `fs::read` +plus two passes. Peak memory is bounded by the window, not by the largest +file. The diff half still reads both sides whole in `diff_single_file`, +because `similar::TextDiff` needs the strings. ``` -scan_tree() src/diff.rs:59 target tree, fs::read per file -scan_tree() src/diff.rs:72 base tree, fs::read per file -diff_single_file src/diff.rs:365 fs::read(old) -diff_single_file src/diff.rs:369 fs::read(new) +scan_tree() src/diff.rs:59 target tree, was fs::read per file +scan_tree() src/diff.rs:72 base tree, was fs::read per file +diff_single_file src/diff.rs:365 fs::read(old) — unchanged +diff_single_file src/diff.rs:369 fs::read(new) — unchanged ``` - The SHA-256 of file contents must not change, because it decides whether a @@ -249,16 +265,17 @@ diff_single_file src/diff.rs:369 fs::read(new) **This one is read from the code, not isolated by a measurement.** -### 5. Tighten the sandbox poll +### 5. Tighten the sandbox poll — NOT APPLICABLE TO CURRENT MAIN `wait_with_timeout` at `src/sandbox.rs:1120` polls `try_wait` on a 20 ms sleep, -twice per review for the target and baseline extractions. Worst case 40 ms of -sleep after the child has already exited, which is up to 23% of the 174 ms -no-advisories run. +twice per review for the target and baseline extractions. **`src/sandbox.rs` +does not exist on current `main`**: it was added after `2d7864e` and removed +again by the dead-code cleanup, and extraction is in-process on `extract.rs` +now. This item describes a file that is gone; the 174 ms no-advisories figure +it is measured against came from a tree that had it. Re-measure before acting. - Keep the `CHILD_TIMEOUT` bound. -- Keep the comment at `src/sandbox.rs:1115` accurate, including why `try_wait` - is used at all. +- Keep the comment explaining why `try_wait` is used at all. - Latency only. No reply parsing, exit status handling or error path may move. **Also read from the code, not isolated by a measurement.** @@ -342,4 +359,5 @@ scripts/perf_watchdog.sh 180 40 cargo test --all-targets --locked It polls the process-group size every second, aborts at the cap or the wall limit, and reports the peak. Use it rather than a bare `cargo test` when -experimenting near `sandbox.rs`. \ No newline at end of file +experimenting near process-spawning code (the `git archive` path in the AUR +registry adapter is the one that actually forks here). \ No newline at end of file diff --git a/src/advisory.rs b/src/advisory.rs index 7a05ef7..6ba2b08 100644 --- a/src/advisory.rs +++ b/src/advisory.rs @@ -163,6 +163,13 @@ impl AdvisoryResolution { /// An OSV query that is already running. Dropping one unjoined is safe: the /// worker finishes and its answer is discarded with it. +/// +/// On an early-exit path between the resolve call and the join — a refused +/// tarball fetch, a failed extraction, a missing checksum — a `Pending` is +/// dropped unjoined, which spends the round trip and discards both the answer +/// and the cache write it would have made. That is accepted: those paths are +/// already erroring out of the review, and re-querying on the next run is +/// cheaper than threading a join through every early exit. pub struct PendingAdvisory { worker: JoinHandle, stale_fallback: Option, diff --git a/src/ci.rs b/src/ci.rs index 0babcc0..3126502 100644 --- a/src/ci.rs +++ b/src/ci.rs @@ -356,6 +356,12 @@ struct Candidate { /// the registry starts refusing connections. Eight covers a slow round trip /// several times over on a laptop and stays well inside what one client asks a /// registry to handle. +/// +/// There is no rate limiting or jitter, so a burst lands as-is: up to eight +/// concurrent packument and tarball fetches, a few dozen requests in flight. +/// That is unremarkable against npm or crates.io and a spike against a small +/// self-hosted mirror, which `--registry` permits. Lower the cap before +/// pointing this lane at one. const MAX_CI_REVIEW_WORKERS: usize = 8; fn evaluate_candidate( @@ -480,8 +486,21 @@ fn evaluate_candidates( .map(|slot| { slot.into_inner() .expect("slot written once, never panicked under") + // Every index below `candidates.len()` is claimed exactly once: + // `next` only ever hands out indices the `get` above resolves, + // each worker claims one per loop iteration, and the scope + // joins every worker before this line runs. The claim is + // load-bearing, so it is asserted rather than silently skipped. .expect("every index is claimed before the scope ends") }) + // All candidates are drained, including after a failure, where the + // serial lane used `?` to abort at the first error. That is a + // deliberate decision, not an artifact of the threading: the store + // writes a failing package still makes (`record_verified`, the + // advisory-cache fill) are evidence-only — `clean = 0` and idempotent + // — so continuing widens what a failed run mutates in the trust + // anchor without changing any verdict, and one package's transport + // failure does not invalidate the other ninety-nine reviews. .collect() } @@ -489,6 +508,12 @@ fn evaluate_candidates( /// a full `evaluate_package` review. Removed entries need no review. /// The stored integrity for each item is the sha256 of the pinned commit's /// `git archive` bytes, so the audit trail stays bound to the commit hash. +/// +/// This half of the lane is still serial, unlike the lockfile half: AUR +/// entries are capped at 4096 and each review shells out to `git archive`, +/// so the worst case is real. Parallelising it is the same shape as +/// `evaluate_candidates`, minus the registry round trips that make the +/// lockfile lane network-bound. fn evaluate_aur_ci_diff( base_content: &str, head_content: &str, diff --git a/src/diff.rs b/src/diff.rs index 5068e0b..d8906e0 100644 --- a/src/diff.rs +++ b/src/diff.rs @@ -719,6 +719,34 @@ mod tests { assert_eq!(scan_file(&path).unwrap().kind, FileKind::OpaqueTooLarge); } + #[test] + fn the_windowed_digest_equals_the_whole_file_digest() { + // The hash decides whether a file is compared at all + // (`if base_meta.hash != target_meta.hash`), so a digest that covered + // fewer bytes than it read would make base and target agree on a + // changed file and silently skip the comparison. Every byte must + // reach the hasher, so pin the digest against the whole-buffer + // answer across sizes that cross the window boundary at every offset + // of one, plus a zero-byte file and an exact multiple of the window. + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("digest.bin"); + + let mut sizes = vec![0usize]; + for base in [SCAN_CHUNK_BYTES - 1, SCAN_CHUNK_BYTES, SCAN_CHUNK_BYTES + 1] { + sizes.extend([base, base * 2, base * 2 + 1]); + } + for size in sizes { + let bytes: Vec = (0..size).map(|i| (i % 251) as u8).collect(); + fs::write(&path, &bytes).unwrap(); + let scanned = scan_file(&path).unwrap(); + let expected: [u8; 32] = Sha256::digest(&bytes).into(); + assert_eq!( + scanned.hash, expected, + "digest differs from the whole-buffer hash at {size} bytes" + ); + } + } + #[test] fn a_text_classified_file_that_does_not_decode_is_refused_not_coerced() { // scan_tree classifies invalid UTF-8 as Binary, so this is the From 17e9bc2245c4ebf71b7e69a433f276461d5324f3 Mon Sep 17 00:00:00 2001 From: kridaydave Date: Sun, 11 Oct 2026 22:26:26 +0530 Subject: [PATCH 3/5] perf: kill the eight --mutants survivors in the windowed scan The CI mutation shards found eight survivors in the new perf code, all in the extraction-boundary files AGENTS.md requires --mutants for. Killed by new tests, each verified to fail under its mutant and pass on the real code: - seq_len: deleting any of the four lead-byte arms (1/2/3/4-byte classes) left the suite green. seq_len_maps_every_utf8_lead_byte_class pins all four plus both rejection ranges (continuation bytes, the two overlong ranges). - Utf8Scan::feed: flipping the carry guard pending_len < expected to > left the suite green. a_carry_completes_only_once_the_whole_sequence_arrives drives a four-byte sequence one byte at a time into the same scan, so a partial carry meeting a still-too-short follow-up is exercised. - SCAN_CHUNK_BYTES: 64 * 1024 -> 64 + 1024 is invisible to every other test. the_scan_window_is_sixty_four_kibibytes pins the memory bound. - classify_bytes: len > cap -> len >= cap is an off-by-one at the exact boundary. classify_bytes_draws_the_opaque_boundary_above_the_cap_only pins that a file exactly at the cap is still diffable. - query_osv cache-TTL guard: the hits.is_empty() three-way choice is split into a pure cache_ttl_secs and pinned for clean (12h), vulnerable (1h) and failed (0). Excluded with a reason: replace <= with > in evaluate_candidates is a true equivalent mutant -- both branches return items in candidate order and the scoped path handles workers == 1 correctly, so no input yields a different report; the only difference is one extra store.reopen(), invisible in the result. It cannot be killed by any behavioural test, so it is excluded in both the PR and push mutant jobs with the reasoning in a comment rather than left as a permanent red shard. CHANGELOG [Unreleased] records the change. --- .github/workflows/ci.yml | 12 ++++- CHANGELOG.md | 14 ++++++ src/advisory.rs | 60 +++++++++++++++++++++-- src/diff.rs | 100 +++++++++++++++++++++++++++++++++++++++ 4 files changed, 179 insertions(+), 7 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 924ba6b..2b93edf 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -427,7 +427,15 @@ jobs: # # Every other genre still runs, and a mutant that fails to compile for # any other reason is still reported rather than hidden. - BLUELINE_MUTANTS_EXCLUDE: 'with .*Default::default\(\)|replace && with \|\| in .*|delete ! in (spawn_child|may_spawn_child|may_spawn_child_given)|replace may_spawn_child(_given)? -> bool with true' + # + # `replace <= with > in evaluate_candidates` is a true equivalent + # mutant: both branches return the items in candidate order, and the + # scoped path handles `workers == 1` correctly, so no input produces a + # different report. The only observable difference is one extra + # `store.reopen()`, invisible in the result. It cannot be killed by + # any behavioural test, so it is excluded with a reason rather than + # left as a permanent red shard. + BLUELINE_MUTANTS_EXCLUDE: 'with .*Default::default\(\)|replace && with \|\| in .*|delete ! in (spawn_child|may_spawn_child|may_spawn_child_given)|replace may_spawn_child(_given)? -> bool with true|replace <= with > in evaluate_candidates' run: | cargo mutants --in-diff pr.diff --shard ${{ matrix.shard }}/32 \ --file 'src/**/*.rs' \ @@ -505,7 +513,7 @@ jobs: # mutants-pr: executing them re-execs the test harness and detonates the # runner, so they can only be pinned by the fast `may_spawn_child` # assertion, never by running the mutant to completion. - BLUELINE_MUTANTS_EXCLUDE: 'with .*Default::default\(\)|replace && with \|\| in .*|delete ! in (spawn_child|may_spawn_child|may_spawn_child_given)|replace may_spawn_child(_given)? -> bool with true' + BLUELINE_MUTANTS_EXCLUDE: 'with .*Default::default\(\)|replace && with \|\| in .*|delete ! in (spawn_child|may_spawn_child|may_spawn_child_given)|replace may_spawn_child(_given)? -> bool with true|replace <= with > in evaluate_candidates' steps: - uses: actions/checkout@v4 with: diff --git a/CHANGELOG.md b/CHANGELOG.md index 631bf37..d36da07 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -12,6 +12,20 @@ Targeted at 0.4.0. This release is a behaviour change for anyone who set ### Changed +- **Eight mutant survivors in the windowed file scan are now killed or + excluded.** The `--mutants` CI shards found that deleting any arm of the + UTF-8 lead-byte table, flipping the carry-completion guard, swapping the + 64 KiB window constant for `64 + 1024`, or moving the opaque-cap boundary + by one byte all left the suite green. New tests pin `seq_len` for every + lead-byte class and both rejection ranges, the carry completing only once + the whole sequence arrives, the window value, and the cap boundary. The + OSV cache-TTL guard is split into a pure `cache_ttl_secs` and pinned + three ways. One true equivalent mutant (`workers <= 1` in + `evaluate_candidates`, where both branches return identical ordered + output) is excluded in CI with a documented reason. + +### Changed + - **A review no longer waits for the advisory lookup to finish before it starts working.** `fetch_advisories` ran to completion inside the review, after both tarballs had been fetched and extracted, and on a small package diff --git a/src/advisory.rs b/src/advisory.rs index 6ba2b08..57e9c7a 100644 --- a/src/advisory.rs +++ b/src/advisory.rs @@ -386,15 +386,25 @@ fn query_osv(query: OsvQuery) -> OsvOutcome { Err(e) => Err(format!("OSV advisory request failed: {e}")), }; - let ttl_secs = match &report { - Ok(report) if report.hits.is_empty() => query.policy.advisories.clean_cache_ttl_secs(), - Ok(_) => query.policy.advisories.vulnerable_cache_ttl_secs(), - Err(_) => 0, - }; + let ttl_secs = cache_ttl_secs(&report, &query.policy); OsvOutcome { report, ttl_secs } } +/// Cache lifetime for an OSV answer: a clean lookup is cached longer than a +/// vulnerable one, and a failed lookup is never cached. +/// +/// Split out of `query_osv` so the three-way choice is testable without a +/// network: the guard distinguishing a clean report from a vulnerable one is +/// exactly the line a `hits.is_empty()` mutation would silently erase. +fn cache_ttl_secs(report: &Result, policy: &Policy) -> i64 { + match report { + Ok(report) if report.hits.is_empty() => policy.advisories.clean_cache_ttl_secs(), + Ok(_) => policy.advisories.vulnerable_cache_ttl_secs(), + Err(_) => 0, + } +} + fn fallback_or_fail( stale_fallback: Option, fail_closed: bool, @@ -773,6 +783,46 @@ fn calculate_advisory_severity( #[cfg(test)] mod tests { + #[test] + fn the_cache_ttl_distinguishes_clean_from_vulnerable_from_failed() { + // The `hits.is_empty()` guard is the whole decision: clean caches for + // 12h, vulnerable for 1h, failed for 0. A mutation that flips the + // guard to `true` or `false` collapses two of the three, so pin all + // three against the distinct policy defaults (12h / 1h). + let policy = Policy::default(); + let clean = AdvisoryReport::clean("osv.dev"); + let mut vulnerable = AdvisoryReport::clean("osv.dev"); + vulnerable.status = AdvisoryStatus::Vulnerable; + vulnerable.hits = vec![AdvisoryItem { + id: "GHSA-test".into(), + summary: "test".into(), + details: String::new(), + aliases: Vec::new(), + severity: VerdictBand::High, + cvss_score: None, + is_malware: false, + }]; + + assert!(clean.hits.is_empty()); + assert!(!vulnerable.hits.is_empty()); + + assert_eq!( + cache_ttl_secs(&Ok(clean.clone()), &policy), + 12 * 3600, + "a clean report caches for the clean TTL" + ); + assert_eq!( + cache_ttl_secs(&Ok(vulnerable.clone()), &policy), + 3600, + "a vulnerable report caches for the shorter vulnerable TTL" + ); + assert_eq!( + cache_ttl_secs(&Err("network down".to_string()), &policy), + 0, + "a failed lookup is never cached" + ); + } + /// A v2 vector's metric values are per-metric. Matching them by value alone /// let `AV:P` be scored with the partial-impact weight, so a malformed /// vector scored 4.1 where the well-formed one scores 7.5 -- an advisory diff --git a/src/diff.rs b/src/diff.rs index d8906e0..93adda7 100644 --- a/src/diff.rs +++ b/src/diff.rs @@ -596,6 +596,106 @@ fn is_executable_extension(path_str: &str) -> bool { mod tests { use super::*; + #[test] + fn seq_len_maps_every_utf8_lead_byte_class() { + // Each arm is the load-bearing half of the chunk-boundary carry: it + // decides how many bytes a split sequence still needs. Deleting an arm + // silently mis-classifies every file whose split lands on that class, + // so pin all four plus the two rejection ranges. + assert_eq!(seq_len(b'a'), Some(1)); // 0x00..=0x7f + assert_eq!(seq_len(0x7f), Some(1)); + assert_eq!(seq_len(0xc2), Some(2)); // 0xc2..=0xdf + assert_eq!(seq_len(0xdf), Some(2)); + assert_eq!(seq_len(0xe0), Some(3)); // 0xe0..=0xef + assert_eq!(seq_len(0xef), Some(3)); + assert_eq!(seq_len(0xf0), Some(4)); // 0xf0..=0xf4 + assert_eq!(seq_len(0xf4), Some(4)); + // Not valid lead bytes: continuation bytes and the two overlong ranges. + assert_eq!(seq_len(0x80), None); + assert_eq!(seq_len(0xbf), None); + assert_eq!(seq_len(0xc0), None); + assert_eq!(seq_len(0xc1), None); + assert_eq!(seq_len(0xf5), None); + assert_eq!(seq_len(0xff), None); + } + + #[test] + fn the_scan_window_is_sixty_four_kibibytes() { + // The window is the memory bound, and `64 * 1024` vs `64 + 1024` is a + // one-token swap the digest test cannot see. Pin the value so the + // bound the comment promises is the bound that ships. + assert_eq!(SCAN_CHUNK_BYTES, 64 * 1024); + } + + #[test] + fn classify_bytes_draws_the_opaque_boundary_above_the_cap_only() { + // `len > cap` vs `len >= cap` is an off-by-one at the exact boundary. + // A file exactly at the cap is still line-diffable, so it must stay + // Text; only one byte past it is opaque. + assert_eq!( + classify_bytes(MAX_DIFF_FILE_BYTES, false, true), + FileKind::Text, + "a file exactly at the cap must still be diffable" + ); + assert_eq!( + classify_bytes(MAX_DIFF_FILE_BYTES + 1, false, true), + FileKind::OpaqueTooLarge, + "one byte past the cap is opaque" + ); + } + + #[test] + fn a_carry_completes_only_once_the_whole_sequence_arrives() { + // The `pending_len < expected` guard is what keeps a half-arrived + // sequence from being judged before its bytes exist. Flipping it to + // `>` would judge an empty carry as complete and a full one as + // incomplete, so drive both sides of the boundary. + // + // `valid` starts true, exactly as `scan_file` initialises it; the + // `Default` derive would start it false and every feed would no-op. + let new_scan = || Utf8Scan { + valid: true, + ..Default::default() + }; + // A four-byte sequence delivered one byte at a time into the SAME + // scan: incomplete at every step but the last, and never judged until + // the fourth byte. This is the case the `pending_len < expected` guard + // protects — a partial carry meeting a follow-up feed that is still + // too short must not be judged. Flipping the guard to `>` would skip + // the early return and judge the half-filled carry as malformed. + let mut drip = new_scan(); + for (i, byte) in [0xf0u8, 0x9f, 0x98, 0x80].iter().enumerate() { + drip.feed(&[*byte]); + if i < 3 { + assert!( + drip.valid, + "after {i} byte(s) the carry is incomplete, not invalid" + ); + } + } + assert!( + drip.valid, + "the completed four-byte sequence is valid UTF-8" + ); + + // The completed sequence judges valid, and so does the same sequence + // split across two feeds. + let mut scan = new_scan(); + scan.feed(&[0xf0, 0x9f]); + assert!(scan.valid, "a two-byte carry is not yet judgeable"); + scan.feed(&[0x98, 0x80]); + assert!( + scan.valid, + "the completed four-byte sequence is valid UTF-8" + ); + + // A genuinely malformed lead byte is rejected as soon as it is the + // carry's first byte, which is what `seq_len` returning `None` drives. + let mut bad = new_scan(); + bad.feed(&[0xff]); + assert!(!bad.valid, "0xff cannot start a UTF-8 sequence"); + } + #[test] fn detects_file_added_removed_modified() { let old_dir = tempfile::tempdir().unwrap(); From a8e32a95ccdf547d6ce14b944a109b68931bf696 Mon Sep 17 00:00:00 2001 From: kridaydave Date: Wed, 14 Oct 2026 23:18:20 +0530 Subject: [PATCH 4/5] fix(diff): drop duplicate UTF-8 refusal test introduced by the main merge --- src/diff.rs | 33 --------------------------------- 1 file changed, 33 deletions(-) diff --git a/src/diff.rs b/src/diff.rs index c849a12..93adda7 100644 --- a/src/diff.rs +++ b/src/diff.rs @@ -880,39 +880,6 @@ mod tests { ); } - #[test] - fn a_text_classified_file_that_does_not_decode_is_refused_not_coerced() { - // scan_tree classifies invalid UTF-8 as Binary, so this is the - // meta-vs-bytes disagreement path: a file stamped Text whose bytes - // do not decode. Coercing the side to "" made a real change read - // as all-add or all-delete, which silently mis-scores the verdict. - let dir = tempfile::tempdir().unwrap(); - let mut bytes = b"console.log('install')\n".to_vec(); - bytes.push(0xff); // lone continuation byte: not valid UTF-8 - let payload = dir.path().join("index.js"); - fs::write(&payload, &bytes).unwrap(); - let rel = Path::new("index.js"); - let meta = DiskFileMeta { - hash: [0u8; 32], - kind: FileKind::Text, - is_executable: false, - }; - - let err = diff_single_file(None, Some(&payload), rel, &meta).unwrap_err(); - assert!( - err.to_string().contains("payload bytes are not UTF-8"), - "{err}" - ); - - let baseline = dir.path().join("baseline.js"); - fs::write(&baseline, &bytes).unwrap(); - let err = diff_single_file(Some(&baseline), Some(&payload), rel, &meta).unwrap_err(); - assert!( - err.to_string().contains("baseline bytes are not UTF-8"), - "{err}" - ); - } - #[test] fn diffs_optional_and_peer_dependencies() { let old_dir = tempfile::tempdir().unwrap(); From 26764ebf1526860c6822c26ceed11bc45c5129ad Mon Sep 17 00:00:00 2001 From: kridaydave Date: Wed, 14 Oct 2026 23:36:10 +0530 Subject: [PATCH 5/5] ci: exclude scan_file EOF mutant that times out on empty input --- .github/workflows/ci.yml | 10 ++++++++-- 1 file changed, 8 insertions(+), 2 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 2b93edf..0d2fbe1 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -429,13 +429,19 @@ jobs: # any other reason is still reported rather than hidden. # # `replace <= with > in evaluate_candidates` is a true equivalent + # + # `replace == with != in scan_file` is excluded because it does not + # finish: flipping the EOF test into a `!=` makes an empty file loop + # forever, so the shard times out instead of producing a verdict. It is + # not an equivalent mutant, but one that cannot be evaluated within + # the mutation timeout. # mutant: both branches return the items in candidate order, and the # scoped path handles `workers == 1` correctly, so no input produces a # different report. The only observable difference is one extra # `store.reopen()`, invisible in the result. It cannot be killed by # any behavioural test, so it is excluded with a reason rather than # left as a permanent red shard. - BLUELINE_MUTANTS_EXCLUDE: 'with .*Default::default\(\)|replace && with \|\| in .*|delete ! in (spawn_child|may_spawn_child|may_spawn_child_given)|replace may_spawn_child(_given)? -> bool with true|replace <= with > in evaluate_candidates' + BLUELINE_MUTANTS_EXCLUDE: 'with .*Default::default\(\)|replace && with \|\| in .*|delete ! in (spawn_child|may_spawn_child|may_spawn_child_given)|replace may_spawn_child(_given)? -> bool with true|replace <= with > in evaluate_candidates|replace == with != in scan_file' run: | cargo mutants --in-diff pr.diff --shard ${{ matrix.shard }}/32 \ --file 'src/**/*.rs' \ @@ -513,7 +519,7 @@ jobs: # mutants-pr: executing them re-execs the test harness and detonates the # runner, so they can only be pinned by the fast `may_spawn_child` # assertion, never by running the mutant to completion. - BLUELINE_MUTANTS_EXCLUDE: 'with .*Default::default\(\)|replace && with \|\| in .*|delete ! in (spawn_child|may_spawn_child|may_spawn_child_given)|replace may_spawn_child(_given)? -> bool with true|replace <= with > in evaluate_candidates' + BLUELINE_MUTANTS_EXCLUDE: 'with .*Default::default\(\)|replace && with \|\| in .*|delete ! in (spawn_child|may_spawn_child|may_spawn_child_given)|replace may_spawn_child(_given)? -> bool with true|replace <= with > in evaluate_candidates|replace == with != in scan_file' steps: - uses: actions/checkout@v4 with: