diff --git a/benchmarks/unseen/draft.py b/benchmarks/unseen/draft.py index 7da9c04..a2849bd 100644 --- a/benchmarks/unseen/draft.py +++ b/benchmarks/unseen/draft.py @@ -16,7 +16,8 @@ import random import subprocess import tempfile import time -from collections import Counter, defaultdict +from collections import Counter, defaultdict, deque +from concurrent.futures import ThreadPoolExecutor from contextlib import contextmanager from dataclasses import asdict from pathlib import Path @@ -121,9 +122,21 @@ def write_draft(root: Path, assigned: list[dict], *, seed: int) -> tuple[Path, P class Clones: - """One disposable blobless clone at a time; no name in cache paths or subprocess output.""" + """Disposable blobless clones; each worker receives its own counter accumulator.""" - def __init__(self, root: Path, cache: Path, *, runner=None, limit_bytes=500 * 1024 * 1024, deadline_seconds: float | None = None): + def __init__( + self, + root: Path, + cache: Path, + *, + runner=None, + limit_bytes=500 * 1024 * 1024, + deadline_seconds: float | None = None, + extract_workers: int = 6, + ): + if extract_workers < 1: + raise ValueError("invalid_extract_workers") + self.extract_workers = extract_workers self.cache = external(root, cache) self.runner = runner or subprocess.run self.limit_bytes = limit_bytes @@ -138,8 +151,8 @@ class Clones: path = Path(directory) git = Git(path, runner=self.runner, limit_bytes=self.limit_bytes, deadline_seconds=self.deadline_seconds) try: - # Full commit graph is needed for temporal thirds and OSV brackets; blobs - # are fetched lazily. This is a partial clone, not a truncated history. + # Full commit graph is needed for temporal thirds and OSV brackets. + # repository_changes batches blobs before reading their content. done = self.runner( ["git", "-c", "credential.helper=", "clone", "--quiet", "--filter=blob:none", "--no-checkout", url, str(path)], stdin=subprocess.DEVNULL, @@ -195,18 +208,21 @@ def advisory_index(rows: list[dict], github) -> dict[str, set[str]]: def repository_changes(git, candidate, fixes: set[str], *, seed: int): + # Fetch detached fixes together, then materialize the histories before any diff, + # show or blame can trigger a per-file promisor fetch. + fixes = sorted(fixes | {candidate.fix}) + # The candidate fetch in extract_repository retains the original tag behavior. + # Do not import other fixes' tags before introducing() checks its version bracket. + git.run("fetch", "--quiet", "--no-tags", "--filter=blob:none", "origin", *fixes) + git.materialize(["HEAD", *fixes]) proof = prove_source(git, candidate.parent) change = extract.introducing(git, candidate) known = set() expanded = set() - for fix in sorted(fixes | {candidate.fix}): - # Explicitly fetch detached advisory commits unavailable on default-branch history. - # All git including this possible network call still goes through the injected client. - git.run("fetch", "--quiet", "--filter=blob:none", "origin", fix) + for fix in fixes: full = git.run("rev-parse", fix).strip() expanded.add(full) - parents = git.parents(full) - for parent in parents: + for parent in git.parents(full): sites = extract.fix_sites(git, full, parent, global_ok=True) known.update((site["path"], site["function"]) for site in sites) rows = extract.ordinary_history(git, "HEAD") @@ -255,27 +271,30 @@ def enumerate_advisories(github, journal): return list(records.values()) -def decide(row, github, osv, used, selected_networks): +def decide(row, github, osv, used, selected_networks=()): + # selected_networks is retained for callers; only ordered acceptance consults it. + network = {} try: links = source.fix_links(row) if len(links) != 1: raise source.Rejected("fix_links") name, _ = next(iter(links)) resolved = eligibility.resolve(name, github) - rejected_by = eligibility.rejection(resolved, used, set(selected_networks)) + rejected_by = eligibility.rejection(resolved, used, set()) if rejected_by: raise source.Rejected("used_" + rejected_by) + network = {"network": resolved.network} candidate = source.candidate(row, github, osv) return {"candidate": asdict(candidate), "network": resolved.network} except source.Rejected as exc: - return {"rejected": str(exc)} + return {"rejected": str(exc), **network} except eligibility.RepositoryNotFound: - return {"rejected": "repository_or_commit_not_found"} + return {"rejected": "repository_or_commit_not_found", **network} except ValueError as exc: # A per-candidate transport failure (e.g. get_commit on an unresolvable fix ref) # skips this advisory; the run loop's systemic guard catches a broken instrument. if str(exc) == "http_failure": - return {"rejected": "decide_http_failure"} + return {"rejected": "decide_http_failure", **network} raise @@ -285,18 +304,7 @@ def extract_repository(candidate, github, clones, fixes, seed): with clones.open(candidate.url) as git: git.run("fetch", "--quiet", "--filter=blob:none", "origin", candidate.fix) changes, exclusions, proof = repository_changes(git, candidate, fixes, seed=seed) - pinned = changes[0]["head"] - license_record = github.get_license(candidate.repository, pinned) - spdx, license_class = source.license_info(license_record) - entry = { - **asdict(candidate), - "license_spdx": spdx, - "license_class": license_class, - "license_ref": pinned, - "license_path": license_record.get("path", ""), - "changes": changes, - "instrument": proof, - } + entry = {**asdict(candidate), "changes": changes, "instrument": proof} result = {"entry": entry, "ordinary_exclusions": exclusions} except source.Rejected as exc: result = {"rejected": str(exc), "ordinary_exclusions": getattr(exc, "counts", {})} @@ -305,9 +313,104 @@ def extract_repository(candidate, github, clones, fixes, seed): except (extract.InstrumentFailure, OSError) as exc: result = {"rejected": "extract_" + str(exc), "ordinary_exclusions": {}} result["counts"] = {key: value - before.get(key, 0) for key, value in clones.counts.items()} + return finish_license(result, github) if github is not None else result + + +def finish_license(result, github): + """REST stays on the coordinator, including the license at the extracted pin.""" + if "entry" not in result: + return result + try: + entry = result["entry"] + pinned = entry["changes"][0]["head"] + record = github.get_license(entry["repository"], pinned) + spdx, license_class = source.license_info(record) + entry.update(license_spdx=spdx, license_class=license_class, license_ref=pinned, license_path=record.get("path", "")) + except (source.Rejected, eligibility.RepositoryNotFound) as exc: + reason = "repository_or_commit_not_found" if isinstance(exc, eligibility.RepositoryNotFound) else str(exc) + return {"rejected": reason, "ordinary_exclusions": {}, "counts": result["counts"]} + except (extract.InstrumentFailure, OSError) as exc: + return {"rejected": "extract_" + str(exc), "ordinary_exclusions": {}, "counts": result["counts"]} + except ValueError as exc: + if str(exc) != "http_failure": + raise + return {"rejected": "extract_http_failure", "ordinary_exclusions": {}, "counts": result["counts"]} return result +@contextmanager +def ordered_extractions(rows, api, osv, used, clones, fixes, seed, checkpoint, entries, workers): + """Bounded speculation, ordered consumption, and coordinator-only durable writes. + + At most `workers` candidates are ahead of the acceptance cursor. Counters are + private to each job, including clone cleanup, and merged exactly once here. + Graceful STOP/guard/target exits drain started work into the journal for resume; + a hard interrupt may lose unjournaled work but cannot change accepted order. + """ + pending = deque() + pool = ThreadPoolExecutor(max_workers=workers) + + def collect(item): + row, decision, key, future, local = item + if future is not None: + future.exception() # Wait for cleanup before reading this job's counters. + for name, value in local.counts.items(): + if name == "peak_clone_bytes": + clones.counts[name] = max(clones.counts[name], value) + else: + clones.counts[name] += value + result = finish_license(future.result(), api) + checkpoint.put("extraction", key, result) + return row, decision, checkpoint.rows["extraction"].get(key) + + def iterate(): + ordered = iter(rows) + exhausted = False + while pending or not exhausted: + while not exhausted and len(pending) < min(workers, 300 - len(entries)): + checkpoint.boundary() + row = next(ordered, None) + if row is None: + exhausted = True + break + key = row["ghsa_id"] + if key not in checkpoint.rows["eligibility"]: + checkpoint.put("eligibility", key, decide(row, api, osv, used)) + decision = checkpoint.rows["eligibility"][key] + extraction_key, future, local = None, None, None + if "candidate" in decision: + candidate = source.Candidate(**decision["candidate"]) + extraction_key = digest({"repository": candidate.repository, "advisory": key, "fix": candidate.fix}) + if extraction_key not in checkpoint.rows["extraction"]: + checkpoint.boundary() + local = copy.copy(clones) + local.counts = Counter() + future = pool.submit(extract_repository, candidate, None, local, fixes.get(candidate.repository, set()), seed) + pending.append((row, decision, extraction_key, future, local)) + if not pending: + return + checkpoint.boundary() + yield collect(pending.popleft()) + + interrupted = False + stopped = None + try: + yield iterate() + except BaseException as exc: + interrupted = not isinstance(exc, Exception) + stopped = str(exc) if isinstance(exc, Stopped) else None + raise + finally: + pool.shutdown(wait=True, cancel_futures=True) + if not interrupted: + for item in pending: + future = item[3] + if future is not None and not future.cancelled(): + collect(item) + if stopped is not None: + checkpoint.progress("stopped", reason=stopped) + + def publish(journal, assigned, seed): """Idempotent publication from completed journal data, never inside the checkout.""" public, private = [], [] @@ -339,10 +442,15 @@ def run( journal_path=None, max_hours=None, clock=None, + extract_workers=None, ) -> dict: + workers = extract_workers if extract_workers is not None else getattr(clones, "extract_workers", 6) + if workers < 1: + raise ValueError("invalid_extract_workers") started = time.monotonic() report = {"status": "sourcing", "seed": seed, "candidates": {}, "rejections": {}, "ordinary_exclusions": {}} # Operational budget only: changing it must not invalidate journal resume inputs. + report["extract_workers"] = workers report["extract_deadline_seconds"] = getattr(clones, "deadline_seconds", None) counts, rejected, ordinary_counts = Counter(), Counter(), Counter() checkpoint = None @@ -365,41 +473,39 @@ def run( fixes_by_repo = advisory_index(rows, api) selected_networks, entries = set(), [] instrument_failures = 0 - for row in source.ordered_advisories(rows, seed): - checkpoint.boundary() - key = row["ghsa_id"] - counts[source.ecosystem(row)] += 1 - if key not in checkpoint.rows["eligibility"]: - decision = decide(row, api, osv, used, selected_networks) - checkpoint.put("eligibility", key, decision) - decision = checkpoint.rows["eligibility"][key] - if "rejected" in decision: - rejected[decision["rejected"]] += 1 - instrument_failures += int(decision["rejected"] == "decide_http_failure") + with ordered_extractions( + source.ordered_advisories(rows, seed), + api, + osv, + used, + clones, + fixes_by_repo, + seed, + checkpoint, + entries, + workers, + ) as outcomes: + for row, decision, result in outcomes: + counts[source.ecosystem(row)] += 1 + if decision.get("network") in selected_networks: + # Serial decide skipped this candidate entirely. Even a failed + # speculative extraction must not affect the systemic guard. + rejected["fork_network"] += 1 + elif "rejected" in decision: + rejected[decision["rejected"]] += 1 + instrument_failures += int(decision["rejected"] == "decide_http_failure") + else: + ordinary_counts.update(result["ordinary_exclusions"]) + if "rejected" in result: + rejected[result["rejected"]] += 1 + instrument_failures += int(result["rejected"].startswith("extract_")) + else: + selected_networks.add(decision["network"]) + entries.append(result["entry"]) if (instrument_failures >= 25 and not entries) or instrument_failures > 200: raise extract.InstrumentFailure("systemic_extraction_failure") - continue - candidate = source.Candidate(**decision["candidate"]) - # Multiple advisories for one repository can have different fix/range inputs. - extraction_key = digest({"repository": candidate.repository, "advisory": key, "fix": candidate.fix}) - if extraction_key not in checkpoint.rows["extraction"]: - checkpoint.boundary() - result = extract_repository(candidate, api, clones, fixes_by_repo.get(candidate.repository, set()), seed) - checkpoint.put("extraction", extraction_key, result) - result = checkpoint.rows["extraction"][extraction_key] - ordinary_counts.update(result["ordinary_exclusions"]) - if "rejected" in result: - rejected[result["rejected"]] += 1 - instrument_failures += int(result["rejected"].startswith("extract_")) - else: - selected_networks.add(decision["network"]) - entries.append(result["entry"]) - # Count per-candidate instrument failures (decide + extract, replay included), - # never ordinary eligibility exclusions. - if (instrument_failures >= 25 and not entries) or instrument_failures > 200: - raise extract.InstrumentFailure("systemic_extraction_failure") - if len(entries) == 300: - break + if len(entries) == 300: + break report.update( { "repositories": len(entries), @@ -463,18 +569,26 @@ def main(argv: list[str] | None = None) -> int: parser.add_argument("--cache", type=Path) parser.add_argument("--max-hours", type=float, help="Stop between units after this many hours per invocation") parser.add_argument("--extract-deadline", type=float, default=240.0, help="Extraction wall-clock budget per repository in seconds") + parser.add_argument("--extract-workers", type=int, default=6, help="Maximum concurrent repository extractions (default: 6)") parser.add_argument("--seed", type=int, default=20260601) parser.add_argument("--local-store") parser.add_argument("--s3-store", default="s3://") parser.add_argument("--pointer-manifest", type=Path, action="append", default=[]) parser.add_argument("--known-empty", choices=(*eligibility.SOURCES, "memory_local", "memory_s3"), action="append", default=[]) args = parser.parse_args(argv) + if args.extract_workers < 1: + parser.error("--extract-workers must be at least 1") try: config.load_dotenv(args.root / ".env") # Refuse an existing draft before doing any network work. if any((args.root / p).exists() for p in ("benchmarks/unseen/draft-p1.toml", "benchmarks/unseen/private/draft-p1.toml")): raise FileExistsError("draft_exists") - clones = Clones(args.root, args.cache or results_root() / "unseen-draft", deadline_seconds=args.extract_deadline) + clones = Clones( + args.root, + args.cache or results_root() / "unseen-draft", + deadline_seconds=args.extract_deadline, + extract_workers=args.extract_workers, + ) stores = { "local": memory.open_store(args.local_store or "file://" + str(results_root() / "memory"), read_only=True), "s3": memory.open_store(args.s3_store, read_only=True), diff --git a/benchmarks/unseen/extract.py b/benchmarks/unseen/extract.py index b87a781..2ad673f 100644 --- a/benchmarks/unseen/extract.py +++ b/benchmarks/unseen/extract.py @@ -53,12 +53,12 @@ class Git: raise ExtractionTimeout("extraction_timeout") return min(180, remaining) - def run(self, *args: str, allowed=(0,)) -> str: + def run(self, *args: str, allowed=(0,), input_data: bytes | None = None) -> str: timeout = self.command_timeout() try: done = self.runner( ["git", "-c", "core.quotePath=false", "-C", str(self.path), *args], - stdin=subprocess.DEVNULL, + **({"stdin": subprocess.DEVNULL} if input_data is None else {"input": input_data}), stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=timeout, @@ -75,6 +75,38 @@ class Git: self.last_blob_bytes = len(done.stdout) return done.stdout.decode("utf-8", errors="replace") + def materialize(self, revisions: list[str]) -> None: + """Fetch missing historical blobs in one pack, without changing content reads. + + ordinary_history inspects the complete first-parent history, including both + sides, attributes and non-source diffs. Blame -M/-C and log -L can walk all + ancestors and copy sources. Include the reachable trees, not only tip files, + so these commands keep exactly their original semantics. This is a safe + superset (merge-side trees included), still subject to the disk/deadline caps. + Missing-object enumeration itself does not trigger lazy blob fetches. + """ + data = ("\n".join(revisions) + "\n").encode() + objects = self.run("rev-list", "--objects", "--missing=print", "--stdin", input_data=data) + missing = sorted({line[1:] for line in objects.splitlines() if line.startswith("?")}) + if not all(SHA.fullmatch(oid) for oid in missing): + raise InstrumentFailure("invalid_missing_object") + if missing: + # stdin avoids argv limits; explicit object wants avoid one promisor + # request per show/blame. --no-filter overrides clone's stored filter. + self.run( + "fetch", + "--quiet", + "--no-tags", + "--no-write-fetch-head", + "--no-filter", + "--stdin", + "origin", + input_data=("\n".join(missing) + "\n").encode(), + ) + remaining = self.run("rev-list", "--objects", "--missing=print", "--stdin", input_data=data) + if any(line.startswith("?") for line in remaining.splitlines()): + raise InstrumentFailure("missing_blobs_after_fetch") + def disk_bytes(self) -> int: # git churns .git/objects/pack (incl. transient *.rev) during on-demand blob # fetches, so a path rglob just listed can vanish before stat(): skip it. diff --git a/tests/test_unseen_draft.py b/tests/test_unseen_draft.py index a07b03f..470c12a 100644 --- a/tests/test_unseen_draft.py +++ b/tests/test_unseen_draft.py @@ -242,14 +242,14 @@ def test_systemic_extraction_guard_boundaries(tmp_path, monkeypatch, successes, expected_failures = min(failures, 201 if successes else 25) assert report["status"] == status assert report["rejections"] == {"extract_git_exit_128": expected_failures} - assert clones.counts["clones"] == successes + expected_failures + assert successes + expected_failures <= clones.counts["clones"] <= successes + min(failures, expected_failures + 5) assert not (journal_path / "draft-p1.toml").exists() if status == "instrument_failure": assert report["error_type"] == "InstrumentFailure" - # Replayed failures must trip the same guard without attempting more clones. + # Replayed failures trip the same guard; only missing speculation may run. resumed = d.run(**kwargs) assert resumed["status"] == status and resumed["rejections"] == report["rejections"] - assert clones.counts["clones"] == successes + expected_failures + assert successes + expected_failures <= clones.counts["clones"] <= successes + min(failures, expected_failures + 5) @pytest.mark.parametrize( @@ -281,7 +281,7 @@ def test_non_instrument_extraction_bug_propagates_and_aborts(tmp_path): d.extract_repository(candidate, api, clones, set(), 7) report = d.run(tmp_path, github=api, osv=FakeOSV(), clones=clones, used=used, seed=7, known_empty=set(used.sources)) assert report["status"] == "instrument_failure" and report["error_type"] == "ValueError" - assert report["rejections"] == {} and clones.counts["clones"] == 2 + assert report["rejections"] == {} and 2 <= clones.counts["clones"] <= 3 def test_ordinary_rejections_do_not_trip_extraction_guard(tmp_path, monkeypatch): @@ -345,11 +345,13 @@ def test_systemic_zero_source_is_instrument_failure_with_recorded_rejections(tmp report = d.run(tmp_path, github=FakeAPI(30), osv=FakeOSV(), clones=clones, used=used, seed=1, known_empty=set(used.sources)) assert report["status"] == "instrument_failure" assert report["rejections"] == {"extract_zero_source_bytes": 25} - assert clones.counts["clones"] == 25 + assert 25 <= clones.counts["clones"] <= 30 -@pytest.mark.parametrize("deadline_args,deadline", [([], 240.0), (["--extract-deadline", "12.5"], 12.5)]) -def test_cli_uses_environment_and_injected_stores(tmp_path, monkeypatch, capsys, deadline_args, deadline): +@pytest.mark.parametrize( + "deadline_args,deadline,workers", [([], 240.0, 6), (["--extract-deadline", "12.5", "--extract-workers", "2"], 12.5, 2)] +) +def test_cli_uses_environment_and_injected_stores(tmp_path, monkeypatch, capsys, deadline_args, deadline, workers): import json from openultrasast import config @@ -368,6 +370,7 @@ def test_cli_uses_environment_and_injected_stores(tmp_path, monkeypatch, capsys, assert d.main(args + deadline_args) == 1 output = capsys.readouterr().out assert json.loads(output)["extract_deadline_seconds"] == deadline + assert json.loads(output)["extract_workers"] == workers assert "instrument_failure" in output and "fake-secret" not in output and "fixture/" not in output assert seen[-1] == "fake-secret" assert all(kwargs == {"read_only": True} for _, kwargs in seen[:-1]) @@ -390,7 +393,16 @@ def test_timeout_skips_repository_and_draft_continues(tmp_path, monkeypatch, tim monkeypatch.setattr(d, "repository_changes", lambda *a, **kw: (entries()[0]["changes"], {}, {"bytes": 1, "files": 1})) clones = d.Clones(tmp_path / "repo", tmp_path / "cache", runner=runner, deadline_seconds=240.0) used = e.UsedSet() - report = d.run(tmp_path / "repo", github=FakeAPI(2), osv=FakeOSV(), clones=clones, used=used, seed=1, known_empty=set(used.sources)) + report = d.run( + tmp_path / "repo", + github=FakeAPI(2), + osv=FakeOSV(), + clones=clones, + used=used, + seed=1, + known_empty=set(used.sources), + extract_workers=1, + ) assert report["status"] == "insufficient_repositories" assert report["rejections"] == {"extraction_timeout": 1} assert report["repositories"] == 1 and clone_count == 2 @@ -450,6 +462,9 @@ def test_repository_changes_excludes_all_advisory_sites(monkeypatch): return "app.py\n" return args[-1] if args[0] == "rev-parse" else "" + def materialize(self, revisions): + assert revisions == ["HEAD", "a" * 40, "d" * 40] + def blob(self, revision, path): return "def handler():\n return 1\n" @@ -483,7 +498,8 @@ def test_repository_changes_excludes_all_advisory_sites(monkeypatch): monkeypatch.setattr(d.extract, "draw_ordinary", draw) _, _, proof = d.repository_changes(FakeGit(), candidate, {"d" * 40}, seed=1) assert proof["bytes"] > 0 - assert len([c for c in calls if c[0] == "fetch"]) == 2 + assert len([c for c in calls if c[0] == "fetch"]) == 1 + assert "--no-tags" in next(c for c in calls if c[0] == "fetch") def test_candidate_404_is_counted_but_service_failure_aborts(tmp_path): @@ -911,3 +927,197 @@ def test_decide_http_failure_systemic_guard_aborts(tmp_path): report = d.run(tmp_path, github=api, osv=FakeOSV(), clones=FakeClones(), used=used, seed=7, known_empty=set(used.sources)) assert report["status"] == "instrument_failure" and report["error_type"] == "InstrumentFailure" assert report["rejections"].get("decide_http_failure") == 25 + + +def test_parallel_pool_is_byte_identical_and_fork_selection_is_seed_ordered(tmp_path, monkeypatch): + import threading + + journal_module = importlib.import_module("benchmarks.unseen.journal") + monkeypatch.setattr(journal_module.os, "fsync", lambda fd: None) + coordinator = threading.get_ident() + ordered = s.ordered_advisories(list(FakeAPI(310).advisories()), 7) + names = [row["references"][0].split("/commit/")[0].removeprefix("https://github.com/") for row in ordered] + ranks = {name: i for i, name in enumerate(names)} + networks = {names[1]: names[0], names[3]: names[2], names[5]: names[4]} + extracted, selected, snapshots = [], [], [] + worker_count = [1] + second_done = threading.Event() + active, peak = [0], [0] + lock = threading.Lock() + changes = entries()[0]["changes"] + + class API(FakeAPI): + def get_repo(self, name): + assert threading.get_ident() == coordinator + result = super().get_repo(name) + if name in networks: + result["source"] = {"full_name": networks[name]} + return result + + def get_commit(self, name, sha): + assert threading.get_ident() == coordinator + if ranks.get(name) in {5, 8}: + raise ValueError("http_failure") + return super().get_commit(name, sha) + + def get_license(self, name, ref): + assert threading.get_ident() == coordinator + return super().get_license(name, ref) + + def extract_changes(git, candidate, fixes, *, seed): + assert threading.get_ident() != coordinator + rank = ranks[candidate.repository] + with lock: + active[0] += 1 + peak[0] = max(peak[0], active[0]) + try: + if rank == 0 and worker_count[0] > 1: + assert second_done.wait(10), "extraction did not overlap" + if rank == 1: + second_done.set() + with lock: + extracted.append(rank) + if rank in {0, 7}: + raise d.InstrumentFailure("git_exit_128") + if rank == 6: + raise d.extract.ExtractionTimeout("extraction_timeout") + return copy.deepcopy(changes), {"eligible": 40}, {"bytes": 12, "files": 1} + finally: + with lock: + active[0] -= 1 + + original_assign, original_put = d.assign_slices, d.Journal.put + + def assign(rows, *, seed): + selected.append([row["repository"] for row in rows]) + return original_assign(rows, seed=seed) + + def put(self, *args): + assert threading.get_ident() == coordinator + return original_put(self, *args) + + monkeypatch.setattr(d, "repository_changes", extract_changes) + monkeypatch.setattr(d, "assign_slices", assign) + monkeypatch.setattr(d.Journal, "put", put) + for workers in (1, 6): + worker_count[0] = workers + second_done.clear() + extracted.clear() + peak[0] = 0 + path = tmp_path / str(workers) + used = e.UsedSet() + report = d.run( + tmp_path / "repo", + github=API(310), + osv=FakeOSV(), + clones=FakeClones(), + used=used, + seed=7, + known_empty=set(used.sources), + journal_path=path, + extract_workers=workers, + ) + assert report["status"] == "draft", report + assert report["rejections"] == {"fork_network": 2, "extract_git_exit_128": 2, "extraction_timeout": 1, "decide_http_failure": 1} + assert 1 <= peak[0] <= workers + if workers == 6: + assert peak[0] > 1 and extracted.index(1) < extracted.index(0) + snapshots.append( + (report["manifest_sha256"], *(path.joinpath(name).read_bytes() for name in ("draft-p1.toml", "private-draft-p1.toml"))) + ) + # Independent serial oracle: first successful repository wins each network. + expected = [name for i, name in enumerate(names) if i not in {0, 3, 5, 6, 7, 8}][:300] + assert selected == [expected, expected] + assert snapshots[0] == snapshots[1] + + +@pytest.mark.parametrize("workers", [0, -1]) +def test_invalid_extract_workers_rejected_before_io(tmp_path, workers): + used, api = e.UsedSet(), FakeAPI() + with pytest.raises(ValueError, match="invalid_extract_workers"): + d.run(tmp_path, github=api, osv=FakeOSV(), clones=FakeClones(), used=used, seed=1, extract_workers=workers) + assert api.calls == 0 + + +def test_graceful_stop_drains_workers_and_resume_with_different_width_reuses_results(tmp_path, monkeypatch): + import json + + journal_module = importlib.import_module("benchmarks.unseen.journal") + monkeypatch.setattr(journal_module.os, "fsync", lambda fd: None) + changes = entries()[0]["changes"] + monkeypatch.setattr(d, "repository_changes", lambda *a, **kw: (changes, {}, {"bytes": 12, "files": 1})) + path, full = tmp_path / "resumed", tmp_path / "full" + used = e.UsedSet() + + def run(destination, workers, clones): + return d.run( + tmp_path / "repo", + github=FakeAPI(), + osv=FakeOSV(), + clones=clones, + used=used, + seed=7, + known_empty=set(used.sources), + journal_path=destination, + extract_workers=workers, + ) + + assert run(full, 1, FakeClones())["status"] == "draft" + original = d.Journal.put + + def stop(self, stage, key, value): + original(self, stage, key, value) + if stage == "extraction" and len(self.rows[stage]) == 23: + (self.path / "STOP").touch() + + monkeypatch.setattr(d.Journal, "put", stop) + first = FakeClones() + assert run(path, 6, first)["status"] == "stopped" + progress = json.loads((path / "progress.json").read_text()) + completed = progress["counts"]["extraction"] + assert progress["status"] == "stopped" + assert 23 <= completed <= 28 + assert first.counts["clones"] == completed + (path / "STOP").unlink() + monkeypatch.setattr(d.Journal, "put", original) + rest = FakeClones() + assert run(path, 1, rest)["status"] == "draft" + assert rest.counts["clones"] == 300 - completed + for name in ("draft-p1.toml", "private-draft-p1.toml"): + assert (path / name).read_bytes() == (full / name).read_bytes() + + +@pytest.mark.parametrize( + "failure,reason", + [ + (ValueError("http_failure"), "extract_http_failure"), + (OSError("read_failed"), "extract_read_failed"), + (d.InstrumentFailure("read_failed"), "extract_read_failed"), + (e.RepositoryNotFound("not_found"), "repository_or_commit_not_found"), + ], +) +def test_parallel_license_failure_is_a_resilient_skip(tmp_path, monkeypatch, failure, reason): + changes = entries()[0]["changes"] + changes[0]["head"] = "c" * 40 + monkeypatch.setattr(d, "repository_changes", lambda *a, **kw: (changes, {}, {"bytes": 12, "files": 1})) + + class API(FakeAPI): + def get_license(self, name, ref): + if name == "fixture/project000" and ref == "c" * 40: + raise failure + return super().get_license(name, ref) + + used = e.UsedSet() + report = d.run( + tmp_path / "repo", + github=API(2), + osv=FakeOSV(), + clones=FakeClones(), + used=used, + seed=7, + known_empty=set(used.sources), + journal_path=tmp_path / "journal", + extract_workers=6, + ) + assert report["status"] == "insufficient_repositories" + assert report["repositories"] == 1 and report["rejections"] == {reason: 1} diff --git a/tests/test_unseen_extract.py b/tests/test_unseen_extract.py index 314fd17..1759a8f 100644 --- a/tests/test_unseen_extract.py +++ b/tests/test_unseen_extract.py @@ -372,3 +372,75 @@ def test_disk_bytes_skips_files_that_vanish_mid_walk(tmp_path): git.path = type("P", (), {"rglob": lambda self, pattern: iter([*real, Vanished()])})() assert git.disk_bytes() == 30 + + +def test_partial_clone_batch_matches_lazy_bytes_without_per_file_fetches(history, tmp_path): + """Real promisor packs over file:// only: count lazy fetches via Git's trace.""" + import json + import os + + git, commit, candidate, _, _, _, _ = history + for revision in range(12): + commit( + f"def handler(value):\n return int(value) + {revision}\n", + extra={f"file{revision * 5 + i}.py": f"x = {revision * 5 + i}\n" for i in range(5)}, + ) + git.run("config", "uploadpack.allowFilter", "true") + git.run("config", "uploadpack.allowAnySHA1InWant", "true") + baseline = [e.introducing(git, candidate), e.ordinary_history(git, "HEAD")] + results, fetches = [], [] + for batch in (False, True): + path = tmp_path / ("batch" if batch else "lazy") + trace = tmp_path / (path.name + ".trace") + # Local file transport only, no network. No checkout: all blobs start missing. + subprocess.run( + ["git", "clone", "--quiet", "--filter=blob:none", "--no-checkout", tmp_path.as_uri(), str(path)], + check=True, + capture_output=True, + ) + commands = [] + + def runner(argv, commands=commands, trace=trace, **kwargs): + commands.append(argv) + return subprocess.run(argv, **kwargs, env={**os.environ, "GIT_TRACE": str(trace), "GIT_ALLOW_PROTOCOL": "file"}) + + partial = e.Git(path, runner=runner) + missing = partial.run("rev-list", "--objects", "--missing=print", "HEAD") + assert sum(line.startswith("?") for line in missing.splitlines()) >= 60 + if batch: + partial.materialize(["HEAD", candidate.fix]) + results.append([e.introducing(partial, candidate), e.ordinary_history(partial, "HEAD")]) + assert partial.files_read > 0 and partial.bytes_read > 0 + fetches.append(sum("built-in: git fetch " in line for line in trace.read_text().splitlines())) + if batch: + assert sum("fetch" in argv for argv in commands) == 1 + partial.materialize(["HEAD", candidate.fix]) # Already hydrated: no second fetch. + assert sum("fetch" in argv for argv in commands) == 1 + assert json.dumps(results[0], sort_keys=True).encode() == json.dumps(results[1], sort_keys=True).encode() + assert results[1] == baseline + assert fetches[0] >= 12 and fetches[1] == 1 + + +def test_materialize_timeout_uses_repository_deadline_and_checks_completeness(tmp_path): + oid = "a" * 40 + calls = [] + now = [0.0] + + def runner(argv, **kwargs): + calls.append((argv, kwargs)) + if "fetch" in argv: + assert kwargs["timeout"] == 3 + assert kwargs["input"] == (oid + "\n").encode() + raise subprocess.TimeoutExpired(argv, kwargs["timeout"]) + now[0] += 7 + return subprocess.CompletedProcess(argv, 0, ("?" + oid + "\n").encode(), b"") + + with pytest.raises(e.ExtractionTimeout, match="extraction_timeout"): + e.Git(tmp_path, runner=runner, deadline_seconds=10, _now=lambda: now[0]).materialize(["HEAD"]) + assert len(calls) == 2 + + def incomplete(argv, **kwargs): + return subprocess.CompletedProcess(argv, 0, ("?" + oid + "\n").encode(), b"") + + with pytest.raises(e.InstrumentFailure, match="missing_blobs_after_fetch"): + e.Git(tmp_path, runner=incomplete).materialize(["HEAD"])